Improve decoder's interface to accept streams

1 files changed, 101 insertions, 50 deletions

diff --git a/src/tangara/audio/audio_decoder.cpp b/src/tangara/audio/audio_decoder.cpp
index ae54a11c..0e38bca8 100644
--- a/src/tangara/audio/audio_decoder.cpp
+++ b/src/tangara/audio/audio_decoder.cpp
@@ -6,7 +6,7 @@
 
 #include "audio/audio_decoder.hpp"
 
-#include <algorithm>
+#include <cassert>
 #include <cmath>
 #include <cstddef>
 #include <cstdint>
@@ -23,14 +23,12 @@
 #include "freertos/portmacro.h"
 #include "freertos/projdefs.h"
 #include "freertos/queue.h"
-#include "freertos/ringbuf.h"
 
-#include "audio/audio_converter.hpp"
 #include "audio/audio_events.hpp"
 #include "audio/audio_fsm.hpp"
 #include "audio/audio_sink.hpp"
 #include "audio/audio_source.hpp"
-#include "audio/fatfs_audio_input.hpp"
+#include "audio/processor.hpp"
 #include "codec.hpp"
 #include "database/track.hpp"
 #include "drivers/i2s_dac.hpp"
@@ -42,21 +40,33 @@
 
 namespace audio {
 
-[[maybe_unused]] static const char* kTag = "audio_dec";
+static const char* kTag = "decoder";
 
+/*
+ * The size of the buffer used for holding decoded samples. This buffer is
+ * allocated in internal memory for greater speed, so be careful when
+ * increasing its size.
+ */
 static constexpr std::size_t kCodecBufferLength =
     drivers::kI2SBufferLengthFrames * sizeof(sample::Sample);
 
-auto Decoder::Start(std::shared_ptr<IAudioSource> source,
-                    std::shared_ptr<SampleConverter> sink) -> Decoder* {
-  Decoder* task = new Decoder(source, sink);
+auto Decoder::Start(std::shared_ptr<SampleProcessor> sink) -> Decoder* {
+  Decoder* task = new Decoder(sink);
   tasks::StartPersistent<tasks::Type::kAudioDecoder>([=]() { task->Main(); });
   return task;
 }
 
-Decoder::Decoder(std::shared_ptr<IAudioSource> source,
-                 std::shared_ptr<SampleConverter> mixer)
-    : source_(source), converter_(mixer), codec_(), current_format_() {
+auto Decoder::open(std::shared_ptr<TaggedStream> stream) -> void {
+  NextStream* next = new NextStream();
+  next->stream = stream;
+  // The decoder services its queue very quickly, so blocking on this write
+  // should be fine. If we discover contention here, then adding more space for
+  // items to next_stream_ should be fine too.
+  xQueueSend(next_stream_, &next, portMAX_DELAY);
+}
+
+Decoder::Decoder(std::shared_ptr<SampleProcessor> processor)
+    : processor_(processor), next_stream_(xQueueCreate(1, sizeof(void*))) {
   ESP_LOGI(kTag, "allocating codec buffer, %u KiB", kCodecBufferLength / 1024);
   codec_buffer_ = {
       reinterpret_cast<sample::Sample*>(heap_caps_calloc(
@@ -64,81 +74,122 @@ Decoder::Decoder(std::shared_ptr<IAudioSource> source,
       kCodecBufferLength};
 }
 
+/*
+ * Main decoding loop. Handles watching for new streams, or continuing to nudge
+ * along the current stream if we have one.
+ */
 void Decoder::Main() {
   for (;;) {
-    if (source_->HasNewStream() || !stream_) {
-      std::shared_ptr<TaggedStream> new_stream = source_->NextStream();
-      if (new_stream && BeginDecoding(new_stream)) {
-        stream_ = new_stream;
-      } else {
+    // Check whether there's a new stream to begin. If we're idle, then we
+    // simply park and wait forever for a stream to arrive.
+    TickType_t wait_time = stream_ ? 0 : portMAX_DELAY;
+    NextStream* next;
+    if (xQueueReceive(next_stream_, &next, wait_time)) {
+      // Copy the data out of the queue, then clean up the item.
+      std::shared_ptr<TaggedStream> new_stream = next->stream;
+      delete next;
+
+      // If we were already decoding, then make sure we finish up the current
+      // file gracefully.
+      if (stream_) {
+        finishDecode();
+      }
+
+      // Ensure there's actually stream data; we might have been given nullptr
+      // as a signal to stop.
+      if (!new_stream) {
         continue;
       }
+
+      // Start decoding the new stream.
+      prepareDecode(new_stream);
     }
 
-    if (ContinueDecoding()) {
-      stream_.reset();
+    if (!continueDecode()) {
+      finishDecode();
     }
   }
 }
 
-auto Decoder::BeginDecoding(std::shared_ptr<TaggedStream> stream) -> bool {
-  // Ensure any previous codec is freed before creating a new one.
-  codec_.reset();
+auto Decoder::prepareDecode(std::shared_ptr<TaggedStream> stream) -> void {
+  auto stub_track = std::make_shared<TrackInfo>(TrackInfo{
+      .tags = stream->tags(),
+      .uri = stream->Filepath(),
+      .duration = {},
+      .start_offset = {},
+      .bitrate_kbps = {},
+      .encoding = stream->type(),
+      .format = {},
+  });
+
   codec_.reset(codecs::CreateCodecForType(stream->type()).value_or(nullptr));
   if (!codec_) {
     ESP_LOGE(kTag, "no codec found for stream");
-    return false;
+    events::Audio().Dispatch(
+        internal::DecodingFailedToStart{.track = stub_track});
+    return;
   }
 
   auto open_res = codec_->OpenStream(stream, stream->Offset());
   if (open_res.has_error()) {
     ESP_LOGE(kTag, "codec failed to start: %s",
              codecs::ICodec::ErrorString(open_res.error()).c_str());
-    return false;
-  }
-  stream->SetPreambleFinished();
-  current_sink_format_ = IAudioOutput::Format{
-      .sample_rate = open_res->sample_rate_hz,
-      .num_channels = open_res->num_channels,
-      .bits_per_sample = 16,
-  };
-
-  std::optional<uint32_t> duration;
-  if (open_res->total_samples) {
-    duration = open_res->total_samples.value() / open_res->num_channels /
-               open_res->sample_rate_hz;
+    events::Audio().Dispatch(
+        internal::DecodingFailedToStart{.track = stub_track});
+    return;
   }
 
-  converter_->beginStream(std::make_shared<TrackInfo>(TrackInfo{
+  // Decoding started okay! Fill out the rest of the track info for this
+  // stream.
+  stream_ = stream;
+  track_ = std::make_shared<TrackInfo>(TrackInfo{
       .tags = stream->tags(),
       .uri = stream->Filepath(),
-      .duration = duration,
+      .duration = {},
       .start_offset = stream->Offset(),
-      .bitrate_kbps = open_res->sample_rate_hz,
+      .bitrate_kbps = {},
       .encoding = stream->type(),
-      .format = *current_sink_format_,
-  }));
+      .format =
+          {
+              .sample_rate = open_res->sample_rate_hz,
+              .num_channels = open_res->num_channels,
+              .bits_per_sample = 16,
+          },
+  });
+
+  if (open_res->total_samples) {
+    track_->duration = open_res->total_samples.value() /
+                       open_res->num_channels / open_res->sample_rate_hz;
+  }
 
-  return true;
+  events::Audio().Dispatch(internal::DecodingStarted{.track = track_});
+  processor_->beginStream(track_);
 }
 
-auto Decoder::ContinueDecoding() -> bool {
+auto Decoder::continueDecode() -> bool {
   auto res = codec_->DecodeTo(codec_buffer_);
   if (res.has_error()) {
-    converter_->endStream();
-    return true;
+    return false;
   }
 
   if (res->samples_written > 0) {
-    converter_->continueStream(codec_buffer_.first(res->samples_written));
+    processor_->continueStream(codec_buffer_.first(res->samples_written));
   }
 
-  if (res->is_stream_finished) {
-    converter_->endStream();
-    codec_.reset();
-  }
+  return !res->is_stream_finished;
+}
 
-  return res->is_stream_finished;
+auto Decoder::finishDecode() -> void {
+  assert(track_);
+
+  // Tell everyone we're finished.
+  events::Audio().Dispatch(internal::DecodingFinished{.track = track_});
+  processor_->endStream();
+
+  // Clean up after ourselves.
+  stream_.reset();
+  codec_.reset();
+  track_.reset();
 }
 
 }  // namespace audio