Switch from MessageBuffer to Queue for pipeline comms

1 files changed, 31 insertions, 123 deletions

diff --git a/src/audio/fatfs_audio_input.cpp b/src/audio/fatfs_audio_input.cpp
index 7f11805e..29e03784 100644
--- a/src/audio/fatfs_audio_input.cpp
+++ b/src/audio/fatfs_audio_input.cpp
@@ -1,5 +1,6 @@
 #include "fatfs_audio_input.hpp"
 
+#include <algorithm>
 #include <cstdint>
 #include <memory>
 #include <string>
@@ -10,33 +11,26 @@
 #include "audio_element.hpp"
 #include "chunk.hpp"
 #include "stream_buffer.hpp"
+#include "stream_event.hpp"
 #include "stream_message.hpp"
 
 static const char* kTag = "SRC";
 
 namespace audio {
 
-static const TickType_t kServiceInterval = pdMS_TO_TICKS(50);
-
-static const std::size_t kFileBufferSize = 1024 * 128;
-static const std::size_t kMinFileReadSize = 1024 * 4;
+// 32KiB to match the minimum himen region size.
+static const std::size_t kChunkSize = 1024;
 
 FatfsAudioInput::FatfsAudioInput(std::shared_ptr<drivers::SdStorage> storage)
     : IAudioElement(),
       storage_(storage),
-      raw_file_buffer_(static_cast<std::byte*>(
-          heap_caps_malloc(kFileBufferSize, MALLOC_CAP_SPIRAM))),
-      file_buffer_(raw_file_buffer_, kFileBufferSize),
-      file_buffer_read_pos_(file_buffer_.begin()),
-      file_buffer_write_pos_(file_buffer_.begin()),
       current_file_(),
-      is_file_open_(false),
-      chunk_writer_(nullptr) {
-  // TODO: create our chunk writer whenever the output buffer changes.
-}
+      is_file_open_(false) {}
+
+FatfsAudioInput::~FatfsAudioInput() {}
 
-FatfsAudioInput::~FatfsAudioInput() {
-  free(raw_file_buffer_);
+auto FatfsAudioInput::HasUnprocessedInput() -> bool {
+  return is_file_open_;
 }
 
 auto FatfsAudioInput::ProcessStreamInfo(const StreamInfo& info)
@@ -57,17 +51,12 @@ auto FatfsAudioInput::ProcessStreamInfo(const StreamInfo& info)
 
   is_file_open_ = true;
 
-  auto write_size =
-      WriteMessage(TYPE_STREAM_INFO,
-                   std::bind(&StreamInfo::Encode, info, std::placeholders::_1),
-                   output_buffer_->WriteBuffer());
+  std::unique_ptr<StreamInfo> new_info = std::make_unique<StreamInfo>(info);
+  new_info->ChunkSize(kChunkSize);
 
-  if (write_size.has_error()) {
-    return cpp::fail(IO_ERROR);
-  } else {
-    xMessageBufferSend(output_buffer_, output_buffer_->WriteBuffer().data(),
-                       write_size.value(), portMAX_DELAY);
-  }
+  auto event =
+      StreamEvent::CreateStreamInfo(input_events_, std::move(new_info));
+  SendOrBufferEvent(std::move(event));
 
   return {};
 }
@@ -77,110 +66,29 @@ auto FatfsAudioInput::ProcessChunk(const cpp::span<std::byte>& chunk)
   return cpp::fail(UNSUPPORTED_STREAM);
 }
 
-auto FatfsAudioInput::GetRingBufferDistance() const -> size_t {
-  if (file_buffer_read_pos_ == file_buffer_write_pos_) {
-    return 0;
-  }
-  if (file_buffer_read_pos_ < file_buffer_write_pos_) {
-    return file_buffer_write_pos_ - file_buffer_read_pos_;
-  }
-  return
-      // Read position to end of buffer.
-      (file_buffer_.end() - file_buffer_read_pos_)
-      // Start of buffer to write position.
-      + (file_buffer_write_pos_ - file_buffer_.begin());
-}
-
-auto FatfsAudioInput::ProcessIdle() -> cpp::result<void, AudioProcessingError> {
-  // First, see if we're able to fill up the input buffer with any more of the
-  // file's contents.
+auto FatfsAudioInput::Process() -> cpp::result<void, AudioProcessingError> {
   if (is_file_open_) {
-    size_t ringbuf_distance = GetRingBufferDistance();
-    if (file_buffer_.size() - ringbuf_distance > kMinFileReadSize) {
-      size_t read_size;
-      if (file_buffer_write_pos_ < file_buffer_read_pos_) {
-        // Don't worry about the start of buffer -> read pos size; we can get to
-        // it next iteration.
-        read_size = file_buffer_read_pos_ - file_buffer_write_pos_;
-      } else {
-        read_size = file_buffer_.begin() - file_buffer_write_pos_;
-      }
-
-      ESP_LOGI(kTag, "reading up to %d bytes", (int)read_size);
-
-      UINT bytes_read = 0;
-      FRESULT result =
-          f_read(&current_file_, std::addressof(file_buffer_write_pos_),
-                 read_size, &bytes_read);
-      if (result != FR_OK) {
-        ESP_LOGE(kTag, "file I/O error %d", result);
-        return cpp::fail(IO_ERROR);
-      }
-
-      ESP_LOGI(kTag, "actual read size %d bytes", (int)bytes_read);
-
-      if (f_eof(&current_file_)) {
-        f_close(&current_file_);
-        is_file_open_ = false;
-
-        // TODO: open the next file?
-      }
-
-      file_buffer_write_pos_ += bytes_read;
-      if (file_buffer_write_pos_ == file_buffer_.end()) {
-        file_buffer_write_pos_ = file_buffer_.begin();
-      }
+    auto dest_event = StreamEvent::CreateChunkData(input_events_, kChunkSize);
+    UINT bytes_read = 0;
+
+    FRESULT result =
+        f_read(&current_file_, dest_event->chunk_data.raw_bytes.get(),
+               kChunkSize, &bytes_read);
+    if (result != FR_OK) {
+      ESP_LOGE(kTag, "file I/O error %d", result);
+      return cpp::fail(IO_ERROR);
     }
-  } else if (GetRingBufferDistance() == 0) {
-    // We have no file open, and no data waiting to be written. We're out of
-    // stuff to do, so signal a pause.
-    return cpp::fail(OUT_OF_DATA);
-  }
 
-  // Now stream data into the output buffer until it's full.
-  while (GetRingBufferDistance() > 0) {
-    ESP_LOGI(kTag, "writing up to %d bytes", (int)GetRingBufferDistance());
-    ChunkWriteResult result = chunk_writer_->WriteChunkToStream(
-        [&](cpp::span<std::byte> d) { return SendChunk(d); }, kServiceInterval);
-
-    switch (result) {
-      case CHUNK_WRITE_OKAY:
-        break;
-      case CHUNK_WRITE_TIMEOUT:
-      case CHUNK_OUT_OF_DATA:
-        // Both of these are fine; we will pick back up where we left off in
-        // the next idle call.
-        return {};
-      default:
-        return cpp::fail(IO_ERROR);
+    dest_event->chunk_data.bytes =
+        dest_event->chunk_data.bytes.first(bytes_read);
+    SendOrBufferEvent(std::move(dest_event));
+
+    if (f_eof(&current_file_)) {
+      f_close(&current_file_);
+      is_file_open_ = false;
     }
   }
-
-  // We've finished writing out chunks, but there may be more of the file to
-  // read. Return, and begin again in the next idle call.
   return {};
 }
 
-auto FatfsAudioInput::SendChunk(cpp::span<std::byte> dest) -> size_t {
-  if (file_buffer_read_pos_ == file_buffer_write_pos_) {
-    return 0;
-  }
-  std::size_t chunk_size;
-  if (file_buffer_read_pos_ > file_buffer_write_pos_) {
-    chunk_size = file_buffer_.end() - file_buffer_read_pos_;
-  } else {
-    chunk_size = file_buffer_write_pos_ - file_buffer_read_pos_;
-  }
-  chunk_size = std::min(chunk_size, dest.size());
-
-  cpp::span<std::byte> source(file_buffer_read_pos_, chunk_size);
-  std::copy(source.begin(), source.end(), dest.begin());
-
-  file_buffer_read_pos_ = file_buffer_read_pos_ + chunk_size;
-  if (file_buffer_read_pos_ == file_buffer_.end()) {
-    file_buffer_read_pos_ = file_buffer_.begin();
-  }
-  return chunk_size;
-}
-
 }  // namespace audio