SynchronizedNBufferRTOS.h

Go to the documentation of this file.

#pragma once
#include "AudioToolsConfig.h"
#include "QueueRTOS.h"
 
#define DEFAULT_BUFFER_WAIT 1000
 
namespace audio_tools {
 
template <typename T>
class SynchronizedNBufferRTOST : public NBuffer<T> {
 public:
  SynchronizedNBufferRTOST(int bufferSize, int bufferCount,
                            TickType_t writeMaxWait = DEFAULT_BUFFER_WAIT,
                            TickType_t readMaxWait = DEFAULT_BUFFER_WAIT) {
    TRACED();
    write_max_wait_ = writeMaxWait;
    read_max_wait_ = readMaxWait;
    resize(bufferSize, bufferCount);
  }
 
  ~SynchronizedNBufferRTOST() { cleanup(); }
 
  bool resize(size_t bufferSize, int bufferCount) {
    TRACED();
    if (NBuffer<T>::buffer_size == (int)bufferSize &&
        NBuffer<T>::buffer_count == bufferCount) {
      return true;
    }
    LOGW("resize: bufferSize=%d, bufferCount=%d", (int)bufferSize, bufferCount);
 
    // Update parent's tracking fields
    NBuffer<T>::buffer_count = bufferCount;
    NBuffer<T>::buffer_size = bufferSize;
 
    // Clear held pointers so we don't dangle after cleanup
    NBuffer<T>::actual_write_buffer = nullptr;
    NBuffer<T>::actual_read_buffer = nullptr;
 
    // Free old blocks and resize the RTOS queues
    cleanup();
    available_queue_.resize(bufferCount);
    filled_queue_.resize(bufferCount);
    applyWaitTimes();
 
    // Allocate new blocks into the available queue
    for (int j = 0; j < bufferCount; j++) {
      BaseBuffer<T>* tmp = new SingleBuffer<T>(bufferSize);
      if (tmp == nullptr) {
        LOGE("Not enough memory for buffer %d", j);
        return false;
      }
      available_queue_.enqueue(tmp);
    }
    return true;
  }
 
  void setWriteMaxWait(TickType_t ticks) {
    write_max_wait_ = ticks;
    applyWaitTimes();
  }
 
  void setReadMaxWait(TickType_t ticks) {
    read_max_wait_ = ticks;
    applyWaitTimes();
  }
 
  size_t size() { return NBuffer<T>::buffer_size * NBuffer<T>::buffer_count; }
 
  int bufferCountFilled() { return filled_queue_.size(); }
 
  int bufferCountEmpty() { return available_queue_.size(); }
 
 protected:
  QueueRTOS<BaseBuffer<T>*> available_queue_{0, DEFAULT_BUFFER_WAIT, 0};
  QueueRTOS<BaseBuffer<T>*> filled_queue_{0, DEFAULT_BUFFER_WAIT, 0};
  TickType_t write_max_wait_ = DEFAULT_BUFFER_WAIT;
  TickType_t read_max_wait_ = DEFAULT_BUFFER_WAIT;
 
  void applyWaitTimes() {
    available_queue_.setReadMaxWait(write_max_wait_);   // writer gets free block
    available_queue_.setWriteMaxWait(read_max_wait_);   // reader returns drained block
    filled_queue_.setWriteMaxWait(write_max_wait_);     // writer submits filled block
    filled_queue_.setReadMaxWait(read_max_wait_);       // reader gets filled block
  }
 
  void cleanup() {
    TRACED();
    BaseBuffer<T>* buffer = nullptr;
    while (available_queue_.dequeue(buffer)) {
      delete buffer;
    }
    while (filled_queue_.dequeue(buffer)) {
      delete buffer;
    }
  }
 
  // ── Virtual overrides that route NBuffer's block management through
  //    FreeRTOS queues instead of the non-thread-safe QueueFromVector. ────
 
  BaseBuffer<T>* getNextAvailableBuffer() override {
    TRACED();
    BaseBuffer<T>* result = nullptr;
    return available_queue_.dequeue(result) ? result : nullptr;
  }
 
  bool addAvailableBuffer(BaseBuffer<T>* buffer) override {
    TRACED();
    return available_queue_.enqueue(buffer);
  }
 
  BaseBuffer<T>* getNextFilledBuffer() override {
    TRACED();
    BaseBuffer<T>* result = nullptr;
    return filled_queue_.dequeue(result) ? result : nullptr;
  }
 
  bool addFilledBuffer(BaseBuffer<T>* buffer) override {
    TRACED();
    return filled_queue_.enqueue(buffer);
  }
};
 
using SynchronizedNBufferRTOS = SynchronizedNBufferRTOST<uint8_t>;
 
using SynchronizedNBuffer = SynchronizedNBufferRTOS;
 
}  // namespace audio_tools