Mostly done pipeline arch. Now onto cleanup and building.

7 files changed, 172 insertions, 70 deletions

diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index 2ee43fb7..98ebdc71 100644
--- a/src/audio/include/audio_decoder.hpp
+++ b/src/audio/include/audio_decoder.hpp
@@ -1,27 +1,34 @@
 #pragma once
 
 #include <cstddef>
-#include "audio_element.hpp"
+
 #include "ff.h"
+
+#include "audio_element.hpp"
 #include "codec.hpp"
 
 namespace audio {
 
+/*
+ * An audio element that accepts various kinds of encoded audio streams as
+ * input, and converts them to uncompressed PCM output.
+ */
 class AudioDecoder : public IAudioElement {
  public:
   AudioDecoder();
   ~AudioDecoder();
 
-  auto SetInputBuffer(StreamBufferHandle_t) -> void;
-  auto SetOutputBuffer(StreamBufferHandle_t) -> void;
+  auto SetInputBuffer(StreamBufferHandle_t*) -> void;
+  auto SetOutputBuffer(StreamBufferHandle_t*) -> void;
+
+  AudioDecoder(const AudioDecoder&) = delete;
+  AudioDecoder& operator=(const AudioDecoder&) = delete;
 
  private:
   std::unique_ptr<codecs::ICodec> current_codec_;
+  std::optional<StreamInfo> stream_info_;
 
-  uint8_t *working_buffer_;
-
-  StreamBufferHandle_t input_buffer_;
-  StreamBufferHandle_t output_buffer_;
+  uint8_t* chunk_buffer_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_element.hpp b/src/audio/include/audio_element.hpp
index 0be58f48..2a2f0727 100644
--- a/src/audio/include/audio_element.hpp
+++ b/src/audio/include/audio_element.hpp
@@ -3,30 +3,83 @@
 #include <stdint.h>
 #include <cstdint>
 #include "freertos/portmacro.h"
-#include "types.hpp"
 #include "result.hpp"
+#include "types.hpp"
 
 namespace audio {
 
-extern const std::size_t kMaxFrameSize;
-
+/*
+ * One indepedentent part of an audio pipeline. Each element has an input and
+ * output message stream, and is responsible for taking data from the input
+ * stream, applying some kind of transformation to it, then sending the result
+ * out via the output stream. All communication with an element should be done
+ * over these streams, as an element's methods are only safe to call from the
+ * task that owns it.
+ *
+ * Each element implementation will have its input stream automatically parsed
+ * by its owning task, and its various Process* methods will be invoked
+ * accordingly. Element implementations are responsible for managing their own
+ * writes to their output streams.
+ */
 class IAudioElement {
  public:
+  IAudioElement() : input_buffer_(nullptr), output_buffer_(nullptr) {}
   virtual ~IAudioElement();
 
+  /*
+   * Returns the stack size in bytes that this element requires. This should
+   * be tuned according to the observed stack size of each element, as different
+   * elements have fairly different stack requirements.
+   */
+  virtual auto StackSizeBytes() -> std::size_t { return 2048; };
+
+  /*
+   * How long to wait for new data on the input stream before triggering a call
+   * to ProcessIdle(). If this is portMAX_DELAY (the default), then ProcessIdle
+   * will never be called.
+   */
   virtual auto IdleTimeout() -> TickType_t { return portMAX_DELAY; }
 
-  virtual auto InputBuffer() -> MessageBufferHandle_t* = 0;
+  /* Returns this element's input buffer. */
+  auto InputBuffer() -> MessageBufferHandle_t* { return input_buffer_; }
 
-  virtual auto OutputBuffer() -> MessageBufferHandle_t* = 0;
+  /* Returns this element's output buffer. */
+  auto OutputBuffer() -> MessageBufferHandle_t* { return output_buffer_; }
 
+  /* Errors that may be returned by any of the Process* methods. */
   enum StreamError {
-    BAD_FORMAT
+    // Indicates that this element is unable to handle the upcoming chunks.
+    UNSUPPORTED_STREAM,
+    // Indicates an error with reading or writing stream data.
+    IO_ERROR,
   };
 
-  virtual auto ProcessStreamInfo(StreamInfo &info) -> cpp::result<void, StreamError> = 0;
-  virtual auto ProcessChunk(uint8_t* data, std::size_t length) -> cpp::result<void, StreamError> = 0;
+  /*
+   * Called when a StreamInfo message is received. Used to configure this
+   * element in preperation for incoming chunks.
+   */
+  virtual auto ProcessStreamInfo(StreamInfo&& info)
+      -> cpp::result<void, StreamError> = 0;
+
+  /*
+   * Called when a ChunkHeader message is received. Includes the data associated
+   * with this chunk of stream data. This method should return the number of
+   * bytes in this chunk that were actually used; leftover bytes will be
+   * prepended to the next call.
+   */
+  virtual auto ProcessChunk(uint8_t* data, std::size_t length)
+      -> cpp::result<size_t, StreamError> = 0;
+
+  /*
+   * Called when there has been no data received over the input buffer for some
+   * time. This could be used to synthesize output, or to save memory by
+   * releasing unused resources.
+   */
   virtual auto ProcessIdle() -> cpp::result<void, StreamError> = 0;
+
+ protected:
+  StreamBufferHandle_t* input_buffer_;
+  StreamBufferHandle_t* output_buffer_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_task.hpp b/src/audio/include/audio_task.hpp
index 79604f33..05888170 100644
--- a/src/audio/include/audio_task.hpp
+++ b/src/audio/include/audio_task.hpp
@@ -10,6 +10,8 @@ struct AudioTaskArgs {
   std::shared_ptr<IAudioElement>& element;
 };
 
-void audio_task(void* args);
+auto StartAudioTask(std::shared_ptr<IAudioElement>& element) -> void;
+
+void AudioTaskMain(void* args);
 
 }  // namespace audio
diff --git a/src/audio/include/chunk.hpp b/src/audio/include/chunk.hpp
index 1351ecfb..a3f943ea 100644
--- a/src/audio/include/chunk.hpp
+++ b/src/audio/include/chunk.hpp
@@ -10,6 +10,8 @@
 
 namespace audio {
 
+extern const std::size_t kMaxChunkSize;
+
 enum ChunkWriteResult {
   // Returned when the callback does not write any data.
   CHUNK_OUT_OF_DATA,
@@ -29,7 +31,20 @@ enum ChunkWriteResult {
  * number of bytes it wrote. Return a value of 0 to indicate that there is no
  * more input to read.
  */
-auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer, size_t working_buffer_length, std::function<size_t(uint8_t*,size_t)> callback, TickType_t max_wait) -> EncodeWriteResult;
+auto WriteChunksToStream(MessageBufferHandle_t* stream,
+                         uint8_t* working_buffer,
+                         size_t working_buffer_length,
+                         std::function<size_t(uint8_t*, size_t)> callback,
+                         TickType_t max_wait) -> EncodeWriteResult;
+
+class ChunkReader {
+ public:
+  ChunkReader(MessageBufferHandle_t* stream);
+  ~ChunkReader();
+
+  auto Reset() -> void;
+
+  auto GetLastMessage() -> std::pair<uint8_t*, size_t>;
 
   enum ChunkReadResult {
     // Returned an error in parsing the cbor-encoded header.
@@ -38,20 +53,32 @@ auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer,
     CHUNK_READ_TIMEOUT,
     // Returned when a non-chunk message is received.
     CHUNK_STREAM_ENDED,
+    // Returned when the processing callback does not return a value.
+    CHUNK_PROCESSING_ERROR,
   };
 
-/*
- * Reads chunks of data from the given input stream, and invokes the given
- * callback to process each of them in turn.
- * 
- * The callback will be invoked with a byte buffer and its size. The callback
- * should process as much data as it can from this buffer, and then return the
- * number of bytes it was able to read. Any leftover bytes will be added as a
- * prefix to the next chunk.
- *
- * If this function encounters a message in the stream that is not a chunk, it
- * will place the message at the start of the working_buffer and then return.
- */
-auto ReadChunksFromStream(MessageBufferHandle_t *stream, uint8_t *working_buffer, size_t working_buffer_length, std::function<size_t(uint8_t*,size_t)> callback, TickType_t max_wait) -> EncodeReadResult;
+  /*
+   * Reads chunks of data from the given input stream, and invokes the given
+   * callback to process each of them in turn.
+   *
+   * The callback will be invoked with a byte buffer and its size. The callback
+   * should process as much data as it can from this buffer, and then return the
+   * number of bytes it was able to read. Any leftover bytes will be added as a
+   * prefix to the next chunk.
+   *
+   * If this function encounters a message in the stream that is not a chunk, it
+   * will place the message at the start of the working_buffer and then return.
+   */
+  auto ReadChunkFromStream(
+      std::function<std::optional<size_t>(uint8_t*, size_t)> callback,
+      TickType_t max_wait) -> EncodeReadResult;
+
+ private:
+  MessageBufferHandle_t* stream_;
+  uint8_t* working_buffer_;
+
+  std::size_t leftover_bytes_ = 0;
+  std::size_t last_message_size_ = 0;
+};
 
-} // namespace audio
+}  // namespace audio
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index 5651419d..0fc2729f 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/src/audio/include/fatfs_audio_input.hpp
@@ -20,10 +20,17 @@ class FatfsAudioInput : public IAudioElement {
 
   auto OutputBuffer() -> MessageBufferHandle_t;
 
+  auto SendChunk(uint8_t* buffer, size_t size) -> size_t;
+
  private:
   std::shared_ptr<drivers::SdStorage> storage_;
 
-  uint8_t *working_buffer_;
+  uint8_t* file_buffer_;
+  uint8_t* file_buffer_read_pos_;
+  uint8_t* pending_read_pos_;
+  uint8_t* file_buffer_write_pos_;
+
+  uint8_t* chunk_buffer_;
 
   FIL current_file_;
   bool is_file_open_ = false;
diff --git a/src/audio/include/stream_info.hpp b/src/audio/include/stream_info.hpp
index 2b1429ea..bf5d4c60 100644
--- a/src/audio/include/stream_info.hpp
+++ b/src/audio/include/stream_info.hpp
@@ -9,36 +9,42 @@
 namespace audio {
 
 class StreamInfo {
-  public:
-    enum ParseError {
-      WRONG_TYPE,
-      MISSING_MAP,
-    };
-
-    static auto Create(const uint8_t *buffer, size_t length) -> cpp::result<StreamInfo, ParseError>;
-    StreamInfo(CborValue& map);
-
-    StreamInfo() = default;
-    StreamInfo(const StreamInfo&) = default;
-
-    ~StreamInfo() = default;
-
-    auto Path() const -> const std::optional<std::string>& { return path_; }
-    auto Channels() const -> const std::optional<uint8_t>& { return channels_; }
-    auto BitsPerSample() const -> const std::optional<uint8_t>& { return bits_per_sample_; }
-    auto SampleRate() const -> const std::optional<uint16_t>& { return sample_rate_; }
-
-    enum EncodeError {
-      OUT_OF_MEMORY,
-    };
-
-    auto WriteToStream(CborEncoder encoder) -> cpp::result<void, EncodeError>;
-  private:
-
-    std::optional<std::string> path_;
-    std::optional<uint8_t> channels_;
-    std::optional<uint8_t> bits_per_sample_;
-    std::optional<uint16_t> sample_rate_;
+ public:
+  enum ParseError {
+    WRONG_TYPE,
+    MISSING_MAP,
+    CBOR_ERROR,
+  };
+
+  static auto Create(const uint8_t* buffer, size_t length)
+      -> cpp::result<StreamInfo, ParseError>;
+  StreamInfo(CborValue& map);
+
+  StreamInfo() = default;
+  StreamInfo(const StreamInfo&) = default;
+
+  ~StreamInfo() = default;
+
+  auto Path() const -> const std::optional<std::string>& { return path_; }
+  auto Channels() const -> const std::optional<uint8_t>& { return channels_; }
+  auto BitsPerSample() const -> const std::optional<uint8_t>& {
+    return bits_per_sample_;
+  }
+  auto SampleRate() const -> const std::optional<uint16_t>& {
+    return sample_rate_;
+  }
+
+  enum EncodeError {
+    OUT_OF_MEMORY,
+  };
+
+  auto WriteToMap(CborEncoder encoder) -> cpp::result<size_t, EncodeError>;
+
+ private:
+  std::optional<std::string> path_;
+  std::optional<uint8_t> channels_;
+  std::optional<uint8_t> bits_per_sample_;
+  std::optional<uint16_t> sample_rate_;
 };
 
-} // namespace audio
+}  // namespace audio
diff --git a/src/audio/include/stream_message.hpp b/src/audio/include/stream_message.hpp
index f59aba8d..2791dcd8 100644
--- a/src/audio/include/stream_message.hpp
+++ b/src/audio/include/stream_message.hpp
@@ -1,11 +1,11 @@
 #pragma once
 
 namespace audio {
-  
-  enum MessageType {
-    TYPE_UNKNOWN,
-    TYPE_CHUNK_HEADER,
-    TYPE_STREAM_INFO,
-  };
 
-} // namespace audio
+enum MessageType {
+  TYPE_UNKNOWN,
+  TYPE_CHUNK_HEADER,
+  TYPE_STREAM_INFO,
+};
+
+}  // namespace audio