Mostly done pipeline arch. Now onto cleanup and building.

30 files changed, 990 insertions, 844 deletions

diff --git a/src/audio/CMakeLists.txt b/src/audio/CMakeLists.txt
index 89a2e54d..e3f7dd33 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" "fatfs_audio_input.cpp" "chunk.cpp"
-  "i2s_audio_output.cpp" "stream_info.cpp"
+  SRCS "audio_decoder.cpp" "audio_task.cpp" "chunk.cpp" "fatfs_audio_input.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})
diff --git a/src/audio/audio_decoder.cpp b/src/audio/audio_decoder.cpp
index 7ed67339..02217187 100644
--- a/src/audio/audio_decoder.cpp
+++ b/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
-  static const std::size_t kWorkingBufferSize = kMaxFrameSize;
-
-  AudioDecoder::AudioDecoder() {
-    working_buffer_ = heap_caps_malloc(kWorkingBufferSize, MALLOC_CAP_SPIRAM);
-  }
-
-  AudioDecoder::~AudioDecoder() {
-    free(working_buffer_);
-  }
-
-  auto AudioDecoder::InputBuffer() -> StreamBufferHandle_t {
-    return input_buffer_;
-  }
-
-  auto AudioDecoder::OutputBuffer() -> StreamBufferHandle_t {
-    return output_buffer_;
-  }
-
-  auto AudioDecoder::SetInputBuffer(StreamBufferHandle_t buffer) -> void {
-    input_buffer_ = buffer;
-  }
-
-  auto AudioDecoder::SetOutputBuffer(StreamBufferHandle_t buffer) -> void {
-    output_buffer_ = buffer;
+AudioDecoder::AudioDecoder()
+    : IAudioElement(),
+      chunk_buffer_(heap_caps_malloc(kMaxChunkSize, MALLOC_CAP_SPIRAM)),
+      stream_info_({}) {}
+
+AudioDecoder::~AudioDecoder() {
+  free(chunk_buffer_);
+}
+
+auto AudioDecoder::SetInputBuffer(StreamBufferHandle_t* buffer) -> void {
+  input_buffer_ = buffer;
+}
+
+auto AudioDecoder::SetOutputBuffer(StreamBufferHandle_t* buffer) -> void {
+  output_buffer_ = buffer;
+}
+
+auto AudioDecoder::ProcessStreamInfo(StreamInfo&& info)
+    -> cpp::result<void, StreamError> {
+  stream_info_ = info;
+
+  // Reuse the existing codec if we can. This will help with gapless playback,
+  // since we can potentially just continue to decode as we were before,
+  // without any setup overhead.
+  if (current_codec_->CanHandleFile(info.path)) {
+    current_codec_->ResetForNewStream();
+    return {};
   }
 
-  auto AudioDecoder::ProcessElementCommand(void* command) -> ProcessResult {
-    FatfsAudioInput::OutputCommand *real = std::reinterpret_cast<FatfsAudioInput::OutputCommand*>(command);
-
-    if (current_codec_->CanHandleExtension(real->extension)) {
-      // TODO: Do we need to reset the codec?
-      delete real;
-      return OK;
-    }
-
-    auto result = codecs::CreateCodecForExtension(real->extension);
-    // TODO: handle error case
-    if (result.has_value()) {
-      current_codec_ = result.value();
-    }
-
-    delete real;
-    return OK;
+  auto result = codecs::CreateCodecForFile(info.path);
+  if (result.has_value()) {
+    current_codec_ = std::move(result.value());
+  } else {
+    return cpp::fail(UNSUPPORTED_STREAM);
   }
 
-  auto AudioDecoder::SkipElementCommand(void* command) -> void {
-    FatfsAudioInput::OutputCommand *real = std::reinterpret_cast<FatfsAudioInput::OutputCommand*>(command);
-    delete real;
-  }
-
-  auto AudioDecoder::ProcessData(uint8_t* data, uint16_t length) -> ProcessResult {
-    if (current_codec_ == nullptr) {
-      // TODO: signal this
-      return OK;
-    }
-
-    while (true) {
-      auto result = current_codec_->Process(data, length, working_buffer_, kWorkingBufferSize);
-
-      if (result.has_error()) {
-        // TODO: handle i guess
-        return ERROR;
-      }
-      ICodec::Result process_res = result.value();
+  return {};
+}
 
-      if (process_res.flush_output) {
-        xStreamBufferSend(&output_buffer_, working_buffer_, process_res.output_written, kMaxWaitTicks);
-      }
-
-      if (process_res.need_more_input) {
-        // TODO: wtf do we do about the leftover bytes?
-        return OK;
-      }
-    }
-
-    return OK;
-  }
-
-  auto AudioDecoder::ProcessIdle() -> ProcessResult {
-    // Not used.
-    return OK;
+auto AudioDecoder::ProcessChunk(uint8_t* data, std::size_t length)
+    -> cpp::result<size_t, StreamError> {
+  if (current_codec_ == nullptr) {
+    // Should never happen, but fail explicitly anyway.
+    return cpp::fail(UNSUPPORTED_STREAM);
   }
 
-  auto AudioDecoder::Pause() -> void {
-    // TODO.
-  }
-  auto AudioDecoder::IsPaused() -> bool {
-    // TODO.
+  current_codec_->SetInput(data, length);
+  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()) {
+          // End our output stream immediately if the codec barfed.
+          return 0;
+        }
+        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 (result.has_error()) {
+    return cpp::fail(IO_ERROR);
   }
 
-  auto AudioDecoder::Resume() -> void {
-    // TODO.
-  }
+  return current_codec_->GetOutputProcessed();
+}
 
+auto AudioDecoder::ProcessIdle() -> cpp::result<void, StreamError> {
+  // Not used; we delay forever when waiting on IO.
+  return {};
+}
 
-} // namespace audio
+}  // namespace audio
diff --git a/src/audio/audio_task.cpp b/src/audio/audio_task.cpp
index ad0834e2..a125548a 100644
--- a/src/audio/audio_task.cpp
+++ b/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();
-
-  uint8_t* message_buffer =
-      (uint8_t*)heap_caps_malloc(kFrameSize, MALLOC_CAP_SPIRAM);
+  ChunkReader chunk_reader = ChunkReader(element->InputBuffer());
 
   while (1) {
-    BaseType_t rtos_res;
-    IAudioElement::ProcessResult result;
-
-
-    size_t message_size = 0;
-    if (message_buffer != nullptr) {
-      // TODO: tune delay.
-      message_size = xMessageBufferReceive(stream, &message_buffer, kFrameSize, portMAX_DELAY);
-    }
-
-    if (message_size == 0) {
-      element->ProcessIdle();
-      continue;
-    }
-
-    // We got a valid message. Check what kind it is so that we know how to
-    // process it.
-    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.
-    }
-
-    if (message_type == STREAM_INFO) {
-      errs = StreamInfo::Create(message_buffer, message_size).map(element->ProcessStreamInfo);
-      if (errs.has_error) {
-        // TODO;
+    cpp::result<size_t, IAudioElement::StreamError> process_res;
+
+    // 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
+    // cases.
+    bool has_received_message = false;
+    if (stream != nullptr) {
+      EncodeReadResult chunk_res = chunk_reader.ReadChunkFromStream(
+          [&](uint8_t* data, std::size_t length) -> std::optional<size_t> {
+            process_res = element->ProcessChunk(data, length);
+            if (process_res.has_value()) {
+              return process_res.value();
+            } else {
+              return {};
+            }
+          },
+          element->IdleTimeout());
+
+      if (chunk_res == CHUNK_PROCESSING_ERROR ||
+          chunk_res == CHUNK_DECODING_ERROR) {
+        break;  // TODO.
+      } else if (chunk_res == CHUNK_STREAM_ENDED) {
+        has_received_message = true;
       }
-    } 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;
-        result = element->ProcessData(&data_in, read_length);
-        if (result == IAudioElement::ERROR) {
-          break;
-        }
-
-        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) {
-          vTaskDelay(kIdleTaskDelay);
-        }
-      }
-    } 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) {
-          bytes_in_stream += command.read_size;
-        } else {
-          // This data is for a different stream, so just discard it.
-          xStreamBufferReceive(stream, &frame_buffer, command.read_size, 0);
-        }
-      } else if (command.type == IAudioElement::ELEMENT) {
-        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) {
+    if (has_received_message) {
+      auto& [buffer, length] = chunk_reader.GetLastMessage();
+      auto decoder_res = cbor::ArrayDecoder::Create(buffer, length);
+      if (decoder_res.has_error()) {
+        // TODO.
         break;
       }
+      auto decoder = decoder_res.value();
+      MessageType message_type = decoder->NextValue();
+      if (message_type == TYPE_STREAM_INFO) {
+        element->ProcessStreamInfo(StreamInfo(decoder->Iterator()););
+      }
+    }
+
+    // TODO: Do any out of band reading, such a a pause command, here.
+
+    // Chunk reading must have timed out, or we don't have an input stream.
+    // Signal the element to do any of its idle tasks.
+    process_res = element->ProcessIdle();
+    if (process_res.has_error()) {
+      break;  // TODO.
     }
   }
 
-  element = nullptr;
-  free(frame_buffer);
+  element.clear();
+  free(chunk_buffer_);
 
-  xTaskDelete(NULL);
+  vTaskDelete(NULL);
 }
 
 }  // namespace audio
diff --git a/src/audio/chunk.cpp b/src/audio/chunk.cpp
index 40564069..a157b946 100644
--- a/src/audio/chunk.cpp
+++ b/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,15 +24,16 @@ 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 =
-      callback(
-          working_buffer + header_size,
-          working_buffer_length - header_size);
+    size_t chunk_size = callback(working_buffer + header_size,
+                                 working_buffer_length - header_size);
 
     if (chunk_size == 0) {
       // They had nothing for us, so bail out.
@@ -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,9 +61,8 @@ 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 =
-      xMessageBufferSend(
-          *stream, working_buffer, header_size + chunk_size, max_wait);
+    size_t actual_write_size = xMessageBufferSend(
+        *stream, working_buffer, header_size + chunk_size, max_wait);
 
     header_size = new_header_size;
 
@@ -67,58 +75,78 @@ 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 {
-  // Spillover if the previous iteration did not consume all of the input.
-  size_t leftover_bytes = 0;
-  while (1) {
-    // First, wait for a message to arrive over the buffer.
-    size_t read_size =
-      xMessageBufferReceive(
-          *stream, working_buffer + leftover_bytes, working_buffer_length - leftover_bytes, max_wait);
+ChunkReader::ChunkReader(MessageBufferHandle_t* stream) : stream_(stream) {
+  working_buffer_ = heap_caps_malloc(kWorkingBufferSize, MALLOC_CAP_SPIRAM);
+};
 
-    if (read_size == 0) {
-      return CHUNK_READ_TIMEOUT;
-    }
+ChunkReader::~ChunkReader() {
+  free(working_buffer_);
+}
 
-    auto decoder = cbor::MapDecoder::Create(working_buffer + leftover_bytes, read_size);
-    if (decoder.has_error()) {
-      // Weird; this implies someone is shoving invalid data into the buffer.
-      return CHUNK_DECODING_ERROR;
-    }
+auto ChunkReader::Reset() -> void {
+  leftover_bytes_ = 0;
+  last_message_size_ = 0;
+}
 
-    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
-      // caller handle it.
-      memmove(working_buffer, working_buffer + leftover_bytes, read_size);
-      return CHUNK_STREAM_ENDED;
-    }
+auto ChunkReader::GetLastMessage() -> std::pair<uint8_t*, size_t> {
+  return std::make_pair(working_buffer_ + leftover_bytes_, last_message_size_);
+}
 
-    // Work the size and position of the chunk.
-    header_length = decoder.ParseUnsigned().value_or(0);
-    chunk_length = decoder.ParseUnsigned().value_or(0);
-    if (decoder.Failed()) {
-      return CHUNK_DECODING_ERROR;
-    }
+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.
+  last_message_size_ =
+      xMessageBufferReceive(*stream_, working_buffer_ + leftover_bytes_,
+                            kWorkingBufferSize - leftover_bytes_, max_wait);
 
-    // 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;
-    size_t combined_buffer_size = leftover_bytes + chunk_length;
-    if (leftover_bytes > 0) {
-      memmove(combined_buffer, working_buffer, leftover_bytes);
-    }
+  if (last_message_size_ == 0) {
+    return CHUNK_READ_TIMEOUT;
+  }
 
-    // Tell the callback about the new data.
-    size_t amount_processed = callback(combined_buffer, combined_buffer_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;
+  }
 
-    // Prepare for the next iteration.
-    leftover_bytes = combined_buffer_size - amount_processed;
-    if (leftover_bytes > 0) {
-      memmove(working_buffer, combined_buffer + amount_processed, leftover_bytes);
-    }
+  MessageType type = decoder.value().ParseUnsigned().value_or(TYPE_UNKNOWN);
+  if (type != TYPE_CHUNK_HEADER) {
+    // This message wasn't for us, so let the caller handle it.
+    Reset();
+    return CHUNK_STREAM_ENDED;
+  }
+
+  // Work the size and position of the chunk.
+  header_length = decoder.ParseUnsigned().value_or(0);
+  chunk_length = decoder.ParseUnsigned().value_or(0);
+  if (decoder.Failed()) {
+    return CHUNK_DECODING_ERROR;
+  }
+
+  // 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_;
+  size_t combined_buffer_size = leftover_bytes_ + chunk_length;
+  if (leftover_bytes_ > 0) {
+    memmove(combined_buffer, working_buffer_, leftover_bytes_);
+  }
+
+  // Tell the callback about the new data.
+  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.value();
+  if (leftover_bytes_ > 0) {
+    memmove(working_buffer_, combined_buffer + amount_processed.value(),
+            leftover_bytes_);
   }
 }
 
-} // namespace audio
+}  // namespace audio
diff --git a/src/audio/fatfs_audio_input.cpp b/src/audio/fatfs_audio_input.cpp
index 0cf1abe5..351fd017 100644
--- a/src/audio/fatfs_audio_input.cpp
+++ b/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
-// into memory as soon as possible.
-static constexpr std::size_t kOutputBufferSize = 1024 * 128;
-static constexpr std::size_t kQueueItemSize = sizeof(IAudioElement::Command);
+static const std::size_t kFileBufferSize = 1024 * 128;
+static const std::size_t kMinFileReadSize = 1024 * 4;
+static const std::size_t kOutputBufferSize = 1024 * 4;
 
 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);
-  output_buffer_ =
-      xMessageBufferCreateStatic(kOutputBufferSize, output_buffer_memory_,
-                                &output_buffer_metadata_);
+      heap_caps_malloc(kOutputBufferSize, MALLOC_CAP_SPIRAM);
+  output_buffer_ = xMessageBufferCreateStatic(
+      kOutputBufferSize, output_buffer_memory_, &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 {
-  InputCommand *real = std::reinterpret_cast<InputCommand*>(command);
-
-  if (uxQueueSpacesAvailable(output_queue_) < 2) {
-    return OUTPUT_FULL;
-  }
-
+auto FatfsAudioInput::ProcessStreamInfo(StreamInfo& info)
+    -> cpp::result<void, StreamError> {
   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 ERROR;
-  }
-
-  if (real->seek_to && f_lseek(&current_file_, real->seek_to) {
-      return ERROR;
+    return cpp::fail(IO_ERROR);
   }
 
   is_file_open_ = true;
 
-  if (real->interrupt) {
-    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);
+  // TODO: pass on stream info.
 
-  delete real;
-  return OK;
+  return {};
 }
 
-auto FatfsAudioInput::SkipElementCommand(void* command) -> void {
-  InputCommand *real = std::reinterpret_cast<input_key_service_add_key*>(command);
-  delete real;
+auto FatfsAudioInput::ProcessChunk(uint8_t* data, std::size_t length)
+    -> cpp::result<void, StreamError> {
+  // TODO.
+  return {};
 }
 
-auto FatfsAudioInput::ProcessData(uint8_t* data, uint16_t length) -> void {
-  // Not used, since we have no input stream.
+static auto GetRingBufferDistance() -> size_t {
+  if (file_buffer_read_pos_ == file_buffer_write_pos_) {
+    return 0;
+  }
+  if (file_buffer_read_pos_ < file_buffer_write_pos_) {
+    return file_buffer_write_pos_ - file_buffer_read_pos_;
+  }
+  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_)
 }
 
-auto FatfsAudioInput::ProcessIdle() -> ProcessResult {
-  if (!is_file_open_) {
-    return OK;
+virtual auto FatfsAudioInput::ProcessIdle() -> cpp::result<void, StreamError> {
+  // 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_, file_buffer_write_pos_, read_size,
+                              &bytes_read);
+      if (!FR_OK) {
+        return ERROR;  // TODO;
+      }
+
+      if (f_eof(&current_file_)) {
+        f_close(&current_file_);
+        is_file_open_ = false;
+
+        // TODO: open the next file?
+      }
+
+      file_buffer_write_pos_ += bytes_read;
+      if (file_buffer_write_pos_ == file_buffer_ + kFileBufferSize) {
+        file_buffer_write_pos_ = file_buffer_;
+      }
+    }
   }
 
-  if (xStreamBufferSpacesAvailable(output_buffer) < kMaxFrameSize) {
-    return OUTPUT_FULL;
+  // 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.
   }
+}
 
-  UINT bytes_read = 0;
-  FRESULT result = f_read(&current_file_, working_buffer_, kMaxFrameSize, &bytes_read);
-  if (!FR_OK) {
-    return ERROR;
+auto FatfsAudioInput::SendChunk(uint8_t* buffer, size_t size) -> size_t {
+  if (pending_read_pos_ != nullptr) {
+    file_buffer_read_pos_ = pending_read_pos_;
   }
 
-  xStreamBufferSend(&output_buffer_, working_buffer_, bytes_read, kMaxWaitTicks);
-
-  if (f_eof(&current_file_)) {
-    f_close(&current_file_);
-    is_file_open_ = false;
+  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);
 
-  return OK;
+  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
diff --git a/src/audio/include/audio_decoder.hpp b/src/audio/include/audio_decoder.hpp
index 2ee43fb7..98ebdc71 100644
--- a/src/audio/include/audio_decoder.hpp
+++ b/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 SetOutputBuffer(StreamBufferHandle_t) -> void;
+  auto SetInputBuffer(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_;
-
-  StreamBufferHandle_t input_buffer_;
-  StreamBufferHandle_t output_buffer_;
+  uint8_t* chunk_buffer_;
 };
 
 }  // namespace audio
diff --git a/src/audio/include/audio_element.hpp b/src/audio/include/audio_element.hpp
index 0be58f48..2a2f0727 100644
--- a/src/audio/include/audio_element.hpp
+++ b/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
diff --git a/src/audio/include/audio_task.hpp b/src/audio/include/audio_task.hpp
index 79604f33..05888170 100644
--- a/src/audio/include/audio_task.hpp
+++ b/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
diff --git a/src/audio/include/chunk.hpp b/src/audio/include/chunk.hpp
index 1351ecfb..a3f943ea 100644
--- a/src/audio/include/chunk.hpp
+++ b/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,20 +53,32 @@ 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,
   };
 
-/*
- * Reads chunks of data from the given input stream, and invokes the given
- * callback to process each of them in turn.
- * 
- * The callback will be invoked with a byte buffer and its size. The callback
- * should process as much data as it can from this buffer, and then return the
- * number of bytes it was able to read. Any leftover bytes will be added as a
- * prefix to the next chunk.
- *
- * 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;
+  /*
+   * Reads chunks of data from the given input stream, and invokes the given
+   * callback to process each of them in turn.
+   *
+   * The callback will be invoked with a byte buffer and its size. The callback
+   * should process as much data as it can from this buffer, and then return the
+   * number of bytes it was able to read. Any leftover bytes will be added as a
+   * prefix to the next chunk.
+   *
+   * 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 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
+}  // namespace audio
diff --git a/src/audio/include/fatfs_audio_input.hpp b/src/audio/include/fatfs_audio_input.hpp
index 5651419d..0fc2729f 100644
--- a/src/audio/include/fatfs_audio_input.hpp
+++ b/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;
diff --git a/src/audio/include/stream_info.hpp b/src/audio/include/stream_info.hpp
index 2b1429ea..bf5d4c60 100644
--- a/src/audio/include/stream_info.hpp
+++ b/src/audio/include/stream_info.hpp
@@ -9,36 +9,42 @@
 namespace audio {
 
 class StreamInfo {
-  public:
-    enum ParseError {
-      WRONG_TYPE,
-      MISSING_MAP,
-    };
-
-    static auto Create(const uint8_t *buffer, size_t length) -> cpp::result<StreamInfo, ParseError>;
-    StreamInfo(CborValue& map);
-
-    StreamInfo() = default;
-    StreamInfo(const StreamInfo&) = default;
-
-    ~StreamInfo() = default;
-
-    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 SampleRate() const -> const std::optional<uint16_t>& { return sample_rate_; }
-
-    enum EncodeError {
-      OUT_OF_MEMORY,
-    };
-
-    auto WriteToStream(CborEncoder encoder) -> cpp::result<void, EncodeError>;
-  private:
-
-    std::optional<std::string> path_;
-    std::optional<uint8_t> channels_;
-    std::optional<uint8_t> bits_per_sample_;
-    std::optional<uint16_t> sample_rate_;
+ public:
+  enum ParseError {
+    WRONG_TYPE,
+    MISSING_MAP,
+    CBOR_ERROR,
+  };
+
+  static auto Create(const uint8_t* buffer, size_t length)
+      -> cpp::result<StreamInfo, ParseError>;
+  StreamInfo(CborValue& map);
+
+  StreamInfo() = default;
+  StreamInfo(const StreamInfo&) = default;
+
+  ~StreamInfo() = default;
+
+  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 SampleRate() const -> const std::optional<uint16_t>& {
+    return sample_rate_;
+  }
+
+  enum EncodeError {
+    OUT_OF_MEMORY,
+  };
+
+  auto WriteToMap(CborEncoder encoder) -> cpp::result<size_t, EncodeError>;
+
+ private:
+  std::optional<std::string> path_;
+  std::optional<uint8_t> channels_;
+  std::optional<uint8_t> bits_per_sample_;
+  std::optional<uint16_t> sample_rate_;
 };
 
-} // namespace audio
+}  // namespace audio
diff --git a/src/audio/include/stream_message.hpp b/src/audio/include/stream_message.hpp
index f59aba8d..2791dcd8 100644
--- a/src/audio/include/stream_message.hpp
+++ b/src/audio/include/stream_message.hpp
@@ -1,11 +1,11 @@
 #pragma once
 
 namespace audio {
-  
-  enum MessageType {
-    TYPE_UNKNOWN,
-    TYPE_CHUNK_HEADER,
-    TYPE_STREAM_INFO,
-  };
 
-} // namespace audio
+enum MessageType {
+  TYPE_UNKNOWN,
+  TYPE_CHUNK_HEADER,
+  TYPE_STREAM_INFO,
+};
+
+}  // namespace audio
diff --git a/src/audio/stream_info.cpp b/src/audio/stream_info.cpp
index bb9b1fa2..6011571d 100644
--- a/src/audio/stream_info.cpp
+++ b/src/audio/stream_info.cpp
@@ -1,45 +1,26 @@
 #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 {
 
-  static const char* kKeyPath = "p";
-  static const char* kKeyChannels = "c";
-  static const char* kKeyBitsPerSample = "b";
-  static const char* kKeySampleRate = "r";
-
-  static auto find_uint64(CborValue &map, char *key) -> cpp::optional<uint64_t> {
-    CborValue val;
-    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 const char* kKeyPath = "p";
+static const char* kKeyChannels = "c";
+static const char* kKeyBitsPerSample = "b";
+static const char* kKeySampleRate = "r";
 
-static auto StreamInfo::Create(const uint8_t *buffer, size_t length) -> cpp::result<StreamInfo, ParseError> {
+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)
-      || !cbor_value_get_integer(&value, &type)
-      || type != STREAM_INFO) {
+  if (!cbor_value_is_integer(&value) ||
+      !cbor_value_get_integer(&value, &type) || 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);
-  bits_per_sample_ = find_uint64(map, kKeyBitsPerSample);
-  sample_rate_ = find_uint64(map, kKeySampleRate);
-
-  CborValue val;
-  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);
-  }
+  cbor::MapDecoder decoder(map);
+  channels_ = decoder.FindValue(kKeyChannels);
+  bits_per_sample_ = decoder.FindValue(kKeyBitsPerSample);
+  sample_rate_ = decoder.FindValue(kKeySampleRate);
+  path_ = decoder.FindValue(kKeyPath);
 }
 
-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);
-
-  write_uint64(&map, kKeyChannels, channels_);
-  write_uint64(&map, kKeyBitsPerSample, bits_per_sample_);
-  write_uint64(&map, kKeySampleRate, sample_rate_);
-
-  if (path_) {
-    cbor_encode_text_string(&map, path_->c_str(), path_->size());
-  }
-
-  cbor_encoder_close_container(&encoder, &map);
+  map_encoder.WriteKeyValue(kKeyChannels, channels_);
+  map_encoder.WriteKeyValue(kKeyBitsPerSample, bits_per_sample_);
+  map_encoder.WriteKeyValue(kKeySampleRate, sample_rate_);
+  map_encoder.WriteKeyValue(kKeyPath, path_);
+  return map_encoder.Finish();
 }
 
-} // namespace audio
+}  // namespace audio
diff --git a/src/cbor/cbor_decoder.cpp b/src/cbor/cbor_decoder.cpp
index f0b497b3..20696afb 100644
--- a/src/cbor/cbor_decoder.cpp
+++ b/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 {};
@@ -73,7 +107,7 @@ auto MapDecoder::FindUnsigned(const std::string &key) -> std::optional<uint32_t>
   return ret;
 }
 
-auto MapDecoder::FindSigned(const std::string &key) -> std::optional<int32_t> {
+auto MapDecoder::FindSigned(const std::string& key) -> std::optional<int32_t> {
   CborValue val;
   if (error_ != CborNoError) {
     return {};
@@ -94,4 +128,4 @@ auto MapDecoder::FindSigned(const std::string &key) -> std::optional<int32_t> {
   return ret;
 }
 
-} // namespace cbor
+}  // namespace cbor
diff --git a/src/cbor/cbor_encoder.cpp b/src/cbor/cbor_encoder.cpp
index 863597b4..6940917e 100644
--- a/src/cbor/cbor_encoder.cpp
+++ b/src/cbor/cbor_encoder.cpp
@@ -4,35 +4,41 @@
 
 namespace cbor {
 
-  static const int kEncoderFlags = 0;
+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
-
+}  // namespace cbor
diff --git a/src/cbor/include/cbor_decoder.hpp b/src/cbor/include/cbor_decoder.hpp
index 39151ca8..258d7c0e 100644
--- a/src/cbor/include/cbor_decoder.hpp
+++ b/src/cbor/include/cbor_decoder.hpp
@@ -5,121 +5,138 @@
 
 namespace cbor {
 
-  static auto parse_stdstring(CborValue *val, std::string *out) -> CborError {
-        uint8_t *buf; size_t len;
-        CborError err = cbor_value_dup_byte_string(val, &buf, &len, NULL);
-        if (err != CborNoError) {
-          return err;
-        }
-        *out = std::move(std::string(buf, len));
-        free(buf);
-        return err
+static auto parse_stdstring(CborValue* val, std::string* out) -> CborError {
+  uint8_t* buf;
+  size_t len;
+  CborError err = cbor_value_dup_byte_string(val, &buf, &len, NULL);
+  if (err != CborNoError) {
+    return err;
   }
+  *out = std::move(std::string(buf, len));
+  free(buf);
+  return err
+}
 
-  class ArrayDecoder {
-    public:
-      static auto Create(uint8_t *buffer, size_t buffer_len)
-        -> cpp::result<std::unique_ptr<ArrayDecoder>, CborError>;
-
-      template<typename T>
-      auto NextValue() -> cpp::result<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> {
-        return NextValue(&cbor_value_is_unsigned_integer, &cbor_value_get_uint64);
-      }
-      template<> auto NextValue() -> cpp::result<std::string, CborError> {
-        return NextValue(&cbor_value_is_byte_string, &parse_stdstring);
-      }
-
-      template <typename T>
-      auto NextValue(
-          bool(*is_valid)(CborValue*),
-          CborError(*parse)(CborValue*, T*)) -> cpp::result<T, CborError> {
-        if (error_ != CborNoError) {
-          return cpp::fail(error_);
-        }
-        if (!is_valid(&it_)) {
-          error_ = CborErrorIllegalType;
-          return cpp::fail(error_);
-        }
-        T ret;
-        error_ = parse(&it_, &ret);
-        if (error_ != CborNoError) {
-          return cpp::fail(error_);
-        }
-        error_ = cbor_value_advance(&it_);
-        if (error_ != CborNoError) {
-          return cpp::fail(error_);
-        }
-        return ret;
-      }
-
-      auto Failed() -> CborError { return error_; }
-
-      ArrayDecoder(const ArrayDecoder&) = delete;
-      ArrayDecoder& operator=(const ArrayDecoder&) = delete;
-    private:
-      CborParser parser_;
-      CborValue root_;
-
-      CborValue it_;
-      CborError error_ = CborNoError;
-  };
-
-  class MapDecoder {
-    public:
-      static auto Create(uint8_t *buffer, size_t buffer_len) -> 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> {
-        return FindValue(key, &cbor_value_is_integer, &cbor_value_get_int);
-      }
-      template<> 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> {
-        return FindValue(key, &cbor_value_is_byte_string, &parse_stdstring);
-      }
-
-      template <typename T>
-      auto FindValue(
-          const std::string &key,
-          bool(*is_valid)(CborValue*),
-          CborError(*parse)(CborValue*, T*)) -> std::optional<T> {
-        if (error_ != CborNoError) {
-          return {};
-        }
-        if (cbor_value_map_find_value(&it_, key.c_str(), &val) != CborNoError) {
-          return {};
-        }
-        if (!is_valid(&val)) {
-          error_ = CborErrorIllegalType;
-          return {};
-        }
-        T ret;
-        error_ = parse(&val, &ret);
-        if (error_ != CborNoError) {
-          return cpp::fail(error_);
-        }
-        return ret;
-      }
-
-      auto Failed() -> CborError { return error_; }
-
-      MapDecoder(const MapDecoder&) = delete;
-      MapDecoder& operator=(const MapDecoder&) = delete;
-    private:
-      CborParser parser_;
-      CborValue root_;
-
-      CborValue it_;
-      CborError error_ = CborNoError;
-  };
-
-
-} // namespace cbor
+class ArrayDecoder {
+ public:
+  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> {
+    return NextValue(&cbor_value_is_integer, &cbor_value_get_int);
+  }
+  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> {
+    return NextValue(&cbor_value_is_byte_string, &parse_stdstring);
+  }
+
+  template <typename T>
+  auto NextValue(bool (*is_valid)(CborValue*),
+                 CborError (*parse)(CborValue*, T*))
+      -> cpp::result<T, CborError> {
+    if (error_ != CborNoError) {
+      return cpp::fail(error_);
+    }
+    if (!is_valid(&it_)) {
+      error_ = CborErrorIllegalType;
+      return cpp::fail(error_);
+    }
+    T ret;
+    error_ = parse(&it_, &ret);
+    if (error_ != CborNoError) {
+      return cpp::fail(error_);
+    }
+    error_ = cbor_value_advance(&it_);
+    if (error_ != CborNoError) {
+      return cpp::fail(error_);
+    }
+    return ret;
+  }
+
+  auto Failed() -> CborError { return error_; }
+
+  auto Iterator() -> CborValue& { return it_; }
+
+  ArrayDecoder(const ArrayDecoder&) = delete;
+  ArrayDecoder& operator=(const ArrayDecoder&) = delete;
+
+ private:
+  CborParser parser_;
+  CborValue root_;
+
+  CborValue it_;
+  CborError error_ = CborNoError;
+};
+
+class MapDecoder {
+ public:
+  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> {
+    return FindValue(key, &cbor_value_is_integer, &cbor_value_get_int);
+  }
+  template <>
+  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> {
+    return FindValue(key, &cbor_value_is_byte_string, &parse_stdstring);
+  }
+
+  template <typename T>
+  auto FindValue(const std::string& key,
+                 bool (*is_valid)(CborValue*),
+                 CborError (*parse)(CborValue*, T*)) -> std::optional<T> {
+    if (error_ != CborNoError) {
+      return {};
+    }
+    if (cbor_value_map_find_value(&it_, key.c_str(), &val) != CborNoError) {
+      return {};
+    }
+    if (!is_valid(&val)) {
+      error_ = CborErrorIllegalType;
+      return {};
+    }
+    T ret;
+    error_ = parse(&val, &ret);
+    if (error_ != CborNoError) {
+      return cpp::fail(error_);
+    }
+    return ret;
+  }
+
+  auto Failed() -> CborError { return error_; }
+
+  MapDecoder(const MapDecoder&) = delete;
+  MapDecoder& operator=(const MapDecoder&) = delete;
+
+ private:
+  CborParser parser_;
+  CborValue root_;
+
+  CborValue it_;
+  CborError error_ = CborNoError;
+};
+
+}  // namespace cbor
diff --git a/src/cbor/include/cbor_encoder.hpp b/src/cbor/include/cbor_encoder.hpp
index 0edbbdff..9479a3b6 100644
--- a/src/cbor/include/cbor_encoder.hpp
+++ b/src/cbor/include/cbor_encoder.hpp
@@ -4,27 +4,34 @@
 #include "esp-idf/components/cbor/tinycbor/src/cbor.h"
 namespace cbor {
 
-  class Encoder {
-    public:
-      enum ContainerType {
-        CONTAINER_ARRAY,
-        CONTAINER_MAP
-      };
-      Encoder(ContainerType type, uint32_t container_len, uint8_t *buffer, size_t buffer_len);
-
-      auto WriteString(const std::string &val) -> void;
-      auto WriteUnsigned(uint32_t val) -> void;
-      auto WriteSigned(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_;
-
-      CborError error_ = CborNoError;
-  };
-
-} // namespace cbor
+class Encoder {
+ public:
+  enum ContainerType { CONTAINER_ARRAY, CONTAINER_MAP };
+  Encoder(ContainerType type,
+          uint32_t container_len,
+          uint8_t* buffer,
+          size_t buffer_len);
+
+  template <typename T>
+  auto WriteKeyValue(const std::string& key, const T& val) -> void {
+    WriteValue(key);
+    WriteValue(val);
+  }
+
+  auto WriteValue(const std::string& val) -> void;
+  auto WriteValue(uint32_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_;
+
+  CborError error_ = CborNoError;
+};
+
+}  // namespace cbor
diff --git a/src/codecs/CMakeLists.txt b/src/codecs/CMakeLists.txt
index 522aa885..4a8918ae 100644
--- a/src/codecs/CMakeLists.txt
+++ b/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})
diff --git a/src/codecs/codec.cpp b/src/codecs/codec.cpp
index 70cfe10a..db8d69b9 100644
--- a/src/codecs/codec.cpp
+++ b/src/codecs/codec.cpp
@@ -2,8 +2,9 @@
 
 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);
 }
 
-} // namespace codecs
+}  // namespace codecs
diff --git a/src/codecs/include/codec.hpp b/src/codecs/include/codec.hpp
index 99e786d5..8e82bd71 100644
--- a/src/codecs/include/codec.hpp
+++ b/src/codecs/include/codec.hpp
@@ -8,51 +8,53 @@
 
 namespace codecs {
 
-  enum CreateCodecError {
-    UNKNOWN_EXTENSION
+enum CreateCodecError { UNKNOWN_EXTENSION };
+
+auto CreateCodecForFile(const std::string& extension)
+    -> cpp::result<std::unique_ptr<ICodec>, CreateCodecError>;
+
+class ICodec {
+ public:
+  virtual ~ICodec() {}
+
+  virtual auto CanHandleFile(const std::string& path) -> bool = 0;
+
+  struct OutputFormat {
+    uint8_t num_channels;
+    uint8_t bits_per_sample;
+    int sample_rate_hz;
   };
 
-  auto CreateCodecForExtension(std::string extension) -> cpp::result<std::unique_ptr<ICodec>, CreateCodecError>;
-
-  class ICodec {
-    public:
-      virtual ~ICodec() {}
-
-      virtual auto CanHandleExtension(std::string extension) -> bool = 0;
-
-      struct OutputFormat {
-        uint8_t num_channels;
-        uint8_t bits_per_sample;
-        int sample_rate_hz;
-      };
-
-      virtual auto GetOutputFormat() -> OutputFormat = 0;
-
-      enum Error {};
-
-      struct Result {
-        bool need_more_input;
-        /*
-         * For need_more_input, this is how far we got in the input buffer
-         * before we were unable to process more data. Any remaining data in the
-         * buffer should be moved to the start before the next call.
-         */
-          std::size_t input_processed;
-
-        bool flush_output;
-         /*
-         * For flush_output, this is how far we got in the output buffer before
-         * we ran out of space for samples. The caller should flush this many
-         * bytes downstream.
-         */
-          std::size_t output_written;
-      };
-
-      virtual auto Process(
-          uint8_t *input,
-          std::size_t input_len,
-          uint8_t *output,
-	  std::size_t output_length) -> cpp::result<Result, Error>  = 0;
+  virtual auto GetOutputFormat() -> OutputFormat = 0;
+
+  enum ProcessingError {};
+
+  struct Result {
+    bool need_more_input;
+    /*
+     * For need_more_input, this is how far we got in the input buffer
+     * before we were unable to process more data. Any remaining data in the
+     * buffer should be moved to the start before the next call.
+     */
+    std::size_t input_processed;
+
+    bool flush_output;
+    /*
+     * For flush_output, this is how far we got in the output buffer before
+     * we ran out of space for samples. The caller should flush this many
+     * bytes downstream.
+     */
+    std::size_t output_written;
   };
 
-} // namespace codecs
+  virtual auto ResetForNewStream() -> void = 0;
+
+  virtual auto SetInput(uint8_t* buffer, std::size_t length) = 0;
+  virtual auto Process(uint8_t* output, std::size_t output_length)
+      -> cpp::result<size_t, Error> = 0;
+
+  virtual auto GetOutputProcessed() -> std::size_t;
+  = 0;
+};
+
+}  // namespace codecs
diff --git a/src/codecs/include/mad.hpp b/src/codecs/include/mad.hpp
index cfe4eab7..241ea6c3 100644
--- a/src/codecs/include/mad.hpp
+++ b/src/codecs/include/mad.hpp
@@ -4,23 +4,25 @@
 
 namespace codecs {
 
-  class MadMp3Decoder : public ICodec {
-    public:
-      MadMp3Decoder();
-      ~MadMp3Decoder();
+class MadMp3Decoder : public ICodec {
+ public:
+  MadMp3Decoder();
+  ~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 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;
 
-    private:
-      mad_stream stream_;
-      mad_frame	frame_;
-      mad_synth	synth_;
+ private:
+  mad_stream stream_;
+  mad_frame frame_;
+  mad_synth synth_;
 
-      mad_header header_;
-      bool has_decoded_header_;
+  mad_header header_;
+  bool has_decoded_header_;
 
-      int current_sample_ = -1;
-  };
+  int current_sample_ = -1;
+};
 
-} // namespace codecs
+}  // namespace codecs
diff --git a/src/codecs/include/types.hpp b/src/codecs/include/types.hpp
index 8525a136..a962cf68 100644
--- a/src/codecs/include/types.hpp
+++ b/src/codecs/include/types.hpp
@@ -2,11 +2,11 @@
 
 #include <string>
 
-namespace codecs  {
+namespace codecs {
 
-  enum StreamType {
-    STREAM_MP3,
-  };
+enum StreamType {
+  STREAM_MP3,
+};
 
-  auto GetStreamTypeFromFilename(std::string filename);
-}
+auto GetStreamTypeFromFilename(std::string filename);
+}  // namespace codecs
diff --git a/src/codecs/mad.cpp b/src/codecs/mad.cpp
index f1ed0346..c918e849 100644
--- a/src/codecs/mad.cpp
+++ b/src/codecs/mad.cpp
@@ -5,150 +5,145 @@
 
 namespace codecs {
 
-  static int32_t scaleTo24Bits(mad_fixed_t sample) {
-    // Round the bottom bits.
-    sample += (1L << (MAD_F_FRACBITS - 24));
-
-    // Clip the leftover bits to within range.
-    if (sample >= MAD_F_ONE)
-      sample = MAD_F_ONE - 1;
-    else if (sample < -MAD_F_ONE)
-      sample = -MAD_F_ONE;
-
-    /* quantize */
-    return sample >> (MAD_F_FRACBITS + 1 - 24);
+static int32_t scaleTo24Bits(mad_fixed_t sample) {
+  // Round the bottom bits.
+  sample += (1L << (MAD_F_FRACBITS - 24));
+
+  // Clip the leftover bits to within range.
+  if (sample >= MAD_F_ONE)
+    sample = MAD_F_ONE - 1;
+  else if (sample < -MAD_F_ONE)
+    sample = -MAD_F_ONE;
+
+  /* quantize */
+  return sample >> (MAD_F_FRACBITS + 1 - 24);
+}
+
+MadMp3Decoder::MadMp3Decoder() {
+  mad_stream_init(&stream_);
+  mad_frame_init(&frame_);
+  mad_synth_init(&synth_);
+  mad_header_init(&header_);
+}
+MadMp3Decoder::~MadMp3Decoder() {
+  mad_stream_finish(&stream_);
+  mad_frame_finish(&frame_);
+  mad_synth_finish(&synth_);
+  mad_header_finish(&header_);
+}
+
+auto MadMp3Decoder::CanHandleExtension(std::string extension) -> bool {
+  return extension == "mp3";
+}
+
+auto GetOutputFormat() -> OutputFormat {
+  return OutputFormat{
+      .num_channels = synth_.pcm.channels,
+      .bits_per_sample = 24,
+      .sample_rate_hz = synth_.pcm.samplerate,
+  };
+}
+
+auto MadMp3Decoder::Process(uint8_t* input,
+                            std::size_t input_len,
+                            uint8_t* output,
+                            std::size_t output_length)
+    -> cpp::result<Result, Error> {
+  Result res {
+    .need_more_input = false, .input_processed = 0, .flush_output = false,
+    .output_written = 0,
   }
-
-  MadMp3Decoder::MadMp3Decoder() {
-    mad_stream_init(&stream_);
-    mad_frame_init(&frame_);
-    mad_synth_init(&synth_);
-    mad_header_init(&header_);
-  }
-  MadMp3Decoder::~MadMp3Decoder() {
-    mad_stream_finish(&stream_);
-    mad_frame_finish(&frame_);
-    mad_synth_finish(&synth_);
-    mad_header_finish(&header_);
+  while (true) {
+    // Only process more of the input if we're done sending off the
+    // samples for the previous frame.
+    if (current_sample_ == -1) {
+      ProcessInput(&res, input, input_len);
+    }
+
+    // Write PCM samples to the output buffer. This always needs to be
+    // done, even if we ran out of input, so that we don't keep the last
+    // few samples buffered if the input stream has actually finished.
+    WriteOutputSamples(&res, output, output_length);
+
+    if (res.need_more_input || res.flush_output) {
+      return res;
+    }
   }
 
-      auto MadMp3Decoder::CanHandleExtension(std::string extension) -> bool {
-      return extension == "mp3";
+  auto MadMp3Decoder::ProcessInput(Result * res, uint8_t * input,
+                                   std::size_t input_len)
+      ->void {
+    if (input != stream_.buffer) {
+      mad_stream_buffer(&stream_, input, input_len);
+    }
+
+    if (!has_decoded_header_) {
+      // The header of any given frame should be representative of the
+      // entire stream, so only need to read it once.
+      mad_header_decode(&header_, &stream);
+      has_decoded_header_ = true;
+
+      // TODO: Use the info in the header for something. I think the
+      // 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) {
+      if (MAD_RECOVERABLE(stream_.error)) {
+        // Recoverable errors are usually malformed parts of the stream.
+        // We can recover from them by just retrying the decode.
+        continue;
       }
 
-      auto GetOutputFormat() -> OutputFormat {
-        return OutputFormat {
-          .num_channels = synth_.pcm.channels,
-          .bits_per_sample = 24,
-          .sample_rate_hz = synth_.pcm.samplerate,
-        };
+      if (stream_.error = MAD_ERROR_BUFLEN) {
+        // The decoder ran out of bytes before it completed a frame. We
+        // need to return back to the caller to give us more data. Note
+        // that there might still be some unused data in the input, so we
+        // should calculate that amount and return it.
+        size_t remaining_bytes = stream.bufend - stream_.next_frame;
+        return remaining_bytes;
       }
 
-      auto MadMp3Decoder::Process(
-          uint8_t *input,
-          std::size_t input_len,
-          uint8_t *output,
-	  std::size_t output_length) -> cpp::result<Result, Error> {
-
-        Result res {
-          .need_more_input = false,
-          .input_processed = 0,
-          .flush_output = false,
-          .output_written = 0,
-        }
-        while (true) {
-          // Only process more of the input if we're done sending off the
-          // samples for the previous frame.
-          if (current_sample_ == -1) {
-            ProcessInput(&res, input, input_len);
-          }
-
-          // Write PCM samples to the output buffer. This always needs to be
-          // done, even if we ran out of input, so that we don't keep the last
-          // few samples buffered if the input stream has actually finished.
-          WriteOutputSamples(&res, output, output_length);
-
-          if (res.need_more_input || res.flush_output) {
-            return res;
-          }
-      }
+      // The error is unrecoverable. Give up.
+      return cpp::fail(MALFORMED_DATA);
+    }
 
-      auto MadMp3Decoder::ProcessInput(
-          Result *res, uint8_t *input, std::size_t input_len) -> void {
-
-        if (input != stream_.buffer) {
-          mad_stream_buffer(&stream_, input, input_len);
-        }
-
-          if (!has_decoded_header_) {
-            // The header of any given frame should be representative of the
-            // entire stream, so only need to read it once.
-            mad_header_decode(&header_, &stream);
-            has_decoded_header_ = true;
-
-            // TODO: Use the info in the header for something. I think the
-            // 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) {
-            if (MAD_RECOVERABLE(stream_.error)) {
-              // Recoverable errors are usually malformed parts of the stream.
-              // We can recover from them by just retrying the decode.
-              continue;
-            }
-
-            if (stream_.error = MAD_ERROR_BUFLEN) {
-              // The decoder ran out of bytes before it completed a frame. We
-              // need to return back to the caller to give us more data. Note
-              // that there might still be some unused data in the input, so we
-              // should calculate that amount and return it.
-              size_t remaining_bytes = stream.bufend - stream_.next_frame;
-              return remaining_bytes;
-            }
-
-            // The error is unrecoverable. Give up.
-            return cpp::fail(MALFORMED_DATA);
-          }
-
-          // We've successfully decoded a frame!
-          // Now we need to synthesize PCM samples based on the frame, and send
-          // them downstream.
-          mad_synth_frame(&synth_, &frame_);
-          up_to_sample = 0;
-      }
+    // We've successfully decoded a frame!
+    // Now we need to synthesize PCM samples based on the frame, and send
+    // them downstream.
+    mad_synth_frame(&synth_, &frame_);
+    up_to_sample = 0;
+  }
 
-      auto MadMp3Decoder::WriteOutputSamples(
-          Result *res,
-          uint8_t *output,
-	  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) {
-          return;
-        }
-        res->flush_output = true;
-
-        while (current_sample_ < synth_.pcm.length) {
-          if (output_byte + (3 * synth_.pcm.channels) >= output_length) {
-            res->output_written = output_byte;
-            return;
-          }
-
-          for (int channel = 0; channel < synth_.pcm.channels; channel++) {
-            uint32_t sample_24 = scaleTo24Bits(synth_.pcm.samples[channel][sample]);
-            output[output_byte++] = (sample_24 >> 0) & 0xff;
-            output[output_byte++] = (sample_24 >> 8) & 0xff;
-            output[output_byte++] = (sample_24 >> 16) & 0xff;
-          }
-          current_sample_++;
-        }
-
-        // We wrote everything! Reset, ready for the next frame.
-        current_sample_ = -1;
+  auto MadMp3Decoder::WriteOutputSamples(Result * res, uint8_t * output,
+                                         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) {
+      return;
+    }
+    res->flush_output = true;
+
+    while (current_sample_ < synth_.pcm.length) {
+      if (output_byte + (3 * synth_.pcm.channels) >= output_length) {
         res->output_written = output_byte;
+        return;
       }
 
-} // namespace codecs
+      for (int channel = 0; channel < synth_.pcm.channels; channel++) {
+        uint32_t sample_24 = scaleTo24Bits(synth_.pcm.samples[channel][sample]);
+        output[output_byte++] = (sample_24 >> 0) & 0xff;
+        output[output_byte++] = (sample_24 >> 8) & 0xff;
+        output[output_byte++] = (sample_24 >> 16) & 0xff;
+      }
+      current_sample_++;
+    }
+
+    // We wrote everything! Reset, ready for the next frame.
+    current_sample_ = -1;
+    res->output_written = output_byte;
+  }
+
+}  // namespace codecs
diff --git a/src/drivers/include/fatfs_audio_input.hpp b/src/drivers/include/fatfs_audio_input.hpp
index 3753c136..110651d8 100644
--- a/src/drivers/include/fatfs_audio_input.hpp
+++ b/src/drivers/include/fatfs_audio_input.hpp
@@ -2,39 +2,38 @@
 
 namespace drivers {
 
-  class FatfsAudioInput {
-    public:
-      FatfsAudioInput(std::shared_ptr<SdStorage> storage);
-      ~FatfsAudioInput();
-
-      enum Status {
-        /*
-         * Successfully read data into the output buffer, and there is still
-         * data remaining in the file.
-         */
-        OKAY,
-
-        /*
-         * The ringbuffer was full. No data was read.
-         */
-        RINGBUF_FULL,
-
-        /*
-         * Some data may have been read into the output buffer, but the file is
-         * now empty.
-         */
-        FILE_EMPTY,
-      };
-      auto Process() -> Status;
-
-      auto GetOutputBuffer() -> RingbufHandle_t;
-
-    private:
-      std::shared_ptr<SdStorage> storage_;
-      RingbufHandle_t output_;
-
-      std::string path_;
-
+class FatfsAudioInput {
+ public:
+  FatfsAudioInput(std::shared_ptr<SdStorage> storage);
+  ~FatfsAudioInput();
+
+  enum Status {
+    /*
+     * Successfully read data into the output buffer, and there is still
+     * data remaining in the file.
+     */
+    OKAY,
+
+    /*
+     * The ringbuffer was full. No data was read.
+     */
+    RINGBUF_FULL,
+
+    /*
+     * Some data may have been read into the output buffer, but the file is
+     * now empty.
+     */
+    FILE_EMPTY,
   };
- 
-} // namespace drivers
+  auto Process() -> Status;
+
+  auto GetOutputBuffer() -> RingbufHandle_t;
+
+ private:
+  std::shared_ptr<SdStorage> storage_;
+  RingbufHandle_t output_;
+
+  std::string path_;
+};
+
+}  // namespace drivers
diff --git a/src/main/app_console.cpp b/src/main/app_console.cpp
index 765b17d2..fbb8df87 100644
--- a/src/main/app_console.cpp
+++ b/src/main/app_console.cpp
@@ -83,7 +83,7 @@ int CmdToggle(int argc, char** argv) {
     return 1;
   }
 
-  //sInstance->playback_->Toggle();
+  // sInstance->playback_->Toggle();
 
   return 0;
 }
@@ -110,7 +110,7 @@ int CmdVolume(int argc, char** argv) {
     return 1;
   }
 
-  //sInstance->playback_->SetVolume((uint8_t)raw_vol);
+  // sInstance->playback_->SetVolume((uint8_t)raw_vol);
 
   return 0;
 }
diff --git a/src/main/app_console.hpp b/src/main/app_console.hpp
index 9cd9d50c..d41810b0 100644
--- a/src/main/app_console.hpp
+++ b/src/main/app_console.hpp
@@ -2,15 +2,15 @@
 
 #include <memory>
 
-#include "storage.hpp"
 #include "console.hpp"
+#include "storage.hpp"
 
 namespace console {
 
 class AppConsole : public Console {
  public:
-  AppConsole() {};
-  virtual ~AppConsole() {};
+  AppConsole(){};
+  virtual ~AppConsole(){};
 
  protected:
   virtual auto RegisterExtraComponents() -> void;
diff --git a/src/main/main.cpp b/src/main/main.cpp
index 62cb430e..79a2f42b 100644
--- a/src/main/main.cpp
+++ b/src/main/main.cpp
@@ -135,7 +135,7 @@ extern "C" void app_main(void) {
   console.Launch();
 
   while (1) {
-    //playback->ProcessEvents(5);
+    // playback->ProcessEvents(5);
     vTaskDelay(pdMS_TO_TICKS(100));
   }
 }
diff --git a/src/tasks/CMakeLists.txt b/src/tasks/CMakeLists.txt
new file mode 100644
index 00000000..0503d293
--- /dev/null
+++ b/src/tasks/CMakeLists.txt
@@ -0,0 +1,2 @@
+idf_component_register(SRCS "tasks.cpp" INCLUDE_DIRS ".")
+target_compile_options(${COMPONENT_LIB} PRIVATE ${EXTRA_WARNINGS})
diff --git a/src/tasks/tasks.cpp b/src/tasks/tasks.cpp
new file mode 100644
index 00000000..34100353
--- /dev/null
+++ b/src/tasks/tasks.cpp
@@ -0,0 +1,4 @@
+#include "tasks.cpp"
+
+static const UBaseType_t kTaskPriorityLvgl = 4;
+static const UBaseType_t kTaskPriorityAudio = 5;
diff --git a/src/tasks/tasks.hpp b/src/tasks/tasks.hpp
new file mode 100644
index 00000000..24f8509a
--- /dev/null
+++ b/src/tasks/tasks.hpp
@@ -0,0 +1,6 @@
+#pragma once
+
+#include "freertos/portmacro.h"
+
+extern const UBaseType_t kTaskPriorityLvgl;
+extern const UBaseType_t kTaskPriorityAudio;