StreamingDecoder.h

#pragma once
#include <new>
#include "AudioTools/AudioCodecs/AudioCodecsBase.h"
#include "AudioTools/CoreAudio/AudioBasic/StrView.h"
#include "AudioTools/CoreAudio/AudioMetaData/MimeDetector.h"
#include "AudioTools/CoreAudio/AudioOutput.h"
#include "AudioTools/CoreAudio/BaseStream.h"
 
namespace audio_tools {
 
class StreamingDecoder : public AudioInfoSource, public AudioInfoSupport {
 public:
 
  virtual ~StreamingDecoder() = default;
 
  virtual bool begin() = 0;
 
  virtual void end() = 0;
 
  virtual void setOutput(Print& out_stream) { p_print = &out_stream; }
 
  virtual void setOutput(AudioStream& out_stream) {
    Print* p_print = &out_stream;
    setOutput(*p_print);
    addNotifyAudioChange(out_stream);
  }
 
  virtual void setOutput(AudioOutput& out_stream) {
    Print* p_print = &out_stream;
    setOutput(*p_print);
    addNotifyAudioChange(out_stream);
  }
 
  void setInput(Stream& inStream) { this->p_input = &inStream; }
 
  virtual AudioInfo audioInfo() = 0;
 
  virtual operator bool() = 0;
 
  virtual bool copy() = 0;
 
  bool copyAll() {
    bool result = false;
    while (copy()) {
      result = true;
    }
    return result;
  }
 
  virtual const char* mime() = 0;
 
 protected:
  virtual size_t readBytes(uint8_t* data, size_t len) = 0;
 
  void setAudioInfo(AudioInfo newInfo) override {
    TRACED();
    if (this->info != newInfo) {
      this->info = newInfo;
      notifyAudioChange(info);
    }
  }
 
  Print* p_print = nullptr;   
  Stream* p_input = nullptr;  
  AudioInfo info;
};
 
class StreamingDecoderAdapter : public StreamingDecoder {
 public:
  StreamingDecoderAdapter(AudioDecoder& decoder, const char* mimeStr,
                          int copySize = DEFAULT_BUFFER_SIZE) {
    p_decoder = &decoder;
    p_decoder->addNotifyAudioChange(*this);
    mime_str = mimeStr;
    if (copySize > 0) resize(copySize);
  }
 
  bool begin() override { 
    TRACED();
    if (p_decoder == nullptr) return false;
    if (p_input == nullptr) return false;
    return p_decoder->begin(); 
  }
 
  void end() override { p_decoder->end(); }
 
  void setOutput(Print& out_stream) override {
    p_decoder->setOutput(out_stream);
  }
 
  AudioInfo audioInfo() override { return p_decoder->audioInfo(); }
 
  virtual operator bool() override { return *p_decoder; }
 
  virtual bool copy() override {
    int read = readBytes(buffer.data(), buffer.size());
    int written = 0;
    if (read > 0) written = p_decoder->write(&buffer[0], read);
    bool rc = written > 0;
    LOGI("copy: %s", rc ? "success" : "failure");
    return rc;
  }
 
  void resize(int bufferSize) { buffer.resize(bufferSize); }
 
  const char* mime() override { return mime_str; }
 
 protected:
  AudioDecoder* p_decoder = nullptr;  
  Vector<uint8_t> buffer{0};          
  const char* mime_str = nullptr;     
 
  size_t readBytes(uint8_t* data, size_t len) override {
    if (p_input == nullptr) return 0;
    return p_input->readBytes(data, len);
  }
};
 
class MultiStreamingDecoder : public StreamingDecoder {
 public:
  MultiStreamingDecoder() = default;
 
  ~MultiStreamingDecoder() {
    // Clean up any adapters we created
    for (auto* adapter : adapters) {
      delete adapter;
    }
    adapters.clear();    
  }
 
  bool begin() override {
    mime_detector.begin();
    is_first = true;
    if (p_print == nullptr) {
      LOGE("No output defined");
      return false;
    }
    return true;
  }
 
  void end() override {
    if (actual_decoder.decoder != nullptr && actual_decoder.is_open) {
      actual_decoder.decoder->end();
    }
    actual_decoder.is_open = false;
    actual_decoder.decoder = nullptr;
    actual_decoder.mime = nullptr;
    is_first = true;    
  }
 
  void setOutput(Print& out_stream) override {
    StreamingDecoder::setOutput(out_stream);
  }
 
  void setOutput(AudioStream& out_stream) override {
    StreamingDecoder::setOutput(out_stream);
  }
 
  void setOutput(AudioOutput& out_stream) override {
    StreamingDecoder::setOutput(out_stream);
  }
 
  void setInput(Stream& inStream) { 
    StreamingDecoder::setInput(inStream);
  }
 
  void addDecoder(StreamingDecoder& decoder) {
    decoder.addNotifyAudioChange(*this);
    const char* mime = decoder.mime();
    if (mime != nullptr) {
      DecoderInfo info{mime, &decoder};
      decoders.push_back(info);
    } else {
      LOGE("Decoder mime() returned nullptr - cannot add decoder");
    }
  }
 
  void addDecoder(StreamingDecoder& decoder, const char* mime) {
    if (mime != nullptr) {
      decoder.addNotifyAudioChange(*this);
      DecoderInfo info{mime, &decoder};
      decoders.push_back(info);
    } else {
      LOGE("Decoder mime() returned nullptr - cannot add decoder");
    }
  }
 
  void addDecoder(AudioDecoder& decoder, const char* mime,
                  int bufferSize = DEFAULT_BUFFER_SIZE) {
    if (mime != nullptr) {
      // Create a StreamingDecoderAdapter to wrap the AudioDecoder
      decoder.addNotifyAudioChange(*this);
      auto adapter = new StreamingDecoderAdapter(decoder, mime, bufferSize);
      adapters.push_back(adapter);  // Store for cleanup
 
      DecoderInfo info{mime, adapter};
      decoders.push_back(info);
    } else {
      LOGE("MIME type is nullptr - cannot add AudioDecoder");
    }
  }
 
  virtual operator bool() override {
    if (actual_decoder.decoder == nullptr) return false;
    return is_first || actual_decoder.is_open;
  }
 
  virtual bool copy() override {
    if (p_input == nullptr) return false;
 
    // Automatically select decoder if not already selected
    if (is_first) {
      // determine the mime and select the decoder
      if (!selectDecoder()) {
        return false;
      }
      is_first = false;
    }
 
    // Check if we have a decoder
    if (actual_decoder.decoder == nullptr) return false;
 
    // Use the selected decoder to process data
    return actual_decoder.decoder->copy();
  }
 
  bool selectDecoder(const char* mime) {
    TRACEI();
    bool result = false;
    
    // Guard against null MIME type - cannot proceed without valid MIME
    if (mime == nullptr) {
      LOGE("mime is null");
      return false;
    }
 
    // Optimization: Check if the requested MIME type is already active
    // This avoids unnecessary decoder switching when the same format is detected
    if (StrView(mime).equals(actual_decoder.mime)) {
      is_first = false;  // Mark initialization as complete
      return true;       // Already using the correct decoder
    }
 
    // Clean shutdown of currently active decoder before switching
    // This ensures proper resource cleanup and state reset
    if (actual_decoder.decoder != nullptr) {
      actual_decoder.decoder->end();
      actual_decoder.is_open = false;  // Mark as inactive
    }
 
    // Search through all registered decoders to find one that handles this MIME type
    selected_mime = nullptr;  // Clear previous selection
    for (int j = 0; j < decoders.size(); j++) {
      DecoderInfo info = decoders[j];
      
      // Check if this decoder supports the detected MIME type
      if (StrView(info.mime).equals(mime)) {
        LOGI("Using Decoder %s for %s", toStr(info.mime), toStr(mime));
        
        // Switch to the matching decoder
        actual_decoder = info;
 
        // Configure the decoder's output stream to match our output
        // This ensures decoded audio data flows to the correct destination
        if (p_print != nullptr) {
          actual_decoder.decoder->setOutput(*p_print);
        }
 
        // Initialize the selected decoder and mark it as active
        LOGI("available: %d", p_data_source->available());
        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;
          LOGI("StreamingDecoder %s started", toStr(actual_decoder.mime));
        } else {
          // Decoder failed to start - this is a critical error
          LOGE("Failed to start StreamingDecoder %s", toStr(actual_decoder.mime));
          return false;
        }
 
        // Successfully found and initialized a decoder
        result = true;
        selected_mime = mime;  // Store the MIME type that was selected
        break;                 // Stop searching once we find a match
      }
    }
    
    // Mark initialization phase as complete regardless of success/failure
    is_first = false;
    return result;  // true if decoder was found and started, false otherwise
  }
 
  const char* mime() override {
    // fallback to actual decoder
    if (actual_decoder.decoder != nullptr) {
      return actual_decoder.decoder->mime();
    }
    return nullptr;
  }
 
  const char* selectedMime() { return selected_mime; }
 
  AudioInfo audioInfo() override {
    if (actual_decoder.decoder != nullptr) {
      return actual_decoder.decoder->audioInfo();
    }
    AudioInfo empty;
    return empty;
  }
 
  MimeDetector& mimeDetector() { return mime_detector; }
 
  void setMimeSource(MimeSource& mimeSource) { p_mime_source = &mimeSource; }
 
 protected:
 
  struct DecoderInfo {
    const char* mime = nullptr;           
    StreamingDecoder* decoder = nullptr;  
    bool is_open = false;  
 
    DecoderInfo() = default;
 
    DecoderInfo(const char* mime, StreamingDecoder* decoder) {
      this->mime = mime;
      this->decoder = decoder;
    }
  } actual_decoder;  
 
  Vector<DecoderInfo> decoders{0};  
  Vector<StreamingDecoderAdapter*> adapters{
      0};                      
  MimeDetector mime_detector;  
  Vector<uint8_t> detection_buffer{0};  
  bool is_first = true;                 
  const char* selected_mime = nullptr;  
  MimeSource* p_mime_source =
      nullptr;  
  Stream *p_data_source = nullptr; 
 
  BufferedStream buffered_stream{0};  
  const char* toStr(const char* str){
    return str == nullptr ? "" : str;
  }
 
  bool selectDecoder() {
    // Only perform MIME detection and decoder selection if no decoder is active yet
    // This prevents re-detection on subsequent calls during the same stream
    if (actual_decoder.decoder == nullptr) {
      const char* mime = nullptr;
      p_data_source = nullptr;
 
      // Two methods for MIME type determination: external source or auto-detection
      if (p_mime_source != nullptr) {
        // Option 1: Use externally provided MIME source (e.g., from HTTP headers)
        // This is more efficient as it avoids reading and analyzing stream data
        mime = p_mime_source->mime();
        LOGI("mime from source: %s", toStr(mime));
        assert(p_input != nullptr);
        p_data_source = p_input;
      } else {
        // Option 2: Auto-detect MIME type by analyzing stream content
        // Redirect the decoder to use the buffered stream
        // we use the buffered stream as input
        assert(p_input != nullptr);
        buffered_stream.setStream(*p_input);
        buffered_stream.resize(DEFAULT_BUFFER_SIZE);
        p_data_source = &buffered_stream;
 
        // This requires reading a sample of data to identify the format
        detection_buffer.resize(160);
        size_t bytesRead = buffered_stream.peekBytes(detection_buffer.data(), detection_buffer.size());        // If no data is available, we cannot proceed with detection
        if (bytesRead == 0) return false;
 
        // Feed the sample data to the MIME detector for format analysis
        // The detector examines file headers, magic numbers, etc.
        mime_detector.write(detection_buffer.data(), bytesRead);
        mime = mime_detector.mime();
        LOGI("mime from detector: %s", toStr(mime));
        
      }
      
      // Process the detected/provided MIME type
      if (mime != nullptr) {        
        // Delegate to the overloaded selectDecoder(mime) method to find
        // and initialize the appropriate decoder for this MIME type
        if (!selectDecoder(mime)) {
          LOGE("The decoder could not be selected for %s", toStr(mime));
          return false;  // No registered decoder can handle this format
        }
      } else {
        // MIME detection failed - format is unknown or unsupported
        LOGE("Could not determine mime type");
        return false;
      }
    } else {
      LOGI("Decoder already selected: %s", toStr(actual_decoder.mime));
      assert(p_input != nullptr);
      actual_decoder.decoder->setInput(*p_input);
    }
    
    // Success: either decoder was already selected or selection completed successfully
    return true;
  }
 
  size_t readBytes(uint8_t* data, size_t len) override {
    if (p_input == nullptr) return 0;
    return p_input->readBytes(data, len);
  }
};
 
class DecoderAdapter : public AudioDecoder {
 public:
  DecoderAdapter(StreamingDecoder& dec, int bufferSize) {
    TRACED();
    p_dec = &dec;
    p_dec->setInput(queue);
    resize(bufferSize);
  }
 
  void setOutput(Print& out) override { p_dec->setOutput(out); }
 
  void setInput(Stream& in) { p_dec->setInput(in); }
 
  bool begin() override {
    TRACED();
    active = true;
    bool rc = p_dec->begin();
    return rc;
  }
 
  void end() override {
    TRACED();
    active = false;
  }
 
  void resize(int size) {
    buffer_size = size;
    // setup the buffer only if needed
    if (is_setup) rbuffer.resize(size);
  }
 
  size_t write(const uint8_t* data, size_t len) override {
    TRACED();
    setupLazy();
    size_t result = queue.write((uint8_t*)data, len);
    // Trigger processing - process all available data
    while (p_dec->copy());
 
    return result;
  }
 
  StreamingDecoder* getStreamingDecoder() { return p_dec; }
 
  operator bool() override { return active; }
 
 protected:
  bool active = false;                
  bool is_setup = false;              
  int buffer_size;                    
  StreamingDecoder* p_dec = nullptr;  
  RingBuffer<uint8_t> rbuffer{0};     
  QueueStream<uint8_t> queue{rbuffer};  
 
  void setupLazy() {
    if (!is_setup) {
      rbuffer.resize(buffer_size);
      queue.begin();
      is_setup = true;
    }
  }
};
 
using DecoderFromStreaming = DecoderAdapter;
 
}  // namespace audio_tools