HttpRequest Post Stream with DynamicMemoryStream

[ffaerber]

this is a follow-up question on this #1803 discussions.
a stream post to a local endpoint is working fine.
but if i use an external api with https i get a delay of a few seconds before the records starts.
i think this has to do with the SSH handshake.

my idea is to add a DynamicMemoryStream in between as a Buffer
and run two task. one task to write to the DynamicMemoryStream
and and a second task to read from the DynamicMemoryStream

I2SStream > EncodedAudioStream > DynamicMemoryStream > HttpRequest

this way i can start the recordingTask and uploadTask in parallel.
the recordingTask starts recording immediately.
and the uploadTask starts uploading when he is ready.

the delay is gone but the problem i have now is that the upload is faster than the recording
This leads to reuploading already uploaded audio. the cassette(DynamicMemoryStream) is rewind every time the HttpRequest reaching the end of the cassette.
how to prevent this?

#include "audio.h"

#include <Arduino.h>
#include <SPI.h>
#include <WiFi.h>
#include <Wire.h>

#include "AudioTools.h"
#include "AudioTools/AudioCodecs/CodecMP3Helix.h"
#include "AudioTools/AudioCodecs/CodecWAV.h"
#include "AudioTools/AudioLibs/MemoryManager.h"

namespace audio {
    TaskHandle_t xControllerHandle = NULL;
    TaskHandle_t xRecordHandle = NULL;

    AudioInfo info(16000, 1, 16);

    const char *url_str = "http://ffaerber-ubuntu:3000/stream-audio";
    WiFiClient client;

    HttpRequest http(client);
    Url url(url_str);
    I2SStream microphone;
    MemoryManager memory(500);
    DynamicMemoryStream cassette(false);

    EncodedAudioStream encoder(&cassette, new WAVEncoder());
    StreamCopy copierMicToCassette(encoder, microphone);
    StreamCopy copierCassetteToHttp(http, cassette);

    void stream(void *pvParameters) {
        Serial.println("stream: start");
        http.addRequestHeader("Authorization", "Bearer xxx");
        http.header().put(TRANSFER_ENCODING, CHUNKED);
        if (!http.processBegin(POST, url, "audio/wav")) {
            Serial.println("post failed");
            stop();
        }

        Serial.println("stream: send cassette");
        copierCassetteToHttp.setDelayOnNoData(100);
        copierCassetteToHttp.copyAll();
        http.processEnd();
        cassette.end();
        Serial.println("stream: task delete");
        vTaskDelete(NULL);
    }

    void record(void *pvParameters) {
        Serial.println("record: Start");
        Serial.println("record: encoder and cassette begin");

        cassette.begin();
        encoder.begin();

        uint32_t ulNotificationValue;

        Serial.println("record: recording loop");
        while (1) {
            // Check for stop notification
            if (xTaskNotifyWait(0, ULONG_MAX, &ulNotificationValue, 0) == pdPASS) {
                Serial.println("record: stoping via notification");
                break;
            }
            Serial.print("<");
            copierMicToCassette.copy();
        }
        Serial.println("record: encoder.end();");
        encoder.end();
        Serial.println("record: delete");
        xRecordHandle = NULL;
        vTaskDelete(NULL);
    }


    void controller(void *pvParameter) {
        uint32_t ulNotificationValue;
        const TickType_t xTicksToWait = portMAX_DELAY;
        // client.setCACert(rootCACertificate);
        AudioLogger::instance().begin(Serial, AudioLogger::Error);

        auto micConfig = microphone.defaultConfig(RX_MODE);
        micConfig.copyFrom(info);
        micConfig.signal_type = PDM;
        micConfig.i2s_format = I2S_PCM;
        micConfig.pin_bck = I2S_PIN_NO_CHANGE;
        micConfig.pin_ws = 42;
        micConfig.pin_data = 41;
        microphone.begin(micConfig);

        encoder.begin(info);

        while (true) {
            if (xTaskNotifyWait(pdFALSE, ULONG_MAX, &ulNotificationValue, xTicksToWait) == pdPASS) {
                EventType currentEvent = static_cast<EventType>(ulNotificationValue);

                switch (currentEvent) {
                    case START_RECORD:
                        if (xRecordHandle == NULL) {
                            xTaskCreatePinnedToCore(record, "record", 4 * 1024, NULL, 1, &xRecordHandle, 1);
                            vTaskDelay(100 / portTICK_PERIOD_MS);
                            xTaskCreatePinnedToCore(stream, "stream", 4 * 1024, NULL, 1, NULL, 1);
                        } else {
                            Serial.println("record task is already running");
                        }

                        break;

                    case STOP_RECORD:
                        if (xRecordHandle != NULL) {
                            xTaskNotify(xRecordHandle, 0, eNoAction);
                            xRecordHandle = NULL;
                        }
                        break;

                    default:
                        break;
                }
            }
            vTaskDelay(10 / portTICK_PERIOD_MS);
        }
    }

} 

[pschatzmann]

If you upload to a https server, I think you use the wrong client: only WiFiClientSecure supports https.
But I recommend to stick with http, since https is just adding unnecessary overhead!

If you plan to use multple tasks, I suggest to read the correpsonging Multicore Processing wiki

[ffaerber]

yes WiFiClientSecure is working fine. but very slow. the api i use is behind https
i will take a look add BufferRTOS

[ffaerber]

I've been trying to get BufferRTOS to work for the last few days. It works, but unfortunately only the first time.
the writeTask is only running once.

#include <Arduino.h>
#include <SPI.h>
#include <WiFi.h>
#include <Wire.h>
#include "audio.h"
#include "AudioTools.h"
#include "AudioTools/AudioCodecs/CodecMP3Helix.h"
#include "AudioTools/AudioCodecs/CodecWAV.h"
#include "AudioTools/AudioLibs/MemoryManager.h"
#include "AudioTools/Concurrency/All.h"

namespace audio {
    TaskHandle_t xControllerHandle = NULL;
    TaskHandle_t xRecordStreamHandle = NULL;
    const char *rootCACertificate =
        "-----BEGIN CERTIFICATE-----\n"
        "-----END CERTIFICATE-----\n";

    AudioInfo info(16000, 1, 16);

    const char *url_str = "https://api.example.com/stream";
    WiFiClientSecure client;

    HttpRequest http(client);
    Url url(url_str);

    I2SStream microphone;
    MemoryManager memory(500);

    BufferRTOS<uint8_t> buffer(500 * 1024);  // 500KB buffer
    QueueStream<uint8_t> queue(buffer);
    EncodedAudioStream encoder(&queue, new WAVEncoder());
    StreamCopy copierMicToQueue(encoder, microphone);
    StreamCopy copierQueueToHttp(http, queue);

    Task writeTask("write", 7 * 1024, 1, 1);
    Task readTask("read", 7 * 1024, 1, 1);

    void recordStream(void *pvParameters) {
        Serial.println("recordStream start");

        uint32_t ulNotificationValue;

        encoder.begin(info);
        queue.begin();

        readTask.begin([]() {
            Serial.print("r");
            copierMicToQueue.copy();
        });

        http.addRequestHeader("Authorization", "Bearer xxxxxxxxxxxx");
        http.header().put(TRANSFER_ENCODING, CHUNKED);
        if (!http.processBegin(POST, url, "audio/wav")) {
            Serial.println("post failed");
            stop();
        }

        writeTask.begin([]() {
            Serial.println("writeTask start");
            copierQueueToHttp.copyAll();
            http.processEnd();
        });

        while (1) {
            // Check for stop notification
            if (xTaskNotifyWait(0, ULONG_MAX, &ulNotificationValue, 0) == pdPASS) {
                Serial.println("stop notification received");
                readTask.suspend();
                // Wait until the buffer is empty before exiting
                while (!buffer.isEmpty() || !http.connected()) {
                    vTaskDelay(100 / portTICK_PERIOD_MS);  // Wait and check again
                    Serial.print(".");
                }

                break;  // Safe to stop now
            }
            vTaskDelay(10 / portTICK_PERIOD_MS);  // Prevent tight loop
        }

        Serial.println("stoping");

        encoder.end();
        queue.end();
        buffer.clear();

        Serial.println("delete recordStream");
        vTaskDelete(NULL);
    }

    void controller(void *pvParameter) {
        uint32_t ulNotificationValue;
        const TickType_t xTicksToWait = portMAX_DELAY;

        //client.setCACert(rootCACertificate);
        AudioLogger::instance().begin(Serial, AudioLogger::Error);

        auto micConfig = microphone.defaultConfig(RX_MODE);
        micConfig.copyFrom(info);
        micConfig.signal_type = PDM;
        micConfig.i2s_format = I2S_PCM;
        micConfig.pin_bck = I2S_PIN_NO_CHANGE;
        micConfig.pin_ws = 42;
        micConfig.pin_data = 41;
        microphone.begin(micConfig);

        while (true) {
            if (xTaskNotifyWait(pdFALSE, ULONG_MAX, &ulNotificationValue, xTicksToWait) == pdPASS) {
                EventType currentEvent = static_cast<EventType>(ulNotificationValue);

                switch (currentEvent) {
                    case START_RECORD:
                        if (xRecordStreamHandle == NULL) {
                            xTaskCreatePinnedToCore(recordStream, "recordStream", 4 * 1024, NULL, 1, &xRecordStreamHandle, 1);
                        } else {
                            Serial.println("recordStream task is already running");
                        }
                        break;

                    case STOP_RECORD:
                        if (xRecordStreamHandle != NULL) {
                            xTaskNotify(xRecordStreamHandle, 0, eNoAction);
                            xRecordStreamHandle = NULL;
                        }
                        break;

                    default:
                        break;
                }
            }
            vTaskDelay(10 / portTICK_PERIOD_MS);
        }
    }

}  // namespace audio

[ffaerber]

It works well now. I'm using http now. I've also removed the recordStream task.

#include "audio.h"
#include <Arduino.h>
#include <SPI.h>
#include <WiFi.h>
#include <Wire.h>

#include "AudioTools.h"
#include "AudioTools/AudioCodecs/CodecMP3Helix.h"
#include "AudioTools/AudioCodecs/CodecWAV.h"
#include "AudioTools/AudioLibs/MemoryManager.h"
#include "AudioTools/Concurrency/All.h"

namespace audio {
    TaskHandle_t xControllerHandle = NULL;
    TaskHandle_t xRecordStreamHandle = NULL;
    WiFiClient client;

    AudioInfo info(16000, 1, 16);

    I2SStream microphone;
    BufferRTOS<uint8_t> buffer(200 * 1024);
    QueueStream<uint8_t> queue(buffer);
    EncodedAudioStream encoder(&queue, new WAVEncoder());
    StreamCopy copierMicToQueue(encoder, microphone);
    HttpRequest http(client);
    StreamCopy copierQueueToHttp(http, queue);

    Task writeTask("write", 7 * 1024, 1, 1);
    Task readTask("read", 7 * 1024, 1, 1);

    void startRecord() {
        Serial.println("start");

        buffer.clear();
        encoder.begin(info);
        queue.begin();

        readTask.begin([]() {
            Serial.print("r");
            copierMicToQueue.copy();
        });

        Url url("http://ffaerber-ubuntu:8000/stream?chatId=6a04803f-a4ac-4918-afd0-61e291922565");
        http.addRequestHeader("Authorization", "Bearer xxxxxxxxxxxxxxxxxxxxxx");
        http.header().put(TRANSFER_ENCODING, CHUNKED);
        if (!http.processBegin(POST, url, "audio/wav")) {
            Serial.println("post failed");
            stop();
        }

        writeTask.begin([]() {
            Serial.println("writeTask start");
            copierQueueToHttp.copyAll();
            Serial.println("writeTask end");
        });
    }

    void stopRecord() {
        Serial.println("stop");
        readTask.suspend();
        while (!buffer.isEmpty()) {
            vTaskDelay(100 / portTICK_PERIOD_MS);  // Wait and check again
            Serial.print('.');
        }
        http.processEnd();
        writeTask.suspend();

        http.end();
        encoder.end();
        queue.end();
        writeTask.end();
        readTask.end();
    }

    void controller(void *pvParameter) {
        uint32_t ulNotificationValue;
        const TickType_t xTicksToWait = portMAX_DELAY;

        AudioLogger::instance().begin(Serial, AudioLogger::Warning);

        // xiao esp32s3 sense microphone Configuration
        auto micConfig = microphone.defaultConfig(RX_MODE);
        micConfig.copyFrom(info);
        micConfig.signal_type = PDM;
        micConfig.i2s_format = I2S_PCM;
        micConfig.pin_bck = I2S_PIN_NO_CHANGE;
        micConfig.pin_ws = 42;
        micConfig.pin_data = 41;
        microphone.begin(micConfig);

        while (true) {
            if (xTaskNotifyWait(pdFALSE, ULONG_MAX, &ulNotificationValue, xTicksToWait) == pdPASS) {
                EventType currentEvent = static_cast<EventType>(ulNotificationValue);

                switch (currentEvent) {
                    case START_RECORD:
                        startRecord();
                        break;

                    case STOP_RECORD:
                        stopRecord();
                        break;

                    default:
                        break;
                }
            }
            vTaskDelay(1000 / portTICK_PERIOD_MS);
        }
    }

}  // namespace audio