6 files changed, 128 insertions, 91 deletions

diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index 083bd564..2ee43fb7 100644
--- a/src/audio/include/audio_decoder.hpp
+++ b/src/audio/include/audio_decoder.hpp
@@ -12,13 +12,6 @@ class AudioDecoder : public IAudioElement {
   AudioDecoder();
   ~AudioDecoder();
 
-  auto Pause() -> void;
-  auto IsPaused() -> bool;
-
-  auto Resume() -> void;
-
-  auto SetInputCommandQueue(QueueHandle_t) -> void;
-  auto SetOutputCommandQueue(QueueHandle_t) -> void;
   auto SetInputBuffer(StreamBufferHandle_t) -> void;
   auto SetOutputBuffer(StreamBufferHandle_t) -> void;
 
@@ -27,8 +20,6 @@ class AudioDecoder : public IAudioElement {
 
   uint8_t *working_buffer_;
 
-  QueueHandle_t input_queue_;
-  QueueHandle_t output_queue_;
   StreamBufferHandle_t input_buffer_;
   StreamBufferHandle_t output_buffer_;
 };
diff --git a/src/audio/include/audio_element.hpp b/src/audio/include/audio_element.hpp
index 03fefd70..0be58f48 100644
--- a/src/audio/include/audio_element.hpp
+++ b/src/audio/include/audio_element.hpp
@@ -1,6 +1,10 @@
 #pragma once
 
+#include <stdint.h>
 #include <cstdint>
+#include "freertos/portmacro.h"
+#include "types.hpp"
+#include "result.hpp"
 
 namespace audio {
 
@@ -10,71 +14,19 @@ class IAudioElement {
  public:
   virtual ~IAudioElement();
 
-  enum CommandType {
-    /*
-     * Sets the sequence number of the most recent byte stream. Any commands
-     * received that have a lower sequence number than this will be discarded.
-     */
-    SEQUENCE_NUMBER,
-    /*
-     * Instructs this element to read a specific number of bytes from its
-     * input buffer.
-     */
-    READ_FRAME,
-    /*
-     * Represents an element-specific command. This handling of this is
-     * delegated to element implementations.
-     */
-    ELEMENT,
-    /* Instructs this element to shut down. */
-    QUIT,
-  };
-
-  struct Command {
-    CommandType type;
-    uint8_t sequence_number;
-    // TODO: tag data's type
-    union {
-      void* data;
-      std::size_t frame_size;
-    };
-  };
+  virtual auto IdleTimeout() -> TickType_t { return portMAX_DELAY; }
 
-  /*
-   * Returns a queue that should be used for all communication with this
-   * element.
-   */
-  virtual auto InputCommandQueue() -> QueueHandle_t = 0;
+  virtual auto InputBuffer() -> MessageBufferHandle_t* = 0;
 
-  /*
-   * Returns a buffer that will be used to stream input bytes to this element.
-   * This may be NULL, if this element represents a source, e.g. a FATFS
-   * reader.
-   */
-  virtual auto InputBuffer() -> StreamBufferHandle_t = 0;
+  virtual auto OutputBuffer() -> MessageBufferHandle_t* = 0;
 
-  enum ProcessResult {
-    OK,
-    OUTPUT_FULL,
-    ERROR,
+  enum StreamError {
+    BAD_FORMAT
   };
 
-  /*
-   * Called when an element-specific command has been received.
-   */
-  virtual auto ProcessElementCommand(void* command) -> ProcessResult = 0;
-
-  virtual auto SkipElementCommand(void* command) -> void = 0;
-
-  /*
-   * Called with the result of a read bytes command.
-   */
-  virtual auto ProcessData(uint8_t* data, uint16_t length) -> ProcessResult = 0;
-
-  /*
-   * Called periodically when there are no pending commands.
-   */
-  virtual auto ProcessIdle() -> ProcessResult = 0;
+  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;
+  virtual auto ProcessIdle() -> cpp::result<void, StreamError> = 0;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/chunk.hpp b/src/audio/include/chunk.hpp
new file mode 100644
index 00000000..1351ecfb
--- /dev/null
+++ b/src/audio/include/chunk.hpp
@@ -0,0 +1,57 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include "esp-idf/components/cbor/tinycbor/src/cbor.h"
+#include "freertos/portmacro.h"
+#include "result.hpp"
+
+namespace audio {
+
+enum ChunkWriteResult {
+  // Returned when the callback does not write any data.
+  CHUNK_OUT_OF_DATA,
+  // Returned when there is an error encoding a chunk header using cbor.
+  CHUNK_ENCODING_ERROR,
+  // Returned when max_wait expires without room in the stream buffer becoming
+  // available.
+  CHUNK_WRITE_TIMEOUT,
+};
+
+/*
+ * Invokes the given callback to receive data, breaks the received data up into
+ * chunks with headers, and writes those chunks to the given output stream.
+ *
+ * The callback will be invoked with a byte buffer and its size. The callback
+ * should write as much data as it can to this buffer, and then return the
+ * 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;
+
+  enum ChunkReadResult {
+    // Returned an error in parsing the cbor-encoded header.
+    CHUNK_DECODING_ERROR,
+    // Returned when max_wait expired before any data was read.
+    CHUNK_READ_TIMEOUT,
+    // Returned when a non-chunk message is received.
+    CHUNK_STREAM_ENDED,
+  };
+
+/*
+ * 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;
+
+} // namespace audio
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index bf5f150d..5651419d 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/src/audio/include/fatfs_audio_input.hpp
@@ -15,42 +15,24 @@ namespace audio {
 
 class FatfsAudioInput : public IAudioElement {
  public:
-  struct InputCommand {
-    std::string filename;
-    size_t seek_to;
-    bool interrupt;
-  };
-
-  struct OutputCommand {
-    std::string extension;
-  };
-
   FatfsAudioInput(std::shared_ptr<drivers::SdStorage> storage);
   ~FatfsAudioInput();
 
-  auto OutputCommandQueue() -> QueueHandle_t;
-  auto OutputBuffer() -> StreamBufferHandle_t;
+  auto OutputBuffer() -> MessageBufferHandle_t;
 
  private:
   std::shared_ptr<drivers::SdStorage> storage_;
 
   uint8_t *working_buffer_;
 
-  uint8_t current_sequence_ = 0;
   FIL current_file_;
   bool is_file_open_ = false;
 
-  uint8_t* input_queue_memory_;
-  StaticQueue_t input_queue_metadata_;
-  QueueHandle_t input_queue_;
-
-  uint8_t* output_queue_memory_;
-  StaticQueue_t output_queue_metadata_;
-  QueueHandle_t output_queue_;
+  MessageBufferHandle_t input_buffer_;
 
   uint8_t* output_buffer_memory_;
-  StaticStreamBuffer_t output_buffer_metadata_;
-  StreamBufferHandle_t output_buffer_;
+  StaticMessageBuffer_t output_buffer_metadata_;
+  MessageBufferHandle_t output_buffer_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/stream_info.hpp b/src/audio/include/stream_info.hpp
new file mode 100644
index 00000000..2b1429ea
--- /dev/null
+++ b/src/audio/include/stream_info.hpp
@@ -0,0 +1,44 @@
+#pragma once
+
+#include <cstdint>
+#include <optional>
+#include <string>
+#include "esp-idf/components/cbor/tinycbor/src/cbor.h"
+#include "result.hpp"
+
+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_;
+};
+
+} // namespace audio
diff --git a/src/audio/include/stream_message.hpp b/src/audio/include/stream_message.hpp
new file mode 100644
index 00000000..f59aba8d
--- /dev/null
+++ b/src/audio/include/stream_message.hpp
@@ -0,0 +1,11 @@
+#pragma once
+
+namespace audio {
+  
+  enum MessageType {
+    TYPE_UNKNOWN,
+    TYPE_CHUNK_HEADER,
+    TYPE_STREAM_INFO,
+  };
+
+} // namespace audio