StreamCopy.h

#pragma once
 
#include "AudioConfig.h"
#include "AudioTools/AudioTypes.h"
#include "AudioTools/Buffers.h"
#include "AudioTools/BaseConverter.h"
#include "AudioTools/AudioLogger.h"
#include "AudioTools/AudioStreams.h"
 
#define NOT_ENOUGH_MEMORY_MSG "Could not allocate enough memory: %d bytes"
 
namespace audio_tools {
 
template <class T>
class StreamCopyT {
    public:
        StreamCopyT() = default;
        
        StreamCopyT(Print &to, AudioStream &from, int buffer_size=DEFAULT_BUFFER_SIZE){
            TRACED();
            begin(to, from);
            resize(buffer_size);
            if (!buffer){
                LOGE(NOT_ENOUGH_MEMORY_MSG, buffer_size);
            }
        }
 
        StreamCopyT(Print &to, Stream &from, int buffer_size=DEFAULT_BUFFER_SIZE){
            TRACED();
            begin(to, from);
            resize(buffer_size);
            if (!buffer){
                LOGE(NOT_ENOUGH_MEMORY_MSG, buffer_size);
            }
        }
 
        StreamCopyT(int buffer_size=DEFAULT_BUFFER_SIZE){
            TRACED();
            resize(buffer_size);
            if (!buffer){
                LOGE(NOT_ENOUGH_MEMORY_MSG, buffer_size);
            }
        }
 
        void begin(){     
            is_first = true;   
            LOGI("buffer_size=%d",buffer_size);    
        }
 
        void end() {
            this->from = nullptr;
            this->to = nullptr;
        }
 
        void begin(Print &to, Stream &from){
            this->from = new AudioStreamWrapper(from);
            this->to = &to;
            is_first = true;
            LOGI("buffer_size=%d",buffer_size);    
        }
 
        void begin(Print &to, AudioStream &from){
            this->from = &from;
            this->to = &to;
            is_first = true;
            LOGI("buffer_size=%d",buffer_size);    
        }
 
        Stream *getFrom(){
            return from;
        }
 
        Print *getTo() {
            return to;
        }
 
        inline size_t copy(){
            LOGD("copy %s", log_name);
            if (!active) return 0;
            // if not initialized we do nothing
            if (from==nullptr && to==nullptr) 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;
            }
 
            size_t result = 0;
            size_t delayCount = 0;
            size_t len = buffer_size;
            if (check_available) {
                len = available();
            }
            size_t bytes_to_read = buffer_size;
            size_t bytes_read = 0; 
 
            if (len > 0){
                bytes_to_read = min(len, static_cast<size_t>(buffer_size));
                // don't overflow buffer
                if (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
                notifyMime(buffer.data(), bytes_to_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;
        }
 
 
        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;
        }
 
        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;
        }
 
        const char* mime() {
            return actual_mime;
        }
 
        void setMimeCallback(void (*callback)(const char*)){
            TRACED();
            this->notifyMimeCallback = callback;
        }
 
        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(buffer_size);
        }
 
        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){
             AudioInfo info = from->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;
        }
 
    protected:
        AudioStream *from = nullptr;
        Print *to = nullptr;
        Vector<uint8_t> buffer{0};
        int buffer_size;
        void (*onWrite)(void*obj, void*buffer, size_t len) = nullptr;
        void (*notifyMimeCallback)(const char*mime) = nullptr;
        int (*availableCallback)(Stream*stream)=nullptr;
        void *onWriteObj = nullptr;
        bool is_first = false;
        bool check_available_for_write = false;
        bool check_available = true;
        const char* actual_mime = nullptr;
        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;
 
        void syncAudioInfo(){
            // synchronize audio info
            if (is_sync_audio_info && from != nullptr && p_audio_info_support != nullptr){
                AudioInfo info_from = from->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);
                    }
                }
 
                CHECK_MEMORY();
            }
            return total;
        }
 
        void notifyMime(void* data, size_t len){
            if (is_first && len>4) {
                const uint8_t *start = (const uint8_t *) data;
                actual_mime = "audio/basic";
                if (start[0]==0xFF && start[1]==0xF1){
                    actual_mime = "audio/aac";
                } else if (memcmp(start,"ID3",3) || start[0]==0xFF || start[0]==0xFE ){
                    actual_mime = "audio/mpeg";
                } else if (memcmp(start,"RIFF",4)){
                    actual_mime = "audio/vnd.wave";
                }
                if (notifyMimeCallback!=nullptr){
                    notifyMimeCallback(actual_mime);
                }
            }
            is_first = false;
        }
 
};
 
class StreamCopy : public StreamCopyT<uint8_t> {
    public:
        StreamCopy(int buffer_size=DEFAULT_BUFFER_SIZE): StreamCopyT<uint8_t>(buffer_size) {            
             TRACED();
        }
 
        StreamCopy(AudioStream &to, AudioStream &from, int buffer_size=DEFAULT_BUFFER_SIZE) : StreamCopyT<uint8_t>(to, from, buffer_size){
             TRACED();
             p_audio_info_support = &to;
        }
 
        StreamCopy(AudioOutput &to, AudioStream &from, int buffer_size=DEFAULT_BUFFER_SIZE) : StreamCopyT<uint8_t>(to, from, buffer_size){
             TRACED();
             p_audio_info_support = &to;
        }
 
        StreamCopy(Print &to, AudioStream &from, int buffer_size=DEFAULT_BUFFER_SIZE) : StreamCopyT<uint8_t>(to, from, buffer_size){
             TRACED();
        }
 
        StreamCopy(Print &to, Stream &from, int buffer_size=DEFAULT_BUFFER_SIZE) : StreamCopyT<uint8_t>(to, from, buffer_size){
             TRACED();
        }
 
        size_t copy(BaseConverter &converter) {
            size_t result = available();
            size_t delayCount = 0;
            syncAudioInfo();
            BaseConverter* coverter_ptr = &converter;
            if (result>0){
                size_t bytes_to_read = min(result, static_cast<size_t>(buffer_size) );
                result = from->readBytes((uint8_t*)&buffer[0], bytes_to_read);
 
                // determine mime
                notifyMime(buffer.data(), bytes_to_read);
                is_first = false;
 
                // callback with unconverted data
                if (onWrite!=nullptr) onWrite(onWriteObj, buffer.data(), result);
 
                // convert data
                coverter_ptr->convert((uint8_t*)buffer.data(),  result );
                write(result, delayCount);
                #ifndef COPY_LOG_OFF
                    LOGI("StreamCopy::copy %u bytes - in %u hops", (unsigned int)result,(unsigned int) delayCount);
                #endif
            } else {
                // give the processor some time 
                delay(delay_on_no_data);
            }
            return result;
        }
        
        size_t copy()  {
            return StreamCopyT<uint8_t>::copy();
        }
        
};
 
} // Namespace