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

1 files changed, 123 insertions, 86 deletions

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;