Mostly done pipeline arch. Now onto cleanup and building.

src/audio/CMakeLists.txt

@@ -1,7 +1,7 @@
 idf_component_register(
-  SRCS "audio_decoder.cpp" "audio_task.cpp" "fatfs_audio_input.cpp" "chunk.cpp"
+  SRCS "audio_decoder.cpp" "audio_task.cpp" "chunk.cpp" "fatfs_audio_input.cpp"
-  "i2s_audio_output.cpp" "stream_info.cpp"
+  "stream_info.cpp"
   INCLUDE_DIRS "include"
-  REQUIRES "codecs" "drivers" "cbor")
+  REQUIRES "codecs" "drivers" "cbor" "result" "tasks")
 
 target_compile_options(${COMPONENT_LIB} PRIVATE ${EXTRA_WARNINGS})

src/audio/audio_decoder.cpp

@@ -1,107 +1,87 @@
 #include "audio_decoder.hpp"
+#include <string.h>
 #include <cstddef>
 #include <cstdint>
-#include <string.h>
+#include "chunk.hpp"
 #include "esp_heap_caps.h"
+#include "freertos/portmacro.h"
 #include "include/audio_element.hpp"
 #include "include/fatfs_audio_input.hpp"
 
 namespace audio {
 
-  // TODO: could this be larger? depends on the codecs i guess
+AudioDecoder::AudioDecoder()
-  static const std::size_t kWorkingBufferSize = kMaxFrameSize;
+    : IAudioElement(),
-
+      chunk_buffer_(heap_caps_malloc(kMaxChunkSize, MALLOC_CAP_SPIRAM)),
-  AudioDecoder::AudioDecoder() {
+      stream_info_({}) {}
-    working_buffer_ = heap_caps_malloc(kWorkingBufferSize, MALLOC_CAP_SPIRAM);
-  }
 
 AudioDecoder::~AudioDecoder() {
-    free(working_buffer_);
+  free(chunk_buffer_);
-  }
-
-  auto AudioDecoder::InputBuffer() -> StreamBufferHandle_t {
-    return input_buffer_;
 }
 
-  auto AudioDecoder::OutputBuffer() -> StreamBufferHandle_t {
+auto AudioDecoder::SetInputBuffer(StreamBufferHandle_t* buffer) -> void {
-    return output_buffer_;
-  }
-
-  auto AudioDecoder::SetInputBuffer(StreamBufferHandle_t buffer) -> void {
   input_buffer_ = buffer;
 }
 
-  auto AudioDecoder::SetOutputBuffer(StreamBufferHandle_t buffer) -> void {
+auto AudioDecoder::SetOutputBuffer(StreamBufferHandle_t* buffer) -> void {
   output_buffer_ = buffer;
 }
 
-  auto AudioDecoder::ProcessElementCommand(void* command) -> ProcessResult {
+auto AudioDecoder::ProcessStreamInfo(StreamInfo&& info)
-    FatfsAudioInput::OutputCommand *real = std::reinterpret_cast<FatfsAudioInput::OutputCommand*>(command);
+    -> cpp::result<void, StreamError> {
+  stream_info_ = info;
 
-    if (current_codec_->CanHandleExtension(real->extension)) {
+  // Reuse the existing codec if we can. This will help with gapless playback,
-      // TODO: Do we need to reset the codec?
+  // since we can potentially just continue to decode as we were before,
-      delete real;
+  // without any setup overhead.
-      return OK;
+  if (current_codec_->CanHandleFile(info.path)) {
+    current_codec_->ResetForNewStream();
+    return {};
   }
 
-    auto result = codecs::CreateCodecForExtension(real->extension);
+  auto result = codecs::CreateCodecForFile(info.path);
-    // TODO: handle error case
   if (result.has_value()) {
-      current_codec_ = result.value();
+    current_codec_ = std::move(result.value());
+  } else {
+    return cpp::fail(UNSUPPORTED_STREAM);
   }
 
-    delete real;
+  return {};
-    return OK;
 }
 
-  auto AudioDecoder::SkipElementCommand(void* command) -> void {
+auto AudioDecoder::ProcessChunk(uint8_t* data, std::size_t length)
-    FatfsAudioInput::OutputCommand *real = std::reinterpret_cast<FatfsAudioInput::OutputCommand*>(command);
+    -> cpp::result<size_t, StreamError> {
-    delete real;
-  }
-
-  auto AudioDecoder::ProcessData(uint8_t* data, uint16_t length) -> ProcessResult {
   if (current_codec_ == nullptr) {
-      // TODO: signal this
+    // Should never happen, but fail explicitly anyway.
-      return OK;
+    return cpp::fail(UNSUPPORTED_STREAM);
   }
 
-    while (true) {
+  current_codec_->SetInput(data, length);
-      auto result = current_codec_->Process(data, length, working_buffer_, kWorkingBufferSize);
+  cpp::result<size_t, codecs::ICodec::ProcessingError> result;
-
+  WriteChunksToStream(
+      output_buffer_, working_buffer_, kWorkingBufferSize,
+      [&](uint8_t* buf, size_t len) {
+        result = current_codec_->Process(data, length, buf, len);
         if (result.has_error()) {
-        // TODO: handle i guess
+          // End our output stream immediately if the codec barfed.
-        return ERROR;
+          return 0;
         }
-      ICodec::Result process_res = result.value();
+        return result.value();
+      },
+      // This element doesn't support any kind of out of band commands, so we
+      // can just suspend the whole task if the output buffer fills up.
+      portMAX_DELAY);
 
-      if (process_res.flush_output) {
+  if (result.has_error()) {
-        xStreamBufferSend(&output_buffer_, working_buffer_, process_res.output_written, kMaxWaitTicks);
+    return cpp::fail(IO_ERROR);
-      }
-
-      if (process_res.need_more_input) {
-        // TODO: wtf do we do about the leftover bytes?
-        return OK;
-      }
-    }
-
-    return OK;
   }
 
-  auto AudioDecoder::ProcessIdle() -> ProcessResult {
+  return current_codec_->GetOutputProcessed();
-    // Not used.
-    return OK;
 }
 
-  auto AudioDecoder::Pause() -> void {
+auto AudioDecoder::ProcessIdle() -> cpp::result<void, StreamError> {
-    // TODO.
+  // Not used; we delay forever when waiting on IO.
-  }
+  return {};
-  auto AudioDecoder::IsPaused() -> bool {
-    // TODO.
 }
 
-  auto AudioDecoder::Resume() -> void {
-    // TODO.
-  }
-
-
 }  // namespace audio

src/audio/audio_task.cpp

@@ -4,6 +4,8 @@
 
 #include <cstdint>
 
+#include "cbor_decoder.hpp"
+#include "chunk.hpp"
 #include "esp-idf/components/cbor/tinycbor/src/cbor.h"
 #include "esp_heap_caps.h"
 #include "freertos/portmacro.h"
@@ -13,132 +15,83 @@
 #include "audio_element.hpp"
 #include "include/audio_element.hpp"
 #include "stream_message.hpp"
+#include "tasks.hpp"
 
 namespace audio {
 
 static const TickType_t kCommandWaitTicks = 1;
 static const TickType_t kIdleTaskDelay = 1;
+static const size_t kChunkBufferSize = kMaxChunkSize * 1.5;
 
-void audio_task(void* args) {
+auto StartAudioTask(const std::string& name,
+                    std::shared_ptr<IAudioElement>& element) -> void {
+  AudioTaskArgs* args = new AudioTaskArgs(element);
+  xTaskCreate(&AudioTaskMain, name.c_str(), element->StackSizeBytes(), args,
+              kTaskPriorityAudio, NULL);
+}
+
+void AudioTaskMain(void* args) {
   AudioTaskArgs* real_args = reinterpret_cast<AudioTaskArgs*>(args);
-  std::shared_ptr<IAudioElement> element = real_args->element;
+  std::shared_ptr<IAudioElement> element = std::move(real_args->element);
   delete real_args;
 
-  MessageBufferHandle_t *stream = element->InputBuffer();
+  ChunkReader chunk_reader = ChunkReader(element->InputBuffer());
-
-  uint8_t* message_buffer =
-      (uint8_t*)heap_caps_malloc(kFrameSize, MALLOC_CAP_SPIRAM);
 
   while (1) {
-    BaseType_t rtos_res;
+    cpp::result<size_t, IAudioElement::StreamError> process_res;
-    IAudioElement::ProcessResult result;
+
-
+    // If this element has an input stream, then our top priority is processing
-
+    // any chunks from it. Try doing this first, then fall back to the other
-    size_t message_size = 0;
+    // cases.
-    if (message_buffer != nullptr) {
+    bool has_received_message = false;
-      // TODO: tune delay.
+    if (stream != nullptr) {
-      message_size = xMessageBufferReceive(stream, &message_buffer, kFrameSize, portMAX_DELAY);
+      EncodeReadResult chunk_res = chunk_reader.ReadChunkFromStream(
-    }
+          [&](uint8_t* data, std::size_t length) -> std::optional<size_t> {
-
+            process_res = element->ProcessChunk(data, length);
-    if (message_size == 0) {
+            if (process_res.has_value()) {
-      element->ProcessIdle();
+              return process_res.value();
-      continue;
+            } else {
+              return {};
             }
+          },
+          element->IdleTimeout());
 
-    // We got a valid message. Check what kind it is so that we know how to
+      if (chunk_res == CHUNK_PROCESSING_ERROR ||
-    // process it.
+          chunk_res == CHUNK_DECODING_ERROR) {
-    CborParser parser;
-    CborValue value;
-    cbor_parser_init(message_buffer, message_size, &parser, &value);
-
-    MessageType message_type;
-    if (!cbor_value_is_integer(&value) || !cbor_value_get_integer(&value, &message_type)) {
-      // We weren't able to parse the message type. This is really bad, so just
-      // abort.
         break;  // TODO.
+      } else if (chunk_res == CHUNK_STREAM_ENDED) {
+        has_received_message = true;
       }
-
-    if (message_type == STREAM_INFO) {
-      errs = StreamInfo::Create(message_buffer, message_size).map(element->ProcessStreamInfo);
-      if (errs.has_error) {
-        // TODO;
     }
-    } else if (message_type == CHUNK_HEADER) {
-    } else {
-      // TODO.
-    }
-
-    cbor_value_
-    if (!xQueueReceive(commands, &command, wait_time)) {
-      if (bytes_in_stream > 0) {
-        size_t read_length = std::min(kMaxFrameSize - leftover_data, bytes_in_stream);
-        xStreamBufferReceive(stream, &frame_buffer + leftover_data, read_length, 0);
 
-        uint8_t *data_in = frame_buffer;
+    if (has_received_message) {
-        result = element->ProcessData(&data_in, read_length);
+      auto& [buffer, length] = chunk_reader.GetLastMessage();
-        if (result == IAudioElement::ERROR) {
+      auto decoder_res = cbor::ArrayDecoder::Create(buffer, length);
-          break;
+      if (decoder_res.has_error()) {
-        }
+        // TODO.
-
-        if (result == IAudioElement::LEFTOVER_DATA) {
-          leftover_data = frame_buffer + read_length - data_in;
-          memmove(frame_buffer, data_in, leftover_data);
-        } else {
-          leftover_data = 0;
-        }
-      } else {
-        result = element->ProcessIdle();
-        if (result == IAudioElement::ERROR) {
         break;
       }
-        if (result == IAudioElement::OUTPUT_FULL) {
+      auto decoder = decoder_res.value();
-          vTaskDelay(kIdleTaskDelay);
+      MessageType message_type = decoder->NextValue();
+      if (message_type == TYPE_STREAM_INFO) {
+        element->ProcessStreamInfo(StreamInfo(decoder->Iterator()););
       }
     }
-    } else {
-      if (command.type == IAudioElement::SEQUENCE_NUMBER) {
-        if (command.sequence_number > current_sequence_number) {
-          current_sequence_number = command.sequence_number;
-          bytes_in_stream = 0;
-        }
-      } else if (command.type == IAudioElement::READ) {
-        assert(command.read_size <= kFrameSize);
-        assert(stream != NULL);
 
-        if (command.sequence_number == current_sequence_number) {
+    // TODO: Do any out of band reading, such a a pause command, here.
-          bytes_in_stream += command.read_size;
+
-        } else {
+    // Chunk reading must have timed out, or we don't have an input stream.
-          // This data is for a different stream, so just discard it.
+    // Signal the element to do any of its idle tasks.
-          xStreamBufferReceive(stream, &frame_buffer, command.read_size, 0);
+    process_res = element->ProcessIdle();
-        }
+    if (process_res.has_error()) {
-      } else if (command.type == IAudioElement::ELEMENT) {
+      break;  // TODO.
-        assert(command.data != NULL);
-        if (command.sequence_number == current_sequence_number) {
-          if (bytes_in_stream > 0) {
-            // We're not ready to handle this yet, so put it back.
-            xQueueSendToFront(commands, &command, kMaxWaitTicks);
-          } else {
-            result = element->ProcessElementCommand(command.data);
-            if (result == IAudioElement::ERROR) {
-              break;
-            }
-            if (result == IAudioElement::OUTPUT_FULL) {
-              // TODO: what does this mean lol
-            }
-          }
-        } else {
-          element->SkipElementCommand(command.data);
-        }
-      } else if (command.type == IAudioElement::QUIT) {
-        break;
-      }
     }
   }
 
-  element = nullptr;
+  element.clear();
-  free(frame_buffer);
+  free(chunk_buffer_);
 
-  xTaskDelete(NULL);
+  vTaskDelete(NULL);
 }
 
 }  // namespace audio

src/audio/chunk.cpp

@@ -1,14 +1,21 @@
 #include "chunk.hpp"
 
-#include "cbor_encoder.hpp"
-#include "cbor_decoder.hpp"
 #include <string.h>
+#include <cstddef>
 #include <cstdint>
+#include "cbor_decoder.hpp"
+#include "cbor_encoder.hpp"
 #include "esp-idf/components/cbor/tinycbor/src/cbor.h"
 #include "stream_message.hpp"
 
 namespace audio {
 
+// TODO: tune.
+static const std::size_t kMaxChunkSize = 512;
+
+// TODO: tune
+static const std::size_t kWorkingBufferSize = kMaxChunkSize * 1.5;
+
 /*
  * The amount of space to allocate for the first chunk's header. After the first
  * chunk, we have a more concrete idea of the header's size and can allocate
@@ -17,14 +24,15 @@ namespace audio {
 // TODO: measure how big headers tend to be to pick a better value.
 static const size_t kInitialHeaderSize = 32;
 
-auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer, size_t working_buffer_length, std::function<size_t(uint8_t*,size_t)> callback, TickType_t max_wait) -> EncodeWriteResult {
+auto WriteChunksToStream(MessageBufferHandle_t* stream,
-
+                         uint8_t* working_buffer,
+                         size_t working_buffer_length,
+                         std::function<size_t(uint8_t*, size_t)> callback,
+                         TickType_t max_wait) -> EncodeWriteResult {
   size_t header_size = kInitialHeaderSize;
   while (1) {
     // First, ask the callback for some data to write.
-    size_t chunk_size =
+    size_t chunk_size = callback(working_buffer + header_size,
-      callback(
-          working_buffer + header_size,
                                  working_buffer_length - header_size);
 
     if (chunk_size == 0) {
@@ -33,7 +41,8 @@ auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer,
     }
 
     // Put together a header.
-    cbor::Encoder encoder(cbor::CONTAINER_ARRAY, 3, working_buffer, working_buffer_length);
+    cbor::Encoder encoder(cbor::CONTAINER_ARRAY, 3, working_buffer,
+                          working_buffer_length);
     encoder.WriteUnsigned(TYPE_CHUNK_HEADER);
     encoder.WriteUnsigned(header_size);
     encoder.WriteUnsigned(chunk_size);
@@ -52,8 +61,7 @@ auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer,
 
     // Try to write to the buffer. Note the return type here will be either 0 or
     // header_size + chunk_size, as MessageBuffer doesn't allow partial writes.
-    size_t actual_write_size =
+    size_t actual_write_size = xMessageBufferSend(
-      xMessageBufferSend(
         *stream, working_buffer, header_size + chunk_size, max_wait);
 
     header_size = new_header_size;
@@ -67,20 +75,37 @@ auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer,
   }
 }
 
-auto ReadChunksFromStream(MessageBufferHandle_t *stream, uint8_t *working_buffer, size_t working_buffer_length, std::function<size_t(uint8_t*,size_t)> callback, TickType_t max_wait) -> EncodeReadResult {
+ChunkReader::ChunkReader(MessageBufferHandle_t* stream) : stream_(stream) {
-  // Spillover if the previous iteration did not consume all of the input.
+  working_buffer_ = heap_caps_malloc(kWorkingBufferSize, MALLOC_CAP_SPIRAM);
-  size_t leftover_bytes = 0;
+};
-  while (1) {
+
+ChunkReader::~ChunkReader() {
+  free(working_buffer_);
+}
+
+auto ChunkReader::Reset() -> void {
+  leftover_bytes_ = 0;
+  last_message_size_ = 0;
+}
+
+auto ChunkReader::GetLastMessage() -> std::pair<uint8_t*, size_t> {
+  return std::make_pair(working_buffer_ + leftover_bytes_, last_message_size_);
+}
+
+auto ChunkReader::ReadChunkFromStream(
+    std::function<std::optional<size_t>(uint8_t*, size_t)> callback,
+    TickType_t max_wait) -> EncodeReadResult {
   // First, wait for a message to arrive over the buffer.
-    size_t read_size =
+  last_message_size_ =
-      xMessageBufferReceive(
+      xMessageBufferReceive(*stream_, working_buffer_ + leftover_bytes_,
-          *stream, working_buffer + leftover_bytes, working_buffer_length - leftover_bytes, max_wait);
+                            kWorkingBufferSize - leftover_bytes_, max_wait);
 
-    if (read_size == 0) {
+  if (last_message_size_ == 0) {
     return CHUNK_READ_TIMEOUT;
   }
 
-    auto decoder = cbor::MapDecoder::Create(working_buffer + leftover_bytes, read_size);
+  auto decoder = cbor::MapDecoder::Create(working_buffer_ + leftover_bytes_,
+                                          last_message_size_);
   if (decoder.has_error()) {
     // Weird; this implies someone is shoving invalid data into the buffer.
     return CHUNK_DECODING_ERROR;
@@ -88,9 +113,8 @@ auto ReadChunksFromStream(MessageBufferHandle_t *stream, uint8_t *working_buffer
 
   MessageType type = decoder.value().ParseUnsigned().value_or(TYPE_UNKNOWN);
   if (type != TYPE_CHUNK_HEADER) {
-      // This message wasn't for us, so put it in a consistent place and let the
+    // This message wasn't for us, so let the caller handle it.
-      // caller handle it.
+    Reset();
-      memmove(working_buffer, working_buffer + leftover_bytes, read_size);
     return CHUNK_STREAM_ENDED;
   }
 
@@ -104,20 +128,24 @@ auto ReadChunksFromStream(MessageBufferHandle_t *stream, uint8_t *working_buffer
   // Now we need to stick the end of the last chunk (if it exists) onto the
   // front of the new chunk. Do it this way around bc we assume the old chunk
   // is shorter, and therefore faster to move.
-    uint8_t *combined_buffer = working_buffer + header_length - leftover_bytes;
+  uint8_t* combined_buffer = working_buffer_ + header_length - leftover_bytes_;
-    size_t combined_buffer_size = leftover_bytes + chunk_length;
+  size_t combined_buffer_size = leftover_bytes_ + chunk_length;
-    if (leftover_bytes > 0) {
+  if (leftover_bytes_ > 0) {
-      memmove(combined_buffer, working_buffer, leftover_bytes);
+    memmove(combined_buffer, working_buffer_, leftover_bytes_);
   }
 
   // Tell the callback about the new data.
-    size_t amount_processed = callback(combined_buffer, combined_buffer_size);
+  std::optional<size_t> amount_processed =
+      callback(combined_buffer, combined_buffer_size);
+  if (!amount_processed) {
+    return CHUNK_PROCESSING_ERROR;
+  }
 
   // Prepare for the next iteration.
-    leftover_bytes = combined_buffer_size - amount_processed;
+  leftover_bytes_ = combined_buffer_size - amount_processed.value();
-    if (leftover_bytes > 0) {
+  if (leftover_bytes_ > 0) {
-      memmove(working_buffer, combined_buffer + amount_processed, leftover_bytes);
+    memmove(working_buffer_, combined_buffer + amount_processed.value(),
-    }
+            leftover_bytes_);
   }
 }
 

src/audio/fatfs_audio_input.cpp

@@ -1,6 +1,8 @@
 #include "fatfs_audio_input.hpp<D-c>ccc
 #include <cstdint>
 #include <memory>
+#include "chunk.hpp"
+#include "fatfs_audio_input.hpp"
 
 #include "esp_heap_caps.h"
 
@@ -10,31 +12,32 @@
 
 namespace audio {
 
-static const TickType_t kMaxWaitTicks = portMAX_DELAY;
+static const TickType_t kServiceInterval = pdMS_TO_TICKS(50);
 
-// Large output buffer size, so that we can keep a get as much of the input file
+static const std::size_t kFileBufferSize = 1024 * 128;
-// into memory as soon as possible.
+static const std::size_t kMinFileReadSize = 1024 * 4;
-static constexpr std::size_t kOutputBufferSize = 1024 * 128;
+static const std::size_t kOutputBufferSize = 1024 * 4;
-static constexpr std::size_t kQueueItemSize = sizeof(IAudioElement::Command);
 
 FatfsAudioInput::FatfsAudioInput(std::shared_ptr<drivers::SdStorage> storage)
     : IAudioElement(), storage_(storage) {
-  working_buffer_ = heap_caps_malloc(kMaxFrameSize, MALLOC_CAP_SPIRAM);
+  file_buffer_ = heap_caps_malloc(kMaxFrameSize, MALLOC_CAP_SPIRAM);
+  file_buffer_read_pos_ = file_buffer_;
+  file_buffer_write_pos_ = file_buffer_;
+  chunk_buffer_ = heap_caps_malloc(kMaxChunkSize, MALLOC_CAP_SPIRAM);
 
   output_buffer_memory_ =
-      heap_caps_malloc(kOutputBufferSize + 1, MALLOC_CAP_SPIRAM);
+      heap_caps_malloc(kOutputBufferSize, MALLOC_CAP_SPIRAM);
-  output_buffer_ =
+  output_buffer_ = xMessageBufferCreateStatic(
-      xMessageBufferCreateStatic(kOutputBufferSize, output_buffer_memory_,
+      kOutputBufferSize, output_buffer_memory_, &output_buffer_metadata_);
-                                &output_buffer_metadata_);
 }
 
 FatfsAudioInput::~FatfsAudioInput() {
-  free(working_buffer_);
+  free(file_buffer_);
+  free(chunk_buffer_);
   vMessageBufferDelete(output_buffer_);
   free(output_buffer_memory_);
 }
 
-
 auto FatfsAudioInput::InputBuffer() -> MessageBufferHandle_t {
   return input_buffer_;
 }
@@ -43,80 +46,118 @@ auto FatfsAudioInput::OutputBuffer() -> MessageBufferHandle_t {
   return output_buffer_;
 }
 
-auto FatfsAudioInput::ProcessElementCommand(void* command) -> ProcessResult {
+auto FatfsAudioInput::ProcessStreamInfo(StreamInfo& info)
-  InputCommand *real = std::reinterpret_cast<InputCommand*>(command);
+    -> cpp::result<void, StreamError> {
-
-  if (uxQueueSpacesAvailable(output_queue_) < 2) {
-    return OUTPUT_FULL;
-  }
-
   if (is_file_open_) {
     f_close(&current_file_);
+    is_file_open_ = false;
   }
 
-  FRESULT res = f_open(&current_file_, real->filename.c_str(), FA_READ);
+  FRESULT res = f_open(&current_file_, info.path.c_str(), FA_READ);
   if (res != FR_OK) {
-    delete real;
+    return cpp::fail(IO_ERROR);
-    return ERROR;
-  }
-
-  if (real->seek_to && f_lseek(&current_file_, real->seek_to) {
-      return ERROR;
   }
 
   is_file_open_ = true;
 
-  if (real->interrupt) {
+  // TODO: pass on stream info.
-    Command sequence_update;
-    sequence_update.type = SEQUENCE_NUMBER;
-    sequence_update.sequence_number = current_sequence_++;
-    xQueueSendToFront(output_queue_, &sequence_update, kMaxWaitTicks);
-  }
-
-  OutputCommand *data = new OutputCommand;
-  data->extension = "mp3";
-  Command file_info;
-  file_info.type = ELEMENT;
-  file_info.sequence_number = current_sequence_;
-  file_info.data = &data;
-  xQueueSendToBack(output_queue_, &file_info, kMaxWaitTicks);
 
-  delete real;
+  return {};
-  return OK;
 }
 
-auto FatfsAudioInput::SkipElementCommand(void* command) -> void {
+auto FatfsAudioInput::ProcessChunk(uint8_t* data, std::size_t length)
-  InputCommand *real = std::reinterpret_cast<input_key_service_add_key*>(command);
+    -> cpp::result<void, StreamError> {
-  delete real;
+  // TODO.
+  return {};
 }
 
-auto FatfsAudioInput::ProcessData(uint8_t* data, uint16_t length) -> void {
+static auto GetRingBufferDistance() -> size_t {
-  // Not used, since we have no input stream.
+  if (file_buffer_read_pos_ == file_buffer_write_pos_) {
+    return 0;
   }
-
+  if (file_buffer_read_pos_ < file_buffer_write_pos_) {
-auto FatfsAudioInput::ProcessIdle() -> ProcessResult {
+    return file_buffer_write_pos_ - file_buffer_read_pos_;
-  if (!is_file_open_) {
+  }
-    return OK;
+  return
+      // Read position to end of buffer.
+      (file_buffer_ + kFileBufferSize - file_buffer_read_pos_)
+      // Start of buffer to write position.
+      + (file_buffer_write_pos_ - file_buffer_)
 }
 
-  if (xStreamBufferSpacesAvailable(output_buffer) < kMaxFrameSize) {
+virtual auto FatfsAudioInput::ProcessIdle() -> cpp::result<void, StreamError> {
-    return OUTPUT_FULL;
+  // First, see if we're able to fill up the input buffer with any more of the
+  // file's contents.
+  if (is_file_open_) {
+    size_t ringbuf_distance = GetRingBufferDistance();
+    if (kFileBufferSize - ringbuf_distance > kMinFileReadSize) {
+      size_t read_size;
+      if (file_buffer_write_pos_ < file_buffer_read_pos_) {
+        // Don't worry about the start of buffer -> read pos size; we can get to
+        // it next iteration.
+        read_size = file_buffer_read_pos_ - file_buffer_write_pos_;
+      } else {
+        read_size = file_buffer_ - file_buffer_write_pos_;
       }
 
       UINT bytes_read = 0;
-  FRESULT result = f_read(&current_file_, working_buffer_, kMaxFrameSize, &bytes_read);
+      FRESULT result = f_read(&current_file_, file_buffer_write_pos_, read_size,
+                              &bytes_read);
       if (!FR_OK) {
-    return ERROR;
+        return ERROR;  // TODO;
       }
 
-  xStreamBufferSend(&output_buffer_, working_buffer_, bytes_read, kMaxWaitTicks);
-
       if (f_eof(&current_file_)) {
         f_close(&current_file_);
         is_file_open_ = false;
+
+        // TODO: open the next file?
       }
 
-  return OK;
+      file_buffer_write_pos_ += bytes_read;
+      if (file_buffer_write_pos_ == file_buffer_ + kFileBufferSize) {
+        file_buffer_write_pos_ = file_buffer_;
+      }
+    }
+  }
+
+  // Now stream data into the output buffer until it's full.
+  pending_read_pos_ = nullptr;
+  EncodeWriteResult result =
+      WriteChunksToStream(&output_buffer_, chunk_buffer_, kMaxChunkSize,
+                          &SendChunk, kServiceInterval);
+
+  switch (result) {
+    case CHUNK_WRITE_TIMEOUT:
+    case CHUNK_OUT_OF_DATA:
+      return;  // TODO.
+    default:
+      return;  // TODO.
+  }
+}
+
+auto FatfsAudioInput::SendChunk(uint8_t* buffer, size_t size) -> size_t {
+  if (pending_read_pos_ != nullptr) {
+    file_buffer_read_pos_ = pending_read_pos_;
+  }
+
+  if (file_buffer_read_pos_ == file_buffer_write_pos_) {
+    return 0;
+  }
+  std::size_t write_size;
+  if (file_buffer_read_pos_ > file_buffer_write_pos_) {
+    write_size = file_buffer_ + kFileBufferSize - file_buffer_read_pos_;
+  } else {
+    write_size = file_buffer_write_pos_ - file_buffer_read_pos_;
+  }
+  write_size = std::min(write_size, size);
+  memcpy(buffer, file_buffer_read_pos_, write_size);
+
+  pending_read_pos_ = file_buffer_read_pos_ + write_size;
+  if (pending_read_pos_ == file_buffer_ + kFileBufferSize) {
+    pending_read_pos_ = file_buffer_;
+  }
+  return write_size;
 }
 
 }  // namespace audio

src/audio/include/audio_decoder.hpp

@@ -1,27 +1,34 @@
 #pragma once
 
 #include <cstddef>
-#include "audio_element.hpp"
+
 #include "ff.h"
+
+#include "audio_element.hpp"
 #include "codec.hpp"
 
 namespace audio {
 
+/*
+ * An audio element that accepts various kinds of encoded audio streams as
+ * input, and converts them to uncompressed PCM output.
+ */
 class AudioDecoder : public IAudioElement {
  public:
   AudioDecoder();
   ~AudioDecoder();
 
-  auto SetInputBuffer(StreamBufferHandle_t) -> void;
+  auto SetInputBuffer(StreamBufferHandle_t*) -> void;
-  auto SetOutputBuffer(StreamBufferHandle_t) -> void;
+  auto SetOutputBuffer(StreamBufferHandle_t*) -> void;
+
+  AudioDecoder(const AudioDecoder&) = delete;
+  AudioDecoder& operator=(const AudioDecoder&) = delete;
 
  private:
   std::unique_ptr<codecs::ICodec> current_codec_;
+  std::optional<StreamInfo> stream_info_;
 
-  uint8_t *working_buffer_;
+  uint8_t* chunk_buffer_;
-
-  StreamBufferHandle_t input_buffer_;
-  StreamBufferHandle_t output_buffer_;
 };
 
 }  // namespace audio

src/audio/include/audio_element.hpp

@@ -3,30 +3,83 @@
 #include <stdint.h>
 #include <cstdint>
 #include "freertos/portmacro.h"
-#include "types.hpp"
 #include "result.hpp"
+#include "types.hpp"
 
 namespace audio {
 
-extern const std::size_t kMaxFrameSize;
+/*
-
+ * One indepedentent part of an audio pipeline. Each element has an input and
+ * output message stream, and is responsible for taking data from the input
+ * stream, applying some kind of transformation to it, then sending the result
+ * out via the output stream. All communication with an element should be done
+ * over these streams, as an element's methods are only safe to call from the
+ * task that owns it.
+ *
+ * Each element implementation will have its input stream automatically parsed
+ * by its owning task, and its various Process* methods will be invoked
+ * accordingly. Element implementations are responsible for managing their own
+ * writes to their output streams.
+ */
 class IAudioElement {
  public:
+  IAudioElement() : input_buffer_(nullptr), output_buffer_(nullptr) {}
   virtual ~IAudioElement();
 
+  /*
+   * Returns the stack size in bytes that this element requires. This should
+   * be tuned according to the observed stack size of each element, as different
+   * elements have fairly different stack requirements.
+   */
+  virtual auto StackSizeBytes() -> std::size_t { return 2048; };
+
+  /*
+   * How long to wait for new data on the input stream before triggering a call
+   * to ProcessIdle(). If this is portMAX_DELAY (the default), then ProcessIdle
+   * will never be called.
+   */
   virtual auto IdleTimeout() -> TickType_t { return portMAX_DELAY; }
 
-  virtual auto InputBuffer() -> MessageBufferHandle_t* = 0;
+  /* Returns this element's input buffer. */
+  auto InputBuffer() -> MessageBufferHandle_t* { return input_buffer_; }
 
-  virtual auto OutputBuffer() -> MessageBufferHandle_t* = 0;
+  /* Returns this element's output buffer. */
+  auto OutputBuffer() -> MessageBufferHandle_t* { return output_buffer_; }
 
+  /* Errors that may be returned by any of the Process* methods. */
   enum StreamError {
-    BAD_FORMAT
+    // Indicates that this element is unable to handle the upcoming chunks.
+    UNSUPPORTED_STREAM,
+    // Indicates an error with reading or writing stream data.
+    IO_ERROR,
   };
 
-  virtual auto ProcessStreamInfo(StreamInfo &info) -> cpp::result<void, StreamError> = 0;
+  /*
-  virtual auto ProcessChunk(uint8_t* data, std::size_t length) -> cpp::result<void, StreamError> = 0;
+   * Called when a StreamInfo message is received. Used to configure this
+   * element in preperation for incoming chunks.
+   */
+  virtual auto ProcessStreamInfo(StreamInfo&& info)
+      -> cpp::result<void, StreamError> = 0;
+
+  /*
+   * Called when a ChunkHeader message is received. Includes the data associated
+   * with this chunk of stream data. This method should return the number of
+   * bytes in this chunk that were actually used; leftover bytes will be
+   * prepended to the next call.
+   */
+  virtual auto ProcessChunk(uint8_t* data, std::size_t length)
+      -> cpp::result<size_t, StreamError> = 0;
+
+  /*
+   * Called when there has been no data received over the input buffer for some
+   * time. This could be used to synthesize output, or to save memory by
+   * releasing unused resources.
+   */
   virtual auto ProcessIdle() -> cpp::result<void, StreamError> = 0;
+
+ protected:
+  StreamBufferHandle_t* input_buffer_;
+  StreamBufferHandle_t* output_buffer_;
 };
 
 }  // namespace audio

src/audio/include/audio_task.hpp

@@ -10,6 +10,8 @@ struct AudioTaskArgs {
   std::shared_ptr<IAudioElement>& element;
 };
 
-void audio_task(void* args);
+auto StartAudioTask(std::shared_ptr<IAudioElement>& element) -> void;
+
+void AudioTaskMain(void* args);
 
 }  // namespace audio

src/audio/include/chunk.hpp

@@ -10,6 +10,8 @@
 
 namespace audio {
 
+extern const std::size_t kMaxChunkSize;
+
 enum ChunkWriteResult {
   // Returned when the callback does not write any data.
   CHUNK_OUT_OF_DATA,
@@ -29,7 +31,20 @@ enum ChunkWriteResult {
  * number of bytes it wrote. Return a value of 0 to indicate that there is no
  * more input to read.
  */
-auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer, size_t working_buffer_length, std::function<size_t(uint8_t*,size_t)> callback, TickType_t max_wait) -> EncodeWriteResult;
+auto WriteChunksToStream(MessageBufferHandle_t* stream,
+                         uint8_t* working_buffer,
+                         size_t working_buffer_length,
+                         std::function<size_t(uint8_t*, size_t)> callback,
+                         TickType_t max_wait) -> EncodeWriteResult;
+
+class ChunkReader {
+ public:
+  ChunkReader(MessageBufferHandle_t* stream);
+  ~ChunkReader();
+
+  auto Reset() -> void;
+
+  auto GetLastMessage() -> std::pair<uint8_t*, size_t>;
 
   enum ChunkReadResult {
     // Returned an error in parsing the cbor-encoded header.
@@ -38,6 +53,8 @@ auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer,
     CHUNK_READ_TIMEOUT,
     // Returned when a non-chunk message is received.
     CHUNK_STREAM_ENDED,
+    // Returned when the processing callback does not return a value.
+    CHUNK_PROCESSING_ERROR,
   };
 
   /*
@@ -52,6 +69,16 @@ auto WriteChunksToStream(MessageBufferHandle_t *stream, uint8_t *working_buffer,
    * If this function encounters a message in the stream that is not a chunk, it
    * will place the message at the start of the working_buffer and then return.
    */
-auto ReadChunksFromStream(MessageBufferHandle_t *stream, uint8_t *working_buffer, size_t working_buffer_length, std::function<size_t(uint8_t*,size_t)> callback, TickType_t max_wait) -> EncodeReadResult;
+  auto ReadChunkFromStream(
+      std::function<std::optional<size_t>(uint8_t*, size_t)> callback,
+      TickType_t max_wait) -> EncodeReadResult;
+
+ private:
+  MessageBufferHandle_t* stream_;
+  uint8_t* working_buffer_;
+
+  std::size_t leftover_bytes_ = 0;
+  std::size_t last_message_size_ = 0;
+};
 
 }  // namespace audio

src/audio/include/fatfs_audio_input.hpp

@@ -20,10 +20,17 @@ class FatfsAudioInput : public IAudioElement {
 
   auto OutputBuffer() -> MessageBufferHandle_t;
 
+  auto SendChunk(uint8_t* buffer, size_t size) -> size_t;
+
  private:
   std::shared_ptr<drivers::SdStorage> storage_;
 
-  uint8_t *working_buffer_;
+  uint8_t* file_buffer_;
+  uint8_t* file_buffer_read_pos_;
+  uint8_t* pending_read_pos_;
+  uint8_t* file_buffer_write_pos_;
+
+  uint8_t* chunk_buffer_;
 
   FIL current_file_;
   bool is_file_open_ = false;

src/audio/include/stream_info.hpp

@@ -13,9 +13,11 @@ class StreamInfo {
   enum ParseError {
     WRONG_TYPE,
     MISSING_MAP,
+    CBOR_ERROR,
   };
 
-    static auto Create(const uint8_t *buffer, size_t length) -> cpp::result<StreamInfo, ParseError>;
+  static auto Create(const uint8_t* buffer, size_t length)
+      -> cpp::result<StreamInfo, ParseError>;
   StreamInfo(CborValue& map);
 
   StreamInfo() = default;
@@ -25,16 +27,20 @@ class StreamInfo {
 
   auto Path() const -> const std::optional<std::string>& { return path_; }
   auto Channels() const -> const std::optional<uint8_t>& { return channels_; }
-    auto BitsPerSample() const -> const std::optional<uint8_t>& { return bits_per_sample_; }
+  auto BitsPerSample() const -> const std::optional<uint8_t>& {
-    auto SampleRate() const -> const std::optional<uint16_t>& { return sample_rate_; }
+    return bits_per_sample_;
+  }
+  auto SampleRate() const -> const std::optional<uint16_t>& {
+    return sample_rate_;
+  }
 
   enum EncodeError {
     OUT_OF_MEMORY,
   };
 
-    auto WriteToStream(CborEncoder encoder) -> cpp::result<void, EncodeError>;
+  auto WriteToMap(CborEncoder encoder) -> cpp::result<size_t, EncodeError>;
-  private:
 
+ private:
   std::optional<std::string> path_;
   std::optional<uint8_t> channels_;
   std::optional<uint8_t> bits_per_sample_;

src/audio/stream_info.cpp

@@ -1,7 +1,8 @@
 #include "stream_info.hpp"
-#include "stream_message.hpp"
 #include <cstdint>
+#include "cbor_decoder.hpp"
 #include "esp-idf/components/cbor/tinycbor/src/cbor.h"
+#include "stream_message.hpp"
 
 namespace audio {
 
@@ -10,36 +11,16 @@ namespace audio {
 static const char* kKeyBitsPerSample = "b";
 static const char* kKeySampleRate = "r";
 
-  static auto find_uint64(CborValue &map, char *key) -> cpp::optional<uint64_t> {
+static auto StreamInfo::Create(const uint8_t* buffer, size_t length)
-    CborValue val;
+    -> cpp::result<StreamInfo, ParseError> {
-    cbor_value_map_find_value(&map, key, &val);
-    if (cbor_value_is_unsigned_integer(&val)) {
-      uint64_t raw_val;
-      cbor_value_get_uint64(&val, &raw_val);
-      return raw_val;
-    }
-    return {};
-  }
-
-
-  static auto write_uint64(CborEncoder &map, const char *key, const optional<uint64_t> &val) -> cpp::result<void, StreamInfo::EncodeError> {
-    if (val) {
-      cbor_encode_byte_string(&map, key, 1);
-      cbor_encode_uint(&map, *val);
-    }
-    return {};
-  }
-
-static auto StreamInfo::Create(const uint8_t *buffer, size_t length) -> cpp::result<StreamInfo, ParseError> {
   CborParser parser;
   CborValue value;
 
   cbor_parser_init(buffer, len, 0, &parser, &value);
 
   uint8_t type = 0;
-  if (!cbor_value_is_integer(&value)
+  if (!cbor_value_is_integer(&value) ||
-      || !cbor_value_get_integer(&value, &type)
+      !cbor_value_get_integer(&value, &type) || type != STREAM_INFO) {
-      || type != STREAM_INFO) {
     return cpp::fail(WRONG_TYPE);
   }
 
@@ -55,36 +36,21 @@ static auto StreamInfo::Create(const uint8_t *buffer, size_t length) -> cpp::res
 StreamInfo::StreamInfo(CborValue& map) {
   // TODO: this method is n^2, which seems less than ideal. But you don't do it
   // that frequently, so maybe it's okay? Needs investigation.
-  channels_ = find_uint64(map, kKeyChannels);
+  cbor::MapDecoder decoder(map);
-  bits_per_sample_ = find_uint64(map, kKeyBitsPerSample);
+  channels_ = decoder.FindValue(kKeyChannels);
-  sample_rate_ = find_uint64(map, kKeySampleRate);
+  bits_per_sample_ = decoder.FindValue(kKeyBitsPerSample);
-
+  sample_rate_ = decoder.FindValue(kKeySampleRate);
-  CborValue val;
+  path_ = decoder.FindValue(kKeyPath);
-  cbor_value_map_find_value(&map, kKeyPath, &val);
-  if (cbor_value_is_text_string(&val)) {
-    size_t len;
-    char *str;
-    cbor_value_dup_text_string(&val, &str, &len, &val);
-    path_ = std::string(str, len);
-    free(str);
 }
-}
-
-auto StreamInfo::WriteToStream(CborEncoder encoder) -> cpp::result<void, EncodeError> {
-  cbor_encode_int(&encoder, STREAM_INFO);
 
+auto StreamInfo::WriteToMap(cbor::Encoder& map_encoder)
+    -> cpp::result<size_t, EncodeError> {
   CborEncoder map;
-  cbor_encoder_create_map(&encoder, &map, length);
+  map_encoder.WriteKeyValue(kKeyChannels, channels_);
-
+  map_encoder.WriteKeyValue(kKeyBitsPerSample, bits_per_sample_);
-  write_uint64(&map, kKeyChannels, channels_);
+  map_encoder.WriteKeyValue(kKeySampleRate, sample_rate_);
-  write_uint64(&map, kKeyBitsPerSample, bits_per_sample_);
+  map_encoder.WriteKeyValue(kKeyPath, path_);
-  write_uint64(&map, kKeySampleRate, sample_rate_);
+  return map_encoder.Finish();
-
-  if (path_) {
-    cbor_encode_text_string(&map, path_->c_str(), path_->size());
-  }
-
-  cbor_encoder_close_container(&encoder, &map);
 }
 
 }  // namespace audio

src/cbor/cbor_decoder.cpp

@@ -5,7 +5,8 @@
 
 namespace cbor {
 
-static auto ArrayDecoder::Create(uint8_t *buffer, size_t buffer_len) -> cpp::result<std::unique_ptr<ArrayDecoder>, CborError> {
+static auto ArrayDecoder::Create(uint8_t* buffer, size_t buffer_len)
+    -> cpp::result<std::unique_ptr<ArrayDecoder>, CborError> {
   auto decoder = std::make_unique<ArrayDecoder>();
   cbor_parser_init(buffer, buffer_len, &decoder->parser_, &decoder->root_);
   if (!cbor_value_is_array(&decoder->root_)) {
@@ -18,7 +19,23 @@ static auto ArrayDecoder::Create(uint8_t *buffer, size_t buffer_len) -> cpp::res
   return std::move(decoder);
 }
 
-static auto MapDecoder::Create(uint8_t *buffer, size_t buffer_len) -> cpp::result<std::unique_ptr<MapDecoder>, CborError> {
+static auto ArrayDecoder::Create(CborValue& root)
+    -> cpp::result<std::unique_ptr<ArrayDecoder>, CborError> {
+  auto decoder = std::make_unique<ArrayDecoder>();
+  decoder->root_ = root;
+  if (!cbor_value_is_array(&decoder->root_)) {
+    return cpp::fail(CborErrorIllegalType);
+  }
+
+  CborError err = cbor_value_enter_container(&decoder->root_, &decoder->it_);
+  if (err != CborNoError) {
+    return cpp::fail(err);
+  }
+  return std::move(decoder);
+}
+
+static auto MapDecoder::Create(uint8_t* buffer, size_t buffer_len)
+    -> cpp::result<std::unique_ptr<MapDecoder>, CborError> {
   auto decoder = std::make_unique<MapDecoder>();
   cbor_parser_init(buffer, buffer_len, &decoder->parser_, &decoder->root_);
   if (!cbor_value_is_map(&decoder->root_)) {
@@ -31,7 +48,22 @@ static auto MapDecoder::Create(uint8_t *buffer, size_t buffer_len) -> cpp::resul
   return std::move(decoder);
 }
 
-auto MapDecoder::FindString(const std::string &key) -> std::optional<std::string> {
+static auto MapDecoder::Create(CborValue& root)
+    -> cpp::result<std::unique_ptr<MapDecoder>, CborError> {
+  auto decoder = std::make_unique<MapDecoder>();
+  decoder->root_ = root;
+  if (!cbor_value_is_map(&decoder->root_)) {
+    return cpp::fail(CborErrorIllegalType);
+  }
+  CborError err = cbor_value_enter_container(&decoder->root_, &decoder->it_);
+  if (err != CborNoError) {
+    return cpp::fail(err);
+  }
+  return std::move(decoder);
+}
+
+auto MapDecoder::FindString(const std::string& key)
+    -> std::optional<std::string> {
   CborValue val;
   if (error_ != CborNoError) {
     return {};
@@ -43,7 +75,8 @@ auto MapDecoder::FindString(const std::string &key) -> std::optional<std::string
     error_ = CborErrorIllegalType;
     return {};
   }
-  uint8_t *buf; size_t len;
+  uint8_t* buf;
+  size_t len;
   error_ = cbor_value_dup_byte_string(&val, &buf, &len, NULL);
   if (error_ != CborNoError) {
     return cpp::fail(error_);
@@ -53,7 +86,8 @@ auto MapDecoder::FindString(const std::string &key) -> std::optional<std::string
   return ret;
 }
 
-auto MapDecoder::FindUnsigned(const std::string &key) -> std::optional<uint32_t> {
+auto MapDecoder::FindUnsigned(const std::string& key)
+    -> std::optional<uint32_t> {
   CborValue val;
   if (error_ != CborNoError) {
     return {};

src/cbor/cbor_encoder.cpp

@@ -6,33 +6,39 @@ namespace cbor {
 
 static const int kEncoderFlags = 0;
 
-Encoder::Encoder(ContainerType type, uint32_t container_len, uint8_t *buffer, size_t buffer_len) {
+Encoder::Encoder(ContainerType type,
+                 uint32_t container_len,
+                 uint8_t* buffer,
+                 size_t buffer_len) {
   cbor_encoder_init(&root_encoder, buffer, buffer_len, kEncoderFlags);
   switch (type) {
     case CONTAINER_ARRAY:
-      error_ = cbor_encoder_create_array(&encoder, &container_encoder_, container_len);
+      error_ = cbor_encoder_create_array(&encoder, &container_encoder_,
+                                         container_len);
       break;
     case CONTAINER_MAP:
-      error_ = cbor_encoder_create_map(&encoder, &container_encoder_, container_len);
+      error_ =
+          cbor_encoder_create_map(&encoder, &container_encoder_, container_len);
       break;
   }
 }
 
-auto Encoder::WriteString(const std::string &val) -> void {
+auto Encoder::WriteValue(const std::string& val) -> void {
   if (error_ != CborNoError) {
     return;
   }
-  error_ = cbor_encode_byte_string(&container_encoder_, val.c_str(), val.size());
+  error_ =
+      cbor_encode_byte_string(&container_encoder_, val.c_str(), val.size());
 }
 
-auto Encoder::WriteUnsigned(uint32_t val) -> void {
+auto Encoder::WriteValue(uint32_t val) -> void {
   if (error_ != CborNoError) {
     return;
   }
   error_ = cbor_encode_uint(&container_encoder_, val);
 }
 
-auto Encoder::WriteSigned(int32_t val) -> void {
+auto Encoder::WriteValue(int32_t val) -> void {
   if (error_ != CborNoError) {
     return;
   }
@@ -43,11 +49,12 @@ auto Encoder::Finish() -> cpp::result<size_t, CborError> {
   if (error_ != CborNoError) {
     return cpp::fail(error_);
   }
-  if (CborError final_error = cbor_encoder_close_container(&root_encoder, &container_encoder_) != CborNoError) {
+  if (CborError final_error =
+          cbor_encoder_close_container(&root_encoder, &container_encoder_) !=
+          CborNoError) {
     return cpp::fail(final_error);
   }
   return cbor_encoder_get_buffer_size(&root_encoder);
 }
 
 }  // namespace cbor
-

src/cbor/include/cbor_decoder.hpp

@@ -6,7 +6,8 @@
 namespace cbor {
 
 static auto parse_stdstring(CborValue* val, std::string* out) -> CborError {
-        uint8_t *buf; size_t len;
+  uint8_t* buf;
+  size_t len;
   CborError err = cbor_value_dup_byte_string(val, &buf, &len, NULL);
   if (err != CborNoError) {
     return err;
@@ -21,23 +22,29 @@ namespace cbor {
   static auto Create(uint8_t* buffer, size_t buffer_len)
       -> cpp::result<std::unique_ptr<ArrayDecoder>, CborError>;
 
+  static auto Create(CborValue& root)
+      -> cpp::result<std::unique_ptr<ArrayDecoder>, CborError>;
+
   template <typename T>
   auto NextValue() -> cpp::result<T, CborError>;
 
-      template<> auto NextValue() -> cpp::result<int64_t, CborError> {
+  template <>
+  auto NextValue() -> cpp::result<int64_t, CborError> {
     return NextValue(&cbor_value_is_integer, &cbor_value_get_int);
   }
-      template<> auto NextValue() -> cpp::result<uint64_t, CborError> {
+  template <>
+  auto NextValue() -> cpp::result<uint64_t, CborError> {
     return NextValue(&cbor_value_is_unsigned_integer, &cbor_value_get_uint64);
   }
-      template<> auto NextValue() -> cpp::result<std::string, CborError> {
+  template <>
+  auto NextValue() -> cpp::result<std::string, CborError> {
     return NextValue(&cbor_value_is_byte_string, &parse_stdstring);
   }
 
   template <typename T>
-      auto NextValue(
+  auto NextValue(bool (*is_valid)(CborValue*),
-          bool(*is_valid)(CborValue*),
+                 CborError (*parse)(CborValue*, T*))
-          CborError(*parse)(CborValue*, T*)) -> cpp::result<T, CborError> {
+      -> cpp::result<T, CborError> {
     if (error_ != CborNoError) {
       return cpp::fail(error_);
     }
@@ -59,8 +66,11 @@ namespace cbor {
 
   auto Failed() -> CborError { return error_; }
 
+  auto Iterator() -> CborValue& { return it_; }
+
   ArrayDecoder(const ArrayDecoder&) = delete;
   ArrayDecoder& operator=(const ArrayDecoder&) = delete;
+
  private:
   CborParser parser_;
   CborValue root_;
@@ -71,24 +81,31 @@ namespace cbor {
 
 class MapDecoder {
  public:
-      static auto Create(uint8_t *buffer, size_t buffer_len) -> cpp::result<std::unique_ptr<MapDecoder>, CborError>;
+  static auto Create(uint8_t* buffer, size_t buffer_len)
+      -> cpp::result<std::unique_ptr<MapDecoder>, CborError>;
+
+  static auto Create(CborValue& root)
+      -> cpp::result<std::unique_ptr<MapDecoder>, CborError>;
 
   template <typename T>
   auto FindValue(const std::string& key) -> std::optional<T>;
 
-      template<> auto FindValue(const std::string &key) -> std::optional<int64_t> {
+  template <>
+  auto FindValue(const std::string& key) -> std::optional<int64_t> {
     return FindValue(key, &cbor_value_is_integer, &cbor_value_get_int);
   }
-      template<> auto FindValue(const std::string &key) -> std::optional<uint64_t> {
+  template <>
-        return FindValue(key, &cbor_value_is_unsigned_integer, &cbor_value_get_uint64);
+  auto FindValue(const std::string& key) -> std::optional<uint64_t> {
+    return FindValue(key, &cbor_value_is_unsigned_integer,
+                     &cbor_value_get_uint64);
   }
-      template<> auto FindValue(const std::string &key) -> std::optional<std::string> {
+  template <>
+  auto FindValue(const std::string& key) -> std::optional<std::string> {
     return FindValue(key, &cbor_value_is_byte_string, &parse_stdstring);
   }
 
   template <typename T>
-      auto FindValue(
+  auto FindValue(const std::string& key,
-          const std::string &key,
                  bool (*is_valid)(CborValue*),
                  CborError (*parse)(CborValue*, T*)) -> std::optional<T> {
     if (error_ != CborNoError) {
@@ -113,6 +130,7 @@ namespace cbor {
 
   MapDecoder(const MapDecoder&) = delete;
   MapDecoder& operator=(const MapDecoder&) = delete;
+
  private:
   CborParser parser_;
   CborValue root_;
@@ -121,5 +139,4 @@ namespace cbor {
   CborError error_ = CborNoError;
 };
 
-
 }  // namespace cbor

src/cbor/include/cbor_encoder.hpp

@@ -6,20 +6,27 @@ namespace cbor {
 
 class Encoder {
  public:
-      enum ContainerType {
+  enum ContainerType { CONTAINER_ARRAY, CONTAINER_MAP };
-        CONTAINER_ARRAY,
+  Encoder(ContainerType type,
-        CONTAINER_MAP
+          uint32_t container_len,
-      };
+          uint8_t* buffer,
-      Encoder(ContainerType type, uint32_t container_len, uint8_t *buffer, size_t buffer_len);
+          size_t buffer_len);
+
+  template <typename T>
+  auto WriteKeyValue(const std::string& key, const T& val) -> void {
+    WriteValue(key);
+    WriteValue(val);
+  }
 
-      auto WriteString(const std::string &val) -> void;
+  auto WriteValue(const std::string& val) -> void;
-      auto WriteUnsigned(uint32_t val) -> void;
+  auto WriteValue(uint32_t val) -> void;
-      auto WriteSigned(int32_t val) -> void;
+  auto WriteValue(int32_t val) -> void;
 
   auto Finish() -> cpp::result<size_t, CborError>;
 
   Encoder(const Encoder&) = delete;
   Encoder& operator=(const Encoder&) = delete;
+
  private:
   CborEncoder root_encoder_;
   CborEncoder container_encoder_;

src/codecs/CMakeLists.txt

@@ -1,5 +1,5 @@
 idf_component_register(
-  SRCS "mad.cpp"
+  SRCS "codec.cpp" "mad.cpp"
   INCLUDE_DIRS "include")
 
 target_compile_options(${COMPONENT_LIB} PRIVATE ${EXTRA_WARNINGS})

src/codecs/codec.cpp

@@ -2,7 +2,8 @@
 
 namespace codecs {
 
-auto CreateCodecForExtension(std::string extension) -> cpp::result<std::unique_ptr<ICodec>, CreateCodecError> {
+auto CreateCodecForExtension(std::string extension)
+    -> cpp::result<std::unique_ptr<ICodec>, CreateCodecError> {
   return cpp::fail(UNKNOWN_EXTENSION);
 }
 

src/codecs/include/codec.hpp

@@ -8,17 +8,16 @@
 
 namespace codecs {
 
-  enum CreateCodecError {
+enum CreateCodecError { UNKNOWN_EXTENSION };
-    UNKNOWN_EXTENSION
-  };
 
-  auto CreateCodecForExtension(std::string extension) -> cpp::result<std::unique_ptr<ICodec>, CreateCodecError>;
+auto CreateCodecForFile(const std::string& extension)
+    -> cpp::result<std::unique_ptr<ICodec>, CreateCodecError>;
 
 class ICodec {
  public:
   virtual ~ICodec() {}
 
-      virtual auto CanHandleExtension(std::string extension) -> bool = 0;
+  virtual auto CanHandleFile(const std::string& path) -> bool = 0;
 
   struct OutputFormat {
     uint8_t num_channels;
@@ -28,7 +27,7 @@ namespace codecs {
 
   virtual auto GetOutputFormat() -> OutputFormat = 0;
 
-      enum Error {};
+  enum ProcessingError {};
 
   struct Result {
     bool need_more_input;
@@ -48,11 +47,14 @@ namespace codecs {
     std::size_t output_written;
   };
 
-      virtual auto Process(
+  virtual auto ResetForNewStream() -> void = 0;
-          uint8_t *input,
+
-          std::size_t input_len,
+  virtual auto SetInput(uint8_t* buffer, std::size_t length) = 0;
-          uint8_t *output,
+  virtual auto Process(uint8_t* output, std::size_t output_length)
-	  std::size_t output_length) -> cpp::result<Result, Error>  = 0;
+      -> cpp::result<size_t, Error> = 0;
+
+  virtual auto GetOutputProcessed() -> std::size_t;
+  = 0;
 };
 
 }  // namespace codecs

src/codecs/include/mad.hpp

@@ -10,7 +10,9 @@ namespace codecs {
   ~MadMp3Decoder();
 
   auto ProcessInput(Result* res, uint8_t* input, std::size_t input_len) -> void;
-      auto WriteOutputSamples(Result *res, uint8_t *output, std::size_t output_length) -> void;
+  auto WriteOutputSamples(Result* res,
+                          uint8_t* output,
+                          std::size_t output_length) -> void;
 
  private:
   mad_stream stream_;

src/codecs/include/types.hpp

@@ -9,4 +9,4 @@ namespace codecs  {
 };
 
 auto GetStreamTypeFromFilename(std::string filename);
-}
+}  // namespace codecs

src/codecs/mad.cpp

@@ -44,16 +44,13 @@ namespace codecs {
   };
 }
 
-      auto MadMp3Decoder::Process(
+auto MadMp3Decoder::Process(uint8_t* input,
-          uint8_t *input,
                             std::size_t input_len,
                             uint8_t* output,
-	  std::size_t output_length) -> cpp::result<Result, Error> {
+                            std::size_t output_length)
-
+    -> cpp::result<Result, Error> {
   Result res {
-          .need_more_input = false,
+    .need_more_input = false, .input_processed = 0, .flush_output = false,
-          .input_processed = 0,
-          .flush_output = false,
     .output_written = 0,
   }
   while (true) {
@@ -73,9 +70,9 @@ namespace codecs {
     }
   }
 
-      auto MadMp3Decoder::ProcessInput(
+  auto MadMp3Decoder::ProcessInput(Result * res, uint8_t * input,
-          Result *res, uint8_t *input, std::size_t input_len) -> void {
+                                   std::size_t input_len)
-
+      ->void {
     if (input != stream_.buffer) {
       mad_stream_buffer(&stream_, input, input_len);
     }
@@ -90,7 +87,6 @@ namespace codecs {
       // duration will help with seeking?
     }
 
-
     // Decode the next frame. To signal errors, this returns -1 and
     // stashes an error code in the stream structure.
     if (mad_frame_decode(&frame_, &stream_) < 0) {
@@ -120,10 +116,9 @@ namespace codecs {
     up_to_sample = 0;
   }
 
-      auto MadMp3Decoder::WriteOutputSamples(
+  auto MadMp3Decoder::WriteOutputSamples(Result * res, uint8_t * output,
-          Result *res,
+                                         std::size_t output_length)
-          uint8_t *output,
+      ->void {
-	  std::size_t output_length) -> void {
     size_t output_byte = 0;
     // First ensure that we actually have some samples to send off.
     if (current_sample_ < 0) {

src/drivers/include/fatfs_audio_input.hpp

@@ -34,7 +34,6 @@ namespace drivers {
   RingbufHandle_t output_;
 
   std::string path_;
-
 };
 
 }  // namespace drivers

src/main/app_console.hpp

@@ -2,8 +2,8 @@
 
 #include <memory>
 
-#include "storage.hpp"
 #include "console.hpp"
+#include "storage.hpp"
 
 namespace console {
 

src/tasks/CMakeLists.txt

@@ -0,0 +1,2 @@
+idf_component_register(SRCS "tasks.cpp" INCLUDE_DIRS ".")
+target_compile_options(${COMPONENT_LIB} PRIVATE ${EXTRA_WARNINGS})

src/tasks/tasks.cpp

@@ -0,0 +1,4 @@
+#include "tasks.cpp"
+
+static const UBaseType_t kTaskPriorityLvgl = 4;
+static const UBaseType_t kTaskPriorityAudio = 5;

src/tasks/tasks.hpp

@@ -0,0 +1,6 @@
+#pragma once
+
+#include "freertos/portmacro.h"
+
+extern const UBaseType_t kTaskPriorityLvgl;
+extern const UBaseType_t kTaskPriorityAudio;