8 files changed, 132 insertions, 113 deletions

diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index 0a2df76d..a2591d25 100644
--- a/src/audio/include/audio_decoder.hpp
+++ b/src/audio/include/audio_decoder.hpp
@@ -4,6 +4,7 @@
 #include <cstdint>
 #include <memory>
 
+#include "chunk.hpp"
 #include "ff.h"
 #include "span.hpp"
 
@@ -30,11 +31,13 @@ class AudioDecoder : public IAudioElement {
     return 1024;
   }
 
+  auto HasUnprocessedInput() -> bool override;
+
   auto ProcessStreamInfo(const StreamInfo& info)
       -> cpp::result<void, AudioProcessingError> override;
   auto ProcessChunk(const cpp::span<std::byte>& chunk)
       -> cpp::result<std::size_t, AudioProcessingError> override;
-  auto ProcessIdle() -> cpp::result<void, AudioProcessingError> override;
+  auto Process() -> cpp::result<void, AudioProcessingError> override;
 
   AudioDecoder(const AudioDecoder&) = delete;
   AudioDecoder& operator=(const AudioDecoder&) = delete;
@@ -42,8 +45,11 @@ class AudioDecoder : public IAudioElement {
  private:
   std::unique_ptr<codecs::ICodec> current_codec_;
   std::optional<StreamInfo> stream_info_;
+  std::optional<ChunkReader> chunk_reader_;
 
-  std::unique_ptr<ChunkWriter> chunk_writer_;
+  std::size_t chunk_size_;
+  bool has_samples_to_send_;
+  bool needs_more_input_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_element.hpp b/src/audio/include/audio_element.hpp
index eb700180..8827a0c3 100644
--- a/src/audio/include/audio_element.hpp
+++ b/src/audio/include/audio_element.hpp
@@ -2,6 +2,8 @@
 
 #include <atomic>
 #include <cstdint>
+#include <deque>
+#include <memory>
 
 #include "freertos/FreeRTOS.h"
 
@@ -12,6 +14,7 @@
 #include "span.hpp"
 
 #include "stream_buffer.hpp"
+#include "stream_event.hpp"
 #include "stream_info.hpp"
 #include "types.hpp"
 
@@ -36,6 +39,8 @@ enum AudioProcessingError {
   OUT_OF_DATA,
 };
 
+static const size_t kEventQueueSize = 8;
+
 /*
  * 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
@@ -51,38 +56,38 @@ enum AudioProcessingError {
  */
 class IAudioElement {
  public:
-  IAudioElement()
-      : input_buffer_(nullptr), output_buffer_(nullptr), state_(STATE_RUN) {}
-  virtual ~IAudioElement() {}
+  IAudioElement();
+  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.
+   * elements have fairly different stack requirements (particular decoders).
    */
   virtual auto StackSizeBytes() const -> 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() const -> TickType_t { return 10; }
-
   virtual auto InputMinChunkSize() const -> std::size_t { return 0; }
 
   /* Returns this element's input buffer. */
-  auto InputBuffer() const -> StreamBuffer* { return input_buffer_; }
+  auto InputEventQueue() const -> QueueHandle_t { return input_events_; }
 
   /* Returns this element's output buffer. */
-  auto OutputBuffer() const -> StreamBuffer* { return output_buffer_; }
+  auto OutputEventQueue() const -> QueueHandle_t { return output_events_; }
+
+  auto OutputEventQueue(const QueueHandle_t q) -> void { output_events_ = q; }
 
-  auto InputBuffer(StreamBuffer* b) -> void { input_buffer_ = b; }
+  auto HasUnflushedOutput() -> bool { return !buffered_output_.empty(); }
 
-  auto OutputBuffer(StreamBuffer* b) -> void { output_buffer_ = b; }
+  virtual auto HasUnprocessedInput() -> bool = 0;
 
-  auto ElementState() const -> ElementState { return state_; }
-  auto ElementState(enum ElementState e) -> void { state_ = e; }
+  auto IsOverBuffered() -> bool { return unprocessed_output_chunks_ > 4; }
+
+  auto FlushBufferedOutput() -> bool;
+
+  auto ElementState() const -> ElementState { return current_state_; }
+  auto ElementState(enum ElementState e) -> void { current_state_ = e; }
+
+  virtual auto OnChunkProcessed() -> void { unprocessed_output_chunks_--; }
 
   /*
    * Called when a StreamInfo message is received. Used to configure this
@@ -105,14 +110,26 @@ class IAudioElement {
    * time. This could be used to synthesize output, or to save memory by
    * releasing unused resources.
    */
-  virtual auto ProcessIdle() -> cpp::result<void, AudioProcessingError> = 0;
-
-  virtual auto PrepareForPause() -> void{};
+  virtual auto Process() -> cpp::result<void, AudioProcessingError> = 0;
 
  protected:
-  StreamBuffer* input_buffer_;
-  StreamBuffer* output_buffer_;
-  std::atomic<enum ElementState> state_;
+  auto SendOrBufferEvent(std::unique_ptr<StreamEvent> event) -> bool;
+
+  // Queue for events coming into this element. Owned by us.
+  QueueHandle_t input_events_;
+  // Queue for events going into the next element. Not owned by us, may be null
+  // if we're not yet in a pipeline.
+  // FIXME: it would be nicer if this was non-nullable.
+  QueueHandle_t output_events_;
+
+  // The number of output chunks that we have generated, but have not yet been
+  // processed by the next element in the pipeline. This includes any chunks
+  // that are currently help in buffered_output_.
+  int unprocessed_output_chunks_;
+  // Output events that have been generated, but are yet to be sent downstream.
+  std::deque<std::unique_ptr<StreamEvent>> buffered_output_;
+
+  enum ElementState current_state_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_playback.hpp b/src/audio/include/audio_playback.hpp
index bffc3f02..f05ca327 100644
--- a/src/audio/include/audio_playback.hpp
+++ b/src/audio/include/audio_playback.hpp
@@ -43,11 +43,9 @@ class AudioPlayback {
 
  private:
   auto ConnectElements(IAudioElement* src, IAudioElement* sink) -> void;
-
-  StreamBuffer stream_start_;
-  StreamBuffer stream_end_;
-  std::vector<std::unique_ptr<StreamBuffer>> element_buffers_;
   std::vector<std::unique_ptr<AudioElementHandle>> element_handles_;
+
+  QueueHandle_t input_handle_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/chunk.hpp b/src/audio/include/chunk.hpp
index 5c7e73dd..6154ab25 100644
--- a/src/audio/include/chunk.hpp
+++ b/src/audio/include/chunk.hpp
@@ -2,6 +2,7 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <memory>
 #include <optional>
 #include <string>
 
@@ -17,68 +18,12 @@
 
 namespace audio {
 
-enum ChunkWriteResult {
-  // Returned when the callback does not write any data.
-  CHUNK_WRITE_OKAY,
-  // 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,
-};
-
-class ChunkWriter {
- public:
-  explicit ChunkWriter(StreamBuffer* buffer);
-  ~ChunkWriter();
-
-  auto Reset() -> void;
-
-  auto GetLastMessage() -> cpp::span<std::byte>;
-
-  /*
-   * 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 WriteChunkToStream(std::function<size_t(cpp::span<std::byte>)> callback,
-                          TickType_t max_wait) -> ChunkWriteResult;
-
- private:
-  StreamBuffer* stream_;
-  std::size_t leftover_bytes_ = 0;
-};
-
-enum ChunkReadResult {
-  CHUNK_READ_OKAY,
-  // Returned when the chunk was read successfully, but the consumer did not
-  // use all of the data.
-  CHUNK_LEFTOVER_DATA,
-  // 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,
-  // Returned when the processing callback does not return a value.
-  CHUNK_PROCESSING_ERROR,
-};
-
 class ChunkReader {
  public:
-  explicit ChunkReader(StreamBuffer* buffer);
+  explicit ChunkReader(std::size_t chunk_size);
   ~ChunkReader();
 
-  auto Reset() -> void;
-
-  auto GetLastMessage() -> cpp::span<std::byte>;
+  auto HandleLeftovers(std::size_t bytes_used) -> void;
 
   /*
    * Reads chunks of data from the given input stream, and invokes the given
@@ -92,14 +37,13 @@ class ChunkReader {
    * 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<std::size_t>(cpp::span<std::byte>)> callback,
-      TickType_t max_wait) -> ChunkReadResult;
+  auto HandleNewData(cpp::span<std::byte> data) -> cpp::span<std::byte>;
 
  private:
-  StreamBuffer* stream_;
+  std::byte* raw_working_buffer_;
+  cpp::span<std::byte> working_buffer_;
+  cpp::span<std::byte> last_data_in_working_buffer_;
   std::size_t leftover_bytes_ = 0;
-  std::size_t last_message_size_ = 0;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index f3704f1d..5625d941 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/src/audio/include/fatfs_audio_input.hpp
@@ -22,32 +22,22 @@ class FatfsAudioInput : public IAudioElement {
   explicit FatfsAudioInput(std::shared_ptr<drivers::SdStorage> storage);
   ~FatfsAudioInput();
 
+  auto HasUnprocessedInput() -> bool override;
+
   auto ProcessStreamInfo(const StreamInfo& info)
       -> cpp::result<void, AudioProcessingError> override;
   auto ProcessChunk(const cpp::span<std::byte>& chunk)
       -> cpp::result<std::size_t, AudioProcessingError> override;
-  auto ProcessIdle() -> cpp::result<void, AudioProcessingError> override;
-
-  auto SendChunk(cpp::span<std::byte> dest) -> size_t;
+  auto Process() -> cpp::result<void, AudioProcessingError> override;
 
   FatfsAudioInput(const FatfsAudioInput&) = delete;
   FatfsAudioInput& operator=(const FatfsAudioInput&) = delete;
 
  private:
-  auto GetRingBufferDistance() const -> size_t;
-
   std::shared_ptr<drivers::SdStorage> storage_;
 
-  std::byte* raw_file_buffer_;
-  cpp::span<std::byte> file_buffer_;
-  cpp::span<std::byte>::iterator file_buffer_read_pos_;
-  cpp::span<std::byte>::iterator pending_read_pos_;
-  cpp::span<std::byte>::iterator file_buffer_write_pos_;
-
   FIL current_file_;
   bool is_file_open_;
-
-  std::unique_ptr<ChunkWriter> chunk_writer_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/i2s_audio_output.hpp b/src/audio/include/i2s_audio_output.hpp
index 75a3be76..4fbcad49 100644
--- a/src/audio/include/i2s_audio_output.hpp
+++ b/src/audio/include/i2s_audio_output.hpp
@@ -22,19 +22,14 @@ class I2SAudioOutput : public IAudioElement {
                  std::unique_ptr<drivers::AudioDac> dac);
   ~I2SAudioOutput();
 
-  auto InputMinChunkSize() const -> std::size_t override {
-    // TODO(jacqueline): work out a good value here. Maybe similar to the total
-    // DMA buffer size?
-    return 128;
-  }
+  // TODO.
+  auto HasUnprocessedInput() -> bool override { return false; }
 
-  auto IdleTimeout() const -> TickType_t override;
   auto ProcessStreamInfo(const StreamInfo& info)
       -> cpp::result<void, AudioProcessingError> override;
   auto ProcessChunk(const cpp::span<std::byte>& chunk)
       -> cpp::result<std::size_t, AudioProcessingError> override;
-  auto ProcessIdle() -> cpp::result<void, AudioProcessingError> override;
-  auto PrepareForPause() -> void override;
+  auto Process() -> cpp::result<void, AudioProcessingError> override;
 
   I2SAudioOutput(const I2SAudioOutput&) = delete;
   I2SAudioOutput& operator=(const I2SAudioOutput&) = delete;
diff --git a/src/audio/include/stream_event.hpp b/src/audio/include/stream_event.hpp
new file mode 100644
index 00000000..4dfdab41
--- /dev/null
+++ b/src/audio/include/stream_event.hpp
@@ -0,0 +1,57 @@
+#pragma once
+
+#include <memory>
+
+#include "freertos/FreeRTOS.h"
+#include "freertos/queue.h"
+
+#include "span.hpp"
+#include "stream_info.hpp"
+
+namespace audio {
+
+struct StreamEvent {
+  static auto CreateStreamInfo(QueueHandle_t source,
+                               std::unique_ptr<StreamInfo> payload)
+      -> std::unique_ptr<StreamEvent>;
+  static auto CreateChunkData(QueueHandle_t source, std::size_t chunk_size)
+      -> std::unique_ptr<StreamEvent>;
+  static auto CreateChunkNotification(QueueHandle_t source)
+      -> std::unique_ptr<StreamEvent>;
+
+  StreamEvent();
+  ~StreamEvent();
+  StreamEvent(StreamEvent&&);
+
+  QueueHandle_t source;
+
+  enum {
+    UNINITIALISED,
+    STREAM_INFO,
+    CHUNK_DATA,
+    CHUNK_NOTIFICATION,
+  } tag;
+
+  union {
+    std::unique_ptr<StreamInfo> stream_info;
+
+    // Scott Meyers says:
+    // `About the only situation I can conceive of when a std::unique_ptr<T[]>
+    // would make sense would be when you’re using a C-like API that returns a
+    // raw pointer to a heap array that you assume ownership of.`
+    // :-)
+
+    struct {
+      std::unique_ptr<std::byte*> raw_bytes;
+      cpp::span<std::byte> bytes;
+    } chunk_data;
+
+    // FIXME: It would be nice to also support a pointer to himem data here, to
+    // save a little ordinary heap space.
+  };
+
+  StreamEvent(const StreamEvent&) = delete;
+  StreamEvent& operator=(const StreamEvent&) = delete;
+};
+
+}  // namespace audio
diff --git a/src/audio/include/stream_info.hpp b/src/audio/include/stream_info.hpp
index 45f10fc6..ed3096bb 100644
--- a/src/audio/include/stream_info.hpp
+++ b/src/audio/include/stream_info.hpp
@@ -7,6 +7,7 @@
 
 #include "cbor.h"
 #include "result.hpp"
+#include "sys/_stdint.h"
 
 namespace audio {
 
@@ -24,14 +25,24 @@ class StreamInfo {
 
   auto Channels() const -> const std::optional<uint8_t>& { return channels_; }
 
+  auto BitsPerSample(uint8_t bpp) -> void { bits_per_sample_ = bpp; }
+
   auto BitsPerSample() const -> const std::optional<uint8_t>& {
     return bits_per_sample_;
   }
 
+  auto SampleRate(uint16_t rate) -> void { sample_rate_ = rate; }
+
   auto SampleRate() const -> const std::optional<uint16_t>& {
     return sample_rate_;
   }
 
+  auto ChunkSize() const -> const std::optional<std::size_t>& {
+    return chunk_size_;
+  }
+
+  auto ChunkSize(std::size_t s) -> void { chunk_size_ = s; }
+
   auto Encode(CborEncoder& enc) -> std::optional<CborError>;
 
  private:
@@ -39,6 +50,7 @@ class StreamInfo {
   std::optional<uint8_t> channels_;
   std::optional<uint8_t> bits_per_sample_;
   std::optional<uint16_t> sample_rate_;
+  std::optional<size_t> chunk_size_;
 };
 
 }  // namespace audio