fix build issues with new pipeline

7 files changed, 130 insertions, 6 deletions

diff --git a/src/audio/CMakeLists.txt b/src/audio/CMakeLists.txt
index cf58f00c..4986a7da 100644
--- a/src/audio/CMakeLists.txt
+++ b/src/audio/CMakeLists.txt
@@ -1,7 +1,7 @@
 idf_component_register(
   SRCS "audio_decoder.cpp" "audio_task.cpp" "chunk.cpp" "fatfs_audio_input.cpp"
   "stream_info.cpp" "stream_message.cpp" "i2s_audio_output.cpp"
-  "stream_buffer.cpp" "audio_playback.cpp"
+  "stream_buffer.cpp" "audio_playback.cpp" "audio_element_handle.cpp"
   INCLUDE_DIRS "include"
   REQUIRES "codecs" "drivers" "cbor" "result" "tasks" "span")
 
diff --git a/src/audio/audio_decoder.cpp b/src/audio/audio_decoder.cpp
index 88ddc323..0b3d9878 100644
--- a/src/audio/audio_decoder.cpp
+++ b/src/audio/audio_decoder.cpp
@@ -58,7 +58,7 @@ auto AudioDecoder::ProcessChunk(const cpp::span<std::byte>& chunk)
   bool needs_more_input = false;
   std::optional<codecs::ICodec::ProcessingError> error = std::nullopt;
   while (1) {
-    ChunkWriteResult res = chunk_writer_.WriteChunkToStream(
+    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
diff --git a/src/audio/audio_element_handle.cpp b/src/audio/audio_element_handle.cpp
new file mode 100644
index 00000000..4b746db3
--- /dev/null
+++ b/src/audio/audio_element_handle.cpp
@@ -0,0 +1,79 @@
+#include "audio_element_handle.hpp"
+#include "audio_element.hpp"
+#include "freertos/projdefs.h"
+
+namespace audio {
+
+AudioElementHandle::AudioElementHandle(std::unique_ptr<TaskHandle_t> task,
+                                       std::shared_ptr<IAudioElement> element)
+    : task_(std::move(task)), element_(std::move(element)) {}
+
+AudioElementHandle::~AudioElementHandle() {
+  Quit();
+}
+
+auto AudioElementHandle::CurrentState() -> ElementState {
+  return element_->ElementState();
+}
+
+auto AudioElementHandle::PlayPause(enum PlayPause state) -> void {
+  ElementState s = CurrentState();
+  if (state == PLAY && s == STATE_PAUSE) {
+    // Ensure we actually finished any previous pause command.
+    // TODO: really?
+    PauseSync();
+    SetStateAndWakeUp(STATE_RUN);
+    return;
+  }
+  if (state == PAUSE && s == STATE_RUN) {
+    element_->ElementState(STATE_PAUSE);
+    SetStateAndWakeUp(STATE_PAUSE);
+    return;
+  }
+}
+
+auto AudioElementHandle::Quit() -> void {
+  SetStateAndWakeUp(STATE_QUIT);
+}
+
+auto AudioElementHandle::PauseSync() -> void {
+  PlayPause(PAUSE);
+  MonitorUtilState(eSuspended);
+}
+
+auto AudioElementHandle::QuitSync() -> void {
+  Quit();
+  MonitorUtilState(eDeleted);
+}
+
+auto AudioElementHandle::MonitorUtilState(eTaskState desired) -> void {
+  while (eTaskGetState(task_.get()) != desired) {
+    WakeUpTask();
+    vTaskDelay(pdMS_TO_TICKS(1));
+  }
+}
+
+auto AudioElementHandle::SetStateAndWakeUp(ElementState state) -> void {
+  element_->ElementState(state);
+  WakeUpTask();
+}
+
+auto AudioElementHandle::WakeUpTask() -> void {
+  // TODO: various races where the task isn't blocked yet, but there is a block
+  // between now and its next element state check. Also think about chunk blocks
+  // nested in element bodies.
+  // Maybe we need a big mutex or semaphore somewhere in here.
+  switch (eTaskGetState(task_.get())) {
+    case eBlocked:
+      // TODO: when is this safe?
+      xTaskAbortDelay(task_.get());
+      break;
+    case eSuspended:
+      vTaskResume(task_.get());
+      break;
+    default:
+      return;
+  }
+}
+
+}  // namespace audio
diff --git a/src/audio/fatfs_audio_input.cpp b/src/audio/fatfs_audio_input.cpp
index bc5be42a..3e501154 100644
--- a/src/audio/fatfs_audio_input.cpp
+++ b/src/audio/fatfs_audio_input.cpp
@@ -9,6 +9,7 @@
 
 #include "audio_element.hpp"
 #include "chunk.hpp"
+#include "stream_buffer.hpp"
 #include "stream_message.hpp"
 
 static const char* kTag = "SRC";
@@ -29,7 +30,10 @@ FatfsAudioInput::FatfsAudioInput(std::shared_ptr<drivers::SdStorage> storage)
       file_buffer_read_pos_(file_buffer_.begin()),
       file_buffer_write_pos_(file_buffer_.begin()),
       current_file_(),
-      is_file_open_(false) {}
+      is_file_open_(false),
+      chunk_writer_(nullptr) {
+  // TODO: create our chunk writer whenever the output buffer changes.
+}
 
 FatfsAudioInput::~FatfsAudioInput() {
   free(raw_file_buffer_);
@@ -127,7 +131,7 @@ auto FatfsAudioInput::ProcessIdle() -> cpp::result<void, AudioProcessingError> {
 
   // Now stream data into the output buffer until it's full.
   while (1) {
-    ChunkWriteResult result = chunk_writer_.WriteChunkToStream(
+    ChunkWriteResult result = chunk_writer_->WriteChunkToStream(
         [&](cpp::span<std::byte> d) { return SendChunk(d); }, kServiceInterval);
 
     switch (result) {
diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index 9c0626db..0a2df76d 100644
--- a/src/audio/include/audio_decoder.hpp
+++ b/src/audio/include/audio_decoder.hpp
@@ -43,7 +43,7 @@ class AudioDecoder : public IAudioElement {
   std::unique_ptr<codecs::ICodec> current_codec_;
   std::optional<StreamInfo> stream_info_;
 
-  ChunkWriter chunk_writer_;
+  std::unique_ptr<ChunkWriter> chunk_writer_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_element_handle.hpp b/src/audio/include/audio_element_handle.hpp
new file mode 100644
index 00000000..adb26baa
--- /dev/null
+++ b/src/audio/include/audio_element_handle.hpp
@@ -0,0 +1,41 @@
+#pragma once
+
+#include <memory>
+#include "audio_element.hpp"
+
+namespace audio {
+
+class AudioElementHandle {
+ public:
+  AudioElementHandle(std::unique_ptr<TaskHandle_t> task,
+                     std::shared_ptr<IAudioElement> element);
+  ~AudioElementHandle();
+
+  auto CurrentState() -> ElementState;
+
+  // TODO: think about this contract. Would it ever make sense to pause and
+  // then walk away? Things could keep running for a whole loop if data comes
+  // through, so probably not?
+  enum PlayPause {
+    PLAY,
+    PAUSE,
+  };
+  auto PlayPause(PlayPause state) -> void;
+  auto Quit() -> void;
+
+  auto PauseSync() -> void;
+  auto QuitSync() -> void;
+
+  AudioElementHandle(const AudioElementHandle&) = delete;
+  AudioElementHandle& operator=(const AudioElementHandle&) = delete;
+
+ private:
+  std::unique_ptr<TaskHandle_t> task_;
+  std::shared_ptr<IAudioElement> element_;
+
+  auto MonitorUtilState(eTaskState desired) -> void;
+  auto SetStateAndWakeUp(ElementState state) -> void;
+  auto WakeUpTask() -> void;
+};
+
+}  // namespace audio
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index 040b2b54..f3704f1d 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/src/audio/include/fatfs_audio_input.hpp
@@ -47,7 +47,7 @@ class FatfsAudioInput : public IAudioElement {
   FIL current_file_;
   bool is_file_open_;
 
-  ChunkWriter chunk_writer_;
+  std::unique_ptr<ChunkWriter> chunk_writer_;
 };
 
 }  // namespace audio