StreamCopy.h

#pragma once
 
#include "AudioTools/CoreAudio/AudioLogger.h"
#include "AudioTools/CoreAudio/AudioMetaData/MimeDetector.h"
#include "AudioTools/CoreAudio/AudioStreams.h"
#include "AudioTools/CoreAudio/AudioTypes.h"
#include "AudioTools/CoreAudio/BaseConverter.h"
#include "AudioTools/CoreAudio/Buffers.h"
#include "AudioToolsConfig.h"
 
#define NOT_ENOUGH_MEMORY_MSG "Could not allocate enough memory: %d bytes"
 
namespace audio_tools {
 
template <class T>
class StreamCopyT {
 public:
  StreamCopyT(Allocator &allocator, int bufferSize = DEFAULT_BUFFER_SIZE)
      : _allocator(allocator), buffer_size(bufferSize) {
    TRACED();
    begin();
  }
  StreamCopyT(Print &to, AudioStream &from,
              int bufferSize = DEFAULT_BUFFER_SIZE,
              Allocator &allocator = DefaultAllocator)
      : _allocator(allocator), buffer_size(bufferSize)  {
    TRACED();
    begin(to, from);
  }
 
  StreamCopyT(Print &to, Stream &from, int bufferSize = DEFAULT_BUFFER_SIZE,
              Allocator &allocator = DefaultAllocator)
      : _allocator(allocator), buffer_size(bufferSize) {
    TRACED();
    begin(to, from);
  }
 
  StreamCopyT(int bufferSize = DEFAULT_BUFFER_SIZE,
              Allocator &allocator = DefaultAllocator)
      : _allocator(allocator), buffer_size(bufferSize) {  
    TRACED();
    begin();
  }
 
  ~StreamCopyT() { end(); }
 
  void begin() {
    TRACED();
    if (p_mime_detector != nullptr) {
      p_mime_detector->begin();
    };
    resize(buffer_size);
    if (buffer) {
      LOGI("buffer_size=%d", buffer_size);
    } else {
      LOGE(NOT_ENOUGH_MEMORY_MSG, buffer_size);
    }
  }
 
  void end() {
    this->from = nullptr;
    this->to = nullptr;
  }
 
  void begin(Print &to, Stream &from) {
    this->from = &from;
    this->to = &to;
    begin();
  }
 
  void begin(Print &to, AudioStream &from) {
    this->from = &from;
    this->from_audio = &from;
    this->to = &to;
    begin();
  }
 
  Stream *getFrom() { return from; }
 
  Print *getTo() { return to; }
 
  inline size_t copy() {
    p_converter = nullptr;
    return copyBytes(buffer_size);
  }
 
  inline size_t copy(BaseConverter &converter) {
    p_converter = &converter;
    return copyBytes(buffer_size);
  }
 
  inline size_t copyBytes(size_t bytes) {
    LOGD("copy %d bytes %s", (int)bytes, log_name);
    if (!active) return 0;
    // if not initialized we do nothing
    if (from == nullptr && to == nullptr) return 0;
 
    // if no bytes are requested, we do nothing
    if (bytes == 0) return 0;
 
    // synchronize AudioInfo
    syncAudioInfo();
 
    // E.g. if we try to write to a server we might not have any output
    // destination yet
    int to_write = to->availableForWrite();
    if (check_available_for_write && to_write == 0) {
      delay(500);
      return 0;
    }
 
    // resize copy buffer if necessary
    if (buffer.size() < bytes) {
      LOGI("Resize to %d", (int)bytes);
      buffer.resize(bytes);
    }
 
    size_t result = 0;
    size_t delayCount = 0;
    size_t len = bytes;
    if (check_available) {
      len = available();
    }
    size_t bytes_to_read = len;
    size_t bytes_read = 0;
 
    if (len > 0) {
      bytes_to_read = min(len, static_cast<size_t>(buffer_size));
      // don't overflow buffer
      if (check_available_for_write && to_write > 0) {
        bytes_to_read = min((int)bytes_to_read, to_write);
      }
 
      // round to full frames
      int copy_size = minCopySize();
      if (copy_size > 0) {
        size_t samples = bytes_to_read / minCopySize();
        bytes_to_read = samples * minCopySize();
      }
 
      // get the data now
      bytes_read = 0;
      if (bytes_to_read > 0) {
        bytes_read = from->readBytes((uint8_t *)&buffer[0], bytes_to_read);
      }
 
      // determine mime
      if (p_mime_detector != nullptr) {
        p_mime_detector->write(buffer.data(), bytes_to_read);
      }
 
      // convert data
      if (p_converter != nullptr)
        p_converter->convert((uint8_t *)buffer.data(), bytes_read);
 
      // write data
      result = write(bytes_read, delayCount);
 
      // callback with unconverted data
      if (onWrite != nullptr) onWrite(onWriteObj, &buffer[0], result);
 
#ifndef COPY_LOG_OFF
      LOGI("StreamCopy::copy %s %u -> %u -> %u bytes - in %u hops", log_name,
           (unsigned int)bytes_to_read, (unsigned int)bytes_read,
           (unsigned int)result, (unsigned int)delayCount);
#endif
      // TRACED();
 
      if (result == 0) {
        TRACED();
        // give the processor some time
        delay(delay_on_no_data);
      }
 
      // TRACED();
      CHECK_MEMORY();
    } else {
      // give the processor some time
      delay(delay_on_no_data);
      LOGD("no data %s", log_name);
    }
    // TRACED();
    return result;
  }
 
  size_t copyN(size_t pages) {
    if (!active) return 0;
    size_t total = 0;
    for (size_t j = 0; j < pages; j++) {
      total += copy();
    }
    return total;
  }
 
  size_t copyMs(size_t millis, AudioInfo info) {
    if (!active) return 0;
    size_t pages = AudioTime::toBytes(millis, info) / buffer_size;
    return copyN(pages);
  }
 
  size_t copyAll(int retryCount = 5, int retryWaitMs = 200) {
    TRACED();
    if (!active) return 0;
    size_t result = 0;
    int retry = 0;
 
    if (from == nullptr || to == nullptr) return result;
 
    // copy while source has data available
    int count = 0;
    while (true) {
      count = copy();
      result += count;
      if (count == 0) {
        // wait for more data
        retry++;
        delay(retryWaitMs);
      } else {
        retry = 0;  // after we got new data we restart the counting
      }
      // stop the processing if we passed the retry limit
      if (retry > retryCount) {
        break;
      }
    }
    return result;
  }
 
  int available() {
    int result = 0;
    if (from != nullptr) {
      if (availableCallback != nullptr) {
        result = availableCallback((Stream *)from);
      } else {
        result = from->available();
      }
    } else {
      LOGW("source not defined");
    }
    LOGD("available: %d", result);
    return result;
  }
 
  void setDelayOnNoData(int delayMs) { delay_on_no_data = delayMs; }
 
  void setCallbackOnWrite(void (*onWrite)(void *obj, void *buffer, size_t len),
                          void *obj) {
    TRACED();
    this->onWrite = onWrite;
    this->onWriteObj = obj;
  }
 
  void setAvailableCallback(int (*callback)(Stream *stream)) {
    availableCallback = callback;
  }
 
  void setRetry(int retry) { retryLimit = retry; }
 
  int bufferSize() { return buffer_size; }
 
  void setCheckAvailableForWrite(bool flag) {
    check_available_for_write = flag;
  }
 
  bool isCheckAvailableForWrite() { return check_available_for_write; }
 
  void setCheckAvailable(bool flag) { check_available = flag; }
 
  bool isCheckAvailable() { return check_available; }
 
  void resize(int len) {
    buffer_size = len;
    buffer.resize(len);
  }
 
  void setActive(bool flag) { active = flag; }
 
  bool isActive() { return active; }
 
  void setLogName(const char *name) { log_name = name; }
 
  void setRetryDelay(int delay) { retry_delay = delay; }
 
  int minCopySize() {
    if (min_copy_size == 0 && from_audio != nullptr) {
      AudioInfo info = from_audio->audioInfoOut();
      min_copy_size = info.bits_per_sample / 8 * info.channels;
    }
    return min_copy_size;
  }
 
  void setMinCopySize(int size) { min_copy_size = size; }
 
  void setSynchAudioInfo(bool active) { is_sync_audio_info = active; }
 
  void setMimeDetector(MimeDetector &mime) { p_mime_detector = &mime; }
 
 protected:
  Allocator _allocator;
  Stream *from = nullptr;
  AudioStream *from_audio = nullptr;
  Print *to = nullptr;
  Vector<uint8_t> buffer{0, _allocator};
  int buffer_size = DEFAULT_BUFFER_SIZE;
  void (*onWrite)(void *obj, void *buffer, size_t len) = nullptr;
  int (*availableCallback)(Stream *stream) = nullptr;
  void *onWriteObj = nullptr;
  bool check_available_for_write = false;
  bool check_available = true;
  int retryLimit = COPY_RETRY_LIMIT;
  int delay_on_no_data = COPY_DELAY_ON_NODATA;
  bool active = true;
  const char *log_name = "";
  int retry_delay = 10;
  int channels = 0;
  int min_copy_size = 1;
  bool is_sync_audio_info = false;
  AudioInfoSupport *p_audio_info_support = nullptr;
  BaseConverter *p_converter = nullptr;
  MimeDetector *p_mime_detector = nullptr;
 
  void syncAudioInfo() {
    // synchronize audio info
    if (is_sync_audio_info && from_audio != nullptr &&
        p_audio_info_support != nullptr) {
      AudioInfo info_from = from_audio->audioInfoOut();
      AudioInfo info_to = p_audio_info_support->audioInfo();
      if (info_from != info_to) {
        LOGI("--> StreamCopy: ");
        p_audio_info_support->setAudioInfo(info_from);
      }
    }
  }
 
  size_t write(size_t len, size_t &delayCount) {
    if (!buffer || len == 0) return 0;
    LOGD("write: %d", (int)len);
    size_t total = 0;
    long open = len;
    int retry = 0;
    while (open > 0) {
      size_t written = to->write((const uint8_t *)buffer.data() + total, open);
      LOGD("write: %d -> %d", (int)open, (int)written);
      total += written;
      open -= written;
      delayCount++;
 
      if (open > 0) {
        // if we still have progress we reset the retry counter
        if (written > 0) retry = 0;
 
        // abort if we reached the retry limit
        if (retry++ > retryLimit) {
          LOGE("write %s to target has failed after %d retries! (%ld bytes)",
               log_name, retry, open);
          break;
        }
 
        // wait a bit
        if (retry > 1) {
          delay(retry_delay);
          LOGI("try write %s - %d (open %ld bytes) ", log_name, retry, open);
        }
      }
    }
    return total;
  }
};
 
using StreamCopy = StreamCopyT<uint8_t>;
 
}  // namespace audio_tools