pass stream start/update/end events through the whole pipeline

8 files changed, 173 insertions, 151 deletions

diff --git a/dependencies.lock b/dependencies.lock
index f4489997..a9723d1e 100644
--- a/dependencies.lock
+++ b/dependencies.lock
@@ -4,6 +4,6 @@ dependencies:
     source:
       type: idf
     version: 5.1.1
-manifest_hash: 9e4320e6f25503854c6c93bcbfa9b80f780485bcf066bdbad31a820544492538
+manifest_hash: b9761e0028130d307b778c710e5dd39fb3c942d8084ed429d448d938957fb0e6
 target: esp32
 version: 1.0.0
diff --git a/src/audio/audio_converter.cpp b/src/audio/audio_converter.cpp
index 1b233731..ebbd405f 100644
--- a/src/audio/audio_converter.cpp
+++ b/src/audio/audio_converter.cpp
@@ -28,7 +28,7 @@
 [[maybe_unused]] static constexpr char kTag[] = "mixer";
 
 static constexpr std::size_t kSampleBufferLength =
-    drivers::kI2SBufferLengthFrames * sizeof(sample::Sample);
+    drivers::kI2SBufferLengthFrames * sizeof(sample::Sample) * 2;
 static constexpr std::size_t kSourceBufferLength = kSampleBufferLength * 2;
 
 namespace audio {
@@ -68,24 +68,32 @@ auto SampleConverter::SetOutput(std::shared_ptr<IAudioOutput> output) -> void {
   sink_ = output;
 }
 
-auto SampleConverter::ConvertSamples(cpp::span<sample::Sample> input,
-                                     const IAudioOutput::Format& format,
-                                     bool is_eos) -> void {
+auto SampleConverter::beginStream(std::shared_ptr<TrackInfo> track) -> void {
   Args args{
-      .format = format,
+      .track = new std::shared_ptr<TrackInfo>(track),
+      .samples_available = 0,
+      .is_end_of_stream = false,
+  };
+  xQueueSend(commands_, &args, portMAX_DELAY);
+}
+
+auto SampleConverter::continueStream(cpp::span<sample::Sample> input) -> void {
+  Args args{
+      .track = nullptr,
       .samples_available = input.size(),
-      .is_end_of_stream = is_eos,
+      .is_end_of_stream = false,
   };
   xQueueSend(commands_, &args, portMAX_DELAY);
+  xStreamBufferSend(source_, input.data(), input.size_bytes(), portMAX_DELAY);
+}
 
-  cpp::span<std::byte> input_as_bytes = {
-      reinterpret_cast<std::byte*>(input.data()), input.size_bytes()};
-  size_t bytes_sent = 0;
-  while (bytes_sent < input_as_bytes.size()) {
-    bytes_sent += xStreamBufferSend(
-        source_, input_as_bytes.subspan(bytes_sent).data(),
-        input_as_bytes.size() - bytes_sent, pdMS_TO_TICKS(100));
-  }
+auto SampleConverter::endStream() -> void {
+  Args args{
+      .track = nullptr,
+      .samples_available = 0,
+      .is_end_of_stream = true,
+  };
+  xQueueSend(commands_, &args, portMAX_DELAY);
 }
 
 auto SampleConverter::Main() -> void {
@@ -93,86 +101,93 @@ auto SampleConverter::Main() -> void {
     Args args;
     while (!xQueueReceive(commands_, &args, portMAX_DELAY)) {
     }
-    if (args.format != source_format_) {
-      resampler_.reset();
-      source_format_ = args.format;
-      leftover_bytes_ = 0;
-
-      auto new_target = sink_->PrepareFormat(args.format);
-      if (new_target != target_format_) {
-        // The new format is different to the old one. Wait for the sink to
-        // drain before continuing.
-        while (!xStreamBufferIsEmpty(sink_->stream())) {
-          ESP_LOGI(kTag, "waiting for sink stream to drain...");
-          // TODO(jacqueline): Get the sink drain ISR to notify us of this
-          // via semaphore instead of busy-ish waiting.
-          vTaskDelay(pdMS_TO_TICKS(10));
-        }
-
-        sink_->Configure(new_target);
-      }
-      target_format_ = new_target;
 
-      // Send a final sample count for the previous sample rate.
-      if (samples_sunk_ > 0) {
-        events::Audio().Dispatch(internal::ConverterProgress{
-            .samples_sunk = samples_sunk_,
-        });
+    if (args.track) {
+      handleBeginStream(*args.track);
+      delete args.track;
+    }
+    if (args.samples_available) {
+      handleContinueStream(args.samples_available);
+    }
+    if (args.is_end_of_stream) {
+      handleEndStream();
+    }
+  }
+}
+
+auto SampleConverter::handleBeginStream(std::shared_ptr<TrackInfo> track)
+    -> void {
+  if (track->format != source_format_) {
+    resampler_.reset();
+    source_format_ = track->format;
+    leftover_bytes_ = 0;
+
+    auto new_target = sink_->PrepareFormat(track->format);
+    if (new_target != target_format_) {
+      // The new format is different to the old one. Wait for the sink to
+      // drain before continuing.
+      while (!xStreamBufferIsEmpty(sink_->stream())) {
+        ESP_LOGI(kTag, "waiting for sink stream to drain...");
+        // TODO(jacqueline): Get the sink drain ISR to notify us of this
+        // via semaphore instead of busy-ish waiting.
+        vTaskDelay(pdMS_TO_TICKS(10));
       }
 
-      samples_sunk_ = 0;
-      events::Audio().Dispatch(internal::ConverterConfigurationChanged{
-          .src_format = source_format_,
-          .dst_format = target_format_,
-      });
+      sink_->Configure(new_target);
     }
+    target_format_ = new_target;
+  }
 
-    // Loop until we finish reading all the bytes indicated. There might be
-    // leftovers from each iteration, and from this process as a whole,
-    // depending on the resampling stage.
-    size_t bytes_read = 0;
-    size_t bytes_to_read = args.samples_available * sizeof(sample::Sample);
-    while (bytes_read < bytes_to_read) {
-      // First top up the input buffer, taking care not to overwrite anything
-      // remaining from a previous iteration.
-      size_t bytes_read_this_it = xStreamBufferReceive(
-          source_, input_buffer_as_bytes_.subspan(leftover_bytes_).data(),
-          std::min(input_buffer_as_bytes_.size() - leftover_bytes_,
-                   bytes_to_read - bytes_read),
-          portMAX_DELAY);
-      bytes_read += bytes_read_this_it;
-
-      // Calculate the number of whole samples that are now in the input buffer.
-      size_t bytes_in_buffer = bytes_read_this_it + leftover_bytes_;
-      size_t samples_in_buffer = bytes_in_buffer / sizeof(sample::Sample);
-
-      size_t samples_used =
-          HandleSamples(input_buffer_.first(samples_in_buffer),
-                        args.is_end_of_stream && bytes_read == bytes_to_read);
-
-      // Maybe the resampler didn't consume everything. Maybe the last few
-      // bytes we read were half a frame. Either way, we need to calculate the
-      // size of the remainder in bytes, then move it to the front of our
-      // buffer.
-      size_t bytes_used = samples_used * sizeof(sample::Sample);
-      assert(bytes_used <= bytes_in_buffer);
-
-      leftover_bytes_ = bytes_in_buffer - bytes_used;
-      if (leftover_bytes_ > 0) {
-        std::memmove(input_buffer_as_bytes_.data(),
-                     input_buffer_as_bytes_.data() + bytes_used,
-                     leftover_bytes_);
-      }
+  samples_sunk_ = 0;
+  events::Audio().Dispatch(internal::StreamStarted{
+      .track = track,
+      .src_format = source_format_,
+      .dst_format = target_format_,
+  });
+}
+
+auto SampleConverter::handleContinueStream(size_t samples_available) -> void {
+  // Loop until we finish reading all the bytes indicated. There might be
+  // leftovers from each iteration, and from this process as a whole,
+  // depending on the resampling stage.
+  size_t bytes_read = 0;
+  size_t bytes_to_read = samples_available * sizeof(sample::Sample);
+  while (bytes_read < bytes_to_read) {
+    // First top up the input buffer, taking care not to overwrite anything
+    // remaining from a previous iteration.
+    size_t bytes_read_this_it = xStreamBufferReceive(
+        source_, input_buffer_as_bytes_.subspan(leftover_bytes_).data(),
+        std::min(input_buffer_as_bytes_.size() - leftover_bytes_,
+                 bytes_to_read - bytes_read),
+        portMAX_DELAY);
+    bytes_read += bytes_read_this_it;
+
+    // Calculate the number of whole samples that are now in the input buffer.
+    size_t bytes_in_buffer = bytes_read_this_it + leftover_bytes_;
+    size_t samples_in_buffer = bytes_in_buffer / sizeof(sample::Sample);
+
+    size_t samples_used = handleSamples(input_buffer_.first(samples_in_buffer));
+
+    // Maybe the resampler didn't consume everything. Maybe the last few
+    // bytes we read were half a frame. Either way, we need to calculate the
+    // size of the remainder in bytes, then move it to the front of our
+    // buffer.
+    size_t bytes_used = samples_used * sizeof(sample::Sample);
+    assert(bytes_used <= bytes_in_buffer);
+
+    leftover_bytes_ = bytes_in_buffer - bytes_used;
+    if (leftover_bytes_ > 0) {
+      std::memmove(input_buffer_as_bytes_.data(),
+                   input_buffer_as_bytes_.data() + bytes_used, leftover_bytes_);
     }
   }
 }
 
-auto SampleConverter::HandleSamples(cpp::span<sample::Sample> input,
-                                    bool is_eos) -> size_t {
+auto SampleConverter::handleSamples(cpp::span<sample::Sample> input) -> size_t {
   if (source_format_ == target_format_) {
     // The happiest possible case: the input format matches the output
     // format already.
-    SendToSink(input);
+    sendToSink(input);
     return input.size();
   }
 
@@ -190,7 +205,7 @@ auto SampleConverter::HandleSamples(cpp::span<sample::Sample> input,
 
       size_t read, written;
       std::tie(read, written) = resampler_->Process(input.subspan(samples_used),
-                                                    resampled_buffer_, is_eos);
+                                                    resampled_buffer_, false);
       samples_used += read;
 
       if (read == 0 && written == 0) {
@@ -203,18 +218,40 @@ auto SampleConverter::HandleSamples(cpp::span<sample::Sample> input,
       samples_used = input.size();
     }
 
-    SendToSink(output_source);
+    sendToSink(output_source);
   }
+
   return samples_used;
 }
 
-auto SampleConverter::SendToSink(cpp::span<sample::Sample> samples) -> void {
+auto SampleConverter::handleEndStream() -> void {
+  if (resampler_) {
+    size_t read, written;
+    std::tie(read, written) = resampler_->Process({}, resampled_buffer_, true);
+
+    if (written > 0) {
+      sendToSink(resampled_buffer_.first(written));
+    }
+  }
+
+  // Send a final update to finish off this stream's samples.
+  if (samples_sunk_ > 0) {
+    events::Audio().Dispatch(internal::StreamUpdate{
+        .samples_sunk = samples_sunk_,
+    });
+    samples_sunk_ = 0;
+  }
+
+  events::Audio().Dispatch(internal::StreamEnded{});
+}
+
+auto SampleConverter::sendToSink(cpp::span<sample::Sample> samples) -> void {
   // Update the number of samples sunk so far *before* actually sinking them,
   // since writing to the stream buffer will block when the buffer gets full.
   samples_sunk_ += samples.size();
   if (samples_sunk_ >=
       target_format_.sample_rate * target_format_.num_channels) {
-    events::Audio().Dispatch(internal::ConverterProgress{
+    events::Audio().Dispatch(internal::StreamUpdate{
         .samples_sunk = samples_sunk_,
     });
     samples_sunk_ = 0;
diff --git a/src/audio/audio_decoder.cpp b/src/audio/audio_decoder.cpp
index 55ebc0ec..90c69c16 100644
--- a/src/audio/audio_decoder.cpp
+++ b/src/audio/audio_decoder.cpp
@@ -72,7 +72,6 @@ void Decoder::Main() {
   for (;;) {
     if (source_->HasNewStream() || !stream_) {
       std::shared_ptr<TaggedStream> new_stream = source_->NextStream();
-      ESP_LOGI(kTag, "decoder has new stream");
       if (new_stream && BeginDecoding(new_stream)) {
         stream_ = new_stream;
       } else {
@@ -91,8 +90,7 @@ auto Decoder::BeginDecoding(std::shared_ptr<TaggedStream> stream) -> bool {
   codec_.reset();
   codec_.reset(codecs::CreateCodecForType(stream->type()).value_or(nullptr));
   if (!codec_) {
-    ESP_LOGE(kTag, "no codec found");
-    events::Audio().Dispatch(internal::DecoderError{});
+    ESP_LOGE(kTag, "no codec found for stream");
     return false;
   }
 
@@ -100,7 +98,6 @@ auto Decoder::BeginDecoding(std::shared_ptr<TaggedStream> stream) -> bool {
   if (open_res.has_error()) {
     ESP_LOGE(kTag, "codec failed to start: %s",
              codecs::ICodec::ErrorString(open_res.error()).c_str());
-    events::Audio().Dispatch(internal::DecoderError{});
     return false;
   }
   stream->SetPreambleFinished();
@@ -110,24 +107,21 @@ auto Decoder::BeginDecoding(std::shared_ptr<TaggedStream> stream) -> bool {
       .bits_per_sample = 16,
   };
 
-  ESP_LOGI(kTag, "stream started ok");
-
   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::DecoderOpened{
-      .track = std::make_shared<TrackInfo>(TrackInfo{
-          .tags = stream->tags(),
-          .uri = stream->Filepath(),
-          .duration = duration,
-          .start_offset = stream->Offset(),
-          .bitrate_kbps = open_res->sample_rate_hz,
-          .encoding = stream->type(),
-      }),
-  });
+  converter_->beginStream(std::make_shared<TrackInfo>(TrackInfo{
+      .tags = stream->tags(),
+      .uri = stream->Filepath(),
+      .duration = duration,
+      .start_offset = stream->Offset(),
+      .bitrate_kbps = open_res->sample_rate_hz,
+      .encoding = stream->type(),
+      .format = *current_sink_format_,
+  }));
 
   return true;
 }
@@ -135,18 +129,16 @@ auto Decoder::BeginDecoding(std::shared_ptr<TaggedStream> stream) -> bool {
 auto Decoder::ContinueDecoding() -> bool {
   auto res = codec_->DecodeTo(codec_buffer_);
   if (res.has_error()) {
-    events::Audio().Dispatch(internal::DecoderError{});
+    converter_->endStream();
     return true;
   }
 
   if (res->samples_written > 0) {
-    converter_->ConvertSamples(codec_buffer_.first(res->samples_written),
-                               current_sink_format_.value(),
-                               res->is_stream_finished);
+    converter_->continueStream(codec_buffer_.first(res->samples_written));
   }
 
   if (res->is_stream_finished) {
-    events::Audio().Dispatch(internal::DecoderClosed{});
+    converter_->endStream();
     codec_.reset();
   }
 
diff --git a/src/audio/audio_fsm.cpp b/src/audio/audio_fsm.cpp
index 7a138cba..a6f4f4d1 100644
--- a/src/audio/audio_fsm.cpp
+++ b/src/audio/audio_fsm.cpp
@@ -55,9 +55,9 @@ std::shared_ptr<BluetoothAudioOutput> AudioState::sBtOutput;
 std::shared_ptr<IAudioOutput> AudioState::sOutput;
 
 // Two seconds of samples for two channels, at a representative sample rate.
-constexpr size_t kDrainLatencySamples = 48000;
+constexpr size_t kDrainLatencySamples = 48000 * 2 * 2;
 constexpr size_t kDrainBufferSize =
-    sizeof(sample::Sample) * kDrainLatencySamples * 4;
+    sizeof(sample::Sample) * kDrainLatencySamples;
 
 StreamBufferHandle_t AudioState::sDrainBuffer;
 
@@ -151,33 +151,24 @@ void AudioState::react(const TogglePlayPause& ev) {
   }
 }
 
-void AudioState::react(const internal::DecoderOpened& ev) {
-  ESP_LOGI(kTag, "decoder opened %s", ev.track->uri.c_str());
+void AudioState::react(const internal::StreamStarted& ev) {
+  sCurrentFormat = ev.dst_format;
+  sIsResampling = ev.src_format != ev.dst_format;
   sNextTrack = ev.track;
   sNextTrackCueSamples = sCurrentSamples + kDrainLatencySamples;
-}
 
-void AudioState::react(const internal::DecoderClosed&) {
-  ESP_LOGI(kTag, "decoder closed");
-  // FIXME: only when we were playing the current track
-  sServices->track_queue().finish();
+  ESP_LOGI(kTag, "new stream %s %u ch @ %lu hz (resample=%i)",
+           ev.track->uri.c_str(), sCurrentFormat->num_channels,
+           sCurrentFormat->sample_rate, sIsResampling);
 }
 
-void AudioState::react(const internal::DecoderError&) {
-  ESP_LOGW(kTag, "decoder errored");
+void AudioState::react(const internal::StreamEnded&) {
+  ESP_LOGI(kTag, "stream ended");
   // FIXME: only when we were playing the current track
   sServices->track_queue().finish();
 }
 
-void AudioState::react(const internal::ConverterConfigurationChanged& ev) {
-  sCurrentFormat = ev.dst_format;
-  sIsResampling = ev.src_format != ev.dst_format;
-  ESP_LOGI(kTag, "output format now %u ch @ %lu hz (resample=%i)",
-           sCurrentFormat->num_channels, sCurrentFormat->sample_rate,
-           sIsResampling);
-}
-
-void AudioState::react(const internal::ConverterProgress& ev) {
+void AudioState::react(const internal::StreamUpdate& ev) {
   ESP_LOGI(kTag, "sample converter sunk %lu samples", ev.samples_sunk);
   sCurrentSamples += ev.samples_sunk;
 
diff --git a/src/audio/include/audio_converter.hpp b/src/audio/include/audio_converter.hpp
index dcd068b5..232b5d8e 100644
--- a/src/audio/include/audio_converter.hpp
+++ b/src/audio/include/audio_converter.hpp
@@ -10,6 +10,7 @@
 #include <cstdint>
 #include <memory>
 
+#include "audio_events.hpp"
 #include "audio_sink.hpp"
 #include "audio_source.hpp"
 #include "codec.hpp"
@@ -31,20 +32,23 @@ class SampleConverter {
 
   auto SetOutput(std::shared_ptr<IAudioOutput>) -> void;
 
-  auto ConvertSamples(cpp::span<sample::Sample>,
-                      const IAudioOutput::Format& format,
-                      bool is_eos) -> void;
+  auto beginStream(std::shared_ptr<TrackInfo>) -> void;
+  auto continueStream(cpp::span<sample::Sample>) -> void;
+  auto endStream() -> void;
 
  private:
   auto Main() -> void;
 
-  auto SetTargetFormat(const IAudioOutput::Format& format) -> void;
-  auto HandleSamples(cpp::span<sample::Sample>, bool) -> size_t;
+  auto handleBeginStream(std::shared_ptr<TrackInfo>) -> void;
+  auto handleContinueStream(size_t samples_available) -> void;
+  auto handleEndStream() -> void;
 
-  auto SendToSink(cpp::span<sample::Sample>) -> void;
+  auto handleSamples(cpp::span<sample::Sample>) -> size_t;
+
+  auto sendToSink(cpp::span<sample::Sample>) -> void;
 
   struct Args {
-    IAudioOutput::Format format;
+    std::shared_ptr<TrackInfo>* track;
     size_t samples_available;
     bool is_end_of_stream;
   };
diff --git a/src/audio/include/audio_events.hpp b/src/audio/include/audio_events.hpp
index 9af30467..b8a0dba6 100644
--- a/src/audio/include/audio_events.hpp
+++ b/src/audio/include/audio_events.hpp
@@ -51,6 +51,8 @@ struct TrackInfo {
 
   /* The encoded format of the this track. */
   codecs::StreamType encoding;
+
+  IAudioOutput::Format format;
 };
 
 /*
@@ -136,23 +138,18 @@ struct OutputModeChanged : tinyfsm::Event {};
 
 namespace internal {
 
-struct DecoderOpened : tinyfsm::Event {
+struct StreamStarted : tinyfsm::Event {
   std::shared_ptr<TrackInfo> track;
-};
-
-struct DecoderClosed : tinyfsm::Event {};
-
-struct DecoderError : tinyfsm::Event {};
-
-struct ConverterConfigurationChanged : tinyfsm::Event {
   IAudioOutput::Format src_format;
   IAudioOutput::Format dst_format;
 };
 
-struct ConverterProgress : tinyfsm::Event {
+struct StreamUpdate : tinyfsm::Event {
   uint32_t samples_sunk;
 };
 
+struct StreamEnded : tinyfsm::Event {};
+
 }  // namespace internal
 
 }  // namespace audio
diff --git a/src/audio/include/audio_fsm.hpp b/src/audio/include/audio_fsm.hpp
index 62bb4786..c00813ac 100644
--- a/src/audio/include/audio_fsm.hpp
+++ b/src/audio/include/audio_fsm.hpp
@@ -46,12 +46,9 @@ class AudioState : public tinyfsm::Fsm<AudioState> {
   void react(const SetTrack&);
   void react(const TogglePlayPause&);
 
-  void react(const internal::DecoderOpened&);
-  void react(const internal::DecoderClosed&);
-  void react(const internal::DecoderError&);
-
-  void react(const internal::ConverterConfigurationChanged&);
-  void react(const internal::ConverterProgress&);
+  void react(const internal::StreamStarted&);
+  void react(const internal::StreamUpdate&);
+  void react(const internal::StreamEnded&);
 
   void react(const StepUpVolume&);
   void react(const StepDownVolume&);
diff --git a/src/ui/ui_fsm.cpp b/src/ui/ui_fsm.cpp
index 42c6a99c..acc1bf10 100644
--- a/src/ui/ui_fsm.cpp
+++ b/src/ui/ui_fsm.cpp
@@ -392,7 +392,11 @@ void UiState::react(const audio::QueueUpdate&) {
 }
 
 void UiState::react(const audio::PlaybackUpdate& ev) {
-  sPlaybackTrack.Update(*ev.current_track);
+  if (ev.current_track) {
+    sPlaybackTrack.Update(*ev.current_track);
+  } else {
+    sPlaybackTrack.Update(std::monostate{});
+  }
   sPlaybackPlaying.Update(!ev.paused);
   sPlaybackPosition.Update(static_cast<int>(ev.track_position.value_or(0)));
 }