From f6dcd845fc80da4e3043146e4362258dd8e0c0a1 Mon Sep 17 00:00:00 2001
From: jacqueline <me@jacqueline.id.au>
Date: Thu, 26 Jan 2023 15:02:57 +1100
Subject: Switch from MessageBuffer to Queue for pipeline comms

---
 src/audio/audio_decoder.cpp | 111 +++++++++++++++++++++++---------------------
 1 file changed, 59 insertions(+), 52 deletions(-)

(limited to 'src/audio/audio_decoder.cpp')

diff --git a/src/audio/audio_decoder.cpp b/src/audio/audio_decoder.cpp
index 0b3d9878..8ef90905 100644
--- a/src/audio/audio_decoder.cpp
+++ b/src/audio/audio_decoder.cpp
@@ -4,6 +4,7 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <memory>
 
 #include "freertos/FreeRTOS.h"
 
@@ -14,19 +15,30 @@
 #include "audio_element.hpp"
 #include "chunk.hpp"
 #include "fatfs_audio_input.hpp"
-
-static const char* kTag = "DEC";
+#include "stream_info.hpp"
 
 namespace audio {
 
+static const std::size_t kSamplesPerChunk = 256;
+
 AudioDecoder::AudioDecoder() : IAudioElement(), stream_info_({}) {}
 
 AudioDecoder::~AudioDecoder() {}
 
+auto AudioDecoder::HasUnprocessedInput() -> bool {
+  return !needs_more_input_ || has_samples_to_send_;
+}
+
 auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info)
     -> cpp::result<void, AudioProcessingError> {
   stream_info_ = info;
 
+  if (info.ChunkSize()) {
+    chunk_reader_.emplace(info.ChunkSize().value());
+  } else {
+    // TODO.
+  }
+
   // Reuse the existing codec if we can. This will help with gapless playback,
   // since we can potentially just continue to decode as we were before,
   // without any setup overhead.
@@ -42,71 +54,66 @@ auto AudioDecoder::ProcessStreamInfo(const StreamInfo& info)
     return cpp::fail(UNSUPPORTED_STREAM);
   }
 
+  // TODO: defer until first header read, so we can give better info about
+  // sample rate, chunk size, etc.
+  auto downstream_info = StreamEvent::CreateStreamInfo(
+      input_events_, std::make_unique<StreamInfo>(info));
+  downstream_info->stream_info->BitsPerSample(32);
+  downstream_info->stream_info->SampleRate(48'000);
+  chunk_size_ = 128;
+  downstream_info->stream_info->ChunkSize(chunk_size_);
+
+  SendOrBufferEvent(std::move(downstream_info));
+
   return {};
 }
 
 auto AudioDecoder::ProcessChunk(const cpp::span<std::byte>& chunk)
     -> cpp::result<size_t, AudioProcessingError> {
-  if (current_codec_ == nullptr) {
+  if (current_codec_ == nullptr || !chunk_reader_) {
     // Should never happen, but fail explicitly anyway.
     return cpp::fail(UNSUPPORTED_STREAM);
   }
 
-  current_codec_->SetInput(chunk);
-
-  bool has_samples_to_send = false;
-  bool needs_more_input = false;
-  std::optional<codecs::ICodec::ProcessingError> error = std::nullopt;
-  while (1) {
-    ChunkWriteResult res = chunk_writer_->WriteChunkToStream(
-        [&](cpp::span<std::byte> buffer) -> std::size_t {
-          std::size_t bytes_written = 0;
-          // Continue filling up the output buffer so long as we have samples
-          // leftover, or are able to synthesize more samples from the input.
-          while (has_samples_to_send || !needs_more_input) {
-            if (!has_samples_to_send) {
-              auto result = current_codec_->ProcessNextFrame();
-              has_samples_to_send = true;
-              if (result.has_error()) {
-                error = result.error();
-                // End our output stream immediately if the codec barfed.
-                return 0;
-              } else {
-                needs_more_input = result.value();
-              }
-            } else {
-              auto result = current_codec_->WriteOutputSamples(
-                  buffer.last(buffer.size() - bytes_written));
-              bytes_written += result.first;
-              has_samples_to_send = !result.second;
-            }
-          }
-          return bytes_written;
-        },
-        // TODO
-        portMAX_DELAY);
-
-    switch (res) {
-      case CHUNK_WRITE_OKAY:
-        break;
-      case CHUNK_WRITE_TIMEOUT:
-      case CHUNK_OUT_OF_DATA:
+  current_codec_->SetInput(chunk_reader_->HandleNewData(chunk));
+
+  return {};
+}
+
+auto AudioDecoder::Process() -> cpp::result<void, AudioProcessingError> {
+  if (has_samples_to_send_) {
+    // Writing samples is relatively quick (it's just a bunch of memcopy's), so
+    // do them all at once.
+    while (has_samples_to_send_ && !IsOverBuffered()) {
+      auto buffer = StreamEvent::CreateChunkData(input_events_, chunk_size_);
+      auto write_res =
+          current_codec_->WriteOutputSamples(buffer->chunk_data.bytes);
+      buffer->chunk_data.bytes =
+          buffer->chunk_data.bytes.first(write_res.first);
+      has_samples_to_send_ = !write_res.second;
+
+      if (!SendOrBufferEvent(std::move(buffer))) {
         return {};
-      default:
-        return cpp::fail(IO_ERROR);
+      }
     }
+    // We will process the next frame during the next call to this method.
+    return {};
   }
 
-  if (error) {
-    ESP_LOGE(kTag, "Codec encountered error %d", error.value());
-    return cpp::fail(IO_ERROR);
-  }
+  if (!needs_more_input_) {
+    auto res = current_codec_->ProcessNextFrame();
+    if (res.has_error()) {
+      // todo
+      return {};
+    }
+    needs_more_input_ = res.value();
+    has_samples_to_send_ = true;
 
-  return current_codec_->GetInputPosition();
-}
+    if (needs_more_input_) {
+      chunk_reader_->HandleLeftovers(current_codec_->GetInputPosition());
+    }
+  }
 
-auto AudioDecoder::ProcessIdle() -> cpp::result<void, AudioProcessingError> {
-  // Not used; we delay forever when waiting on IO.
   return {};
 }
 
-- 
cgit v1.2.3