testing BufferedStream

Author: pschatzmann

examples/tests/basic/test-buffered-stream/test-buffered-stream.ino

@@ -0,0 +1,70 @@
+/**
+ * Testing peek and readBytes of BufferedStream
+ * - fill DynamicMemoryStream with data
+ * - check data of DynamicMemoryStream
+ * - peek data
+ * - check data vis BufferedStream
+ */
+#include "AudioTools.h"
+
+DynamicMemoryStream data;
+BufferedStream buffered(data);
+HexDumpOutput out;
+const int processingSize = 1024;
+uint8_t buffer[processingSize];
+
+void fillData() {
+  Serial.println("Filling data...");
+  for (int j = 0; j < processingSize; j++) {
+    buffer[j] = j % 256;
+  }
+  while (data.size() < processingSize * 10) {
+    assert(data.write(buffer, processingSize) == processingSize);
+  }
+}
+
+void testPeek() {
+  int len = buffered.peekBytes(&buffer[0], 160);
+  Serial.println("Peek:");
+  out.write(buffer, len);
+  Serial.println();
+}
+
+void checkData(Stream &data, bool isPrint = false) {
+  // check data
+  Serial.print("TestingData");
+  int n = data.readBytes(buffer, processingSize);
+  Serial.print(" ");
+  Serial.println(n);
+  while (n == processingSize) {
+    for (int j = 0; j < processingSize; j++) {
+      uint8_t expected = j % 256;
+      if (isPrint) {
+        Serial.print(buffer[j]);
+        Serial.print(", ");
+        Serial.println(expected);
+      }
+      assert(buffer[j] == expected);
+      n = data.readBytes(buffer, processingSize);
+      assert(n == 0 || n == processingSize);
+    }
+  }
+  Serial.println("test OK!");
+}
+
+void setup() {
+  Serial.begin(115200);
+  AudioToolsLogger.begin(Serial, AudioToolsLogLevel::Info);
+  delay(5000);
+
+  data.begin();
+  fillData();
+  testPeek();
+  checkData(data);
+  
+  data.rewind();
+  checkData(buffered);
+}
+
+void loop() {
+}

src/AudioTools/AudioCodecs/StreamingDecoder.h

@@ -580,6 +580,7 @@ class MultiStreamingDecoder : public StreamingDecoder {
         // Initialize the selected decoder and mark it as active
         assert(p_data_source != nullptr);
         actual_decoder.decoder->setInput(*p_data_source);
+        actual_decoder.decoder->clearNotifyAudioChange();
         actual_decoder.decoder->addNotifyAudioChange(*this);
         if (actual_decoder.decoder->begin()) {
           actual_decoder.is_open = true;
@@ -720,8 +721,7 @@ class MultiStreamingDecoder : public StreamingDecoder {
   Vector<StreamingDecoderAdapter*> adapters{
       0};                      ///< Collection of internally created adapters
   MimeDetector mime_detector;  ///< MIME type detection engine
-  BufferedStream
-      buffered_stream{DEFAULT_BUFFER_SIZE};  ///< Buffered stream for data preservation
+  BufferedStream buffered_stream{0};  ///< Buffered stream for data preservation
   Vector<uint8_t> detection_buffer{0};  ///< Buffer for format detection data
   bool is_first = true;                 ///< Flag for first copy() call
   const char* selected_mime = nullptr;  ///< MIME type that was selected
@@ -779,6 +779,7 @@ class MultiStreamingDecoder : public StreamingDecoder {
       } else {
         // Option 2: Auto-detect MIME type by analyzing stream content
         // Redirect the decoder to use the buffered stream
+        buffered_stream.resize(DEFAULT_BUFFER_SIZE);
         p_data_source = &buffered_stream;
 
         // This requires reading a sample of data to identify the format
@@ -806,9 +807,6 @@ class MultiStreamingDecoder : public StreamingDecoder {
           LOGE("The decoder could not be selected for %s", toStr(mime));
           return false;  // No registered decoder can handle this format
         }
-        // define data source
-        assert(p_data_source!=nullptr);
-        actual_decoder.decoder->setInput(*p_data_source);
       } else {
         // MIME detection failed - format is unknown or unsupported
         LOGE("Could not determine mime type");

src/AudioTools/CoreAudio/AudioStreams.h

@@ -484,31 +484,31 @@ class BufferedStream : public ModifyingStream {
  public:
   BufferedStream(size_t buffer_size) {
     TRACED();
-    buffer.resize(buffer_size);
+    if (buffer_size > 0) resize(buffer_size);
   }
 
   BufferedStream(Print &out, size_t buffer_size = 1024) {
     TRACED();
     setOutput(out);
-    buffer.resize(buffer_size);
+    if (buffer_size > 0) resize(buffer_size);
   }
 
   BufferedStream(Stream &io, size_t buffer_size = 1024) {
     TRACED();
     setStream(io);
-    buffer.resize(buffer_size);
+    if (buffer_size > 0) resize(buffer_size);
   }
 
   BufferedStream(size_t buffer_size, Print &out) {
     TRACED();
     setOutput(out);
-    buffer.resize(buffer_size);
+    if (buffer_size > 0) resize(buffer_size);
   }
 
   BufferedStream(size_t buffer_size, Stream &io) {
     TRACED();
     setStream(io);
-    buffer.resize(buffer_size);
+    if (buffer_size > 0) resize(buffer_size);
   }
 
   void setOutput(Print &out) override { p_out = &out; }
@@ -567,7 +567,7 @@ class BufferedStream : public ModifyingStream {
 
   /// Use this method !!
   size_t readBytes(uint8_t *data, size_t len) override {
-    if (buffer.isEmpty()) {
+    if (buffer.isEmpty() && len >= minReadBufferSize) {
       return readExt(data, len);
     } else {
       refill(len);
@@ -592,10 +592,19 @@ class BufferedStream : public ModifyingStream {
   /// Clears all the data in the buffer
   void clear() { buffer.reset(); }
 
+  /// Resize the buffer
+  void resize(int size) { buffer.resize(size); }
+
+  /// Defines the minimum direct unbuffered read size to the original source (default is 1024)
+  void setMinUnbufferedReadSize(size_t size) {
+    minReadBufferSize = size;
+  }
+
  protected:
-  SingleBuffer<uint8_t> buffer;
+  SingleBuffer<uint8_t> buffer{0};
   Print *p_out = nullptr;
   Stream *p_in = nullptr;
+  size_t minReadBufferSize = 1024;
 
   /// refills the buffer with data from the source
   void refill() {