FullDuplexStream.h

/*
 * Copyright 2023 The Android Open Source Project
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#ifndef OBOE_FULL_DUPLEX_STREAM_
#define OBOE_FULL_DUPLEX_STREAM_
 
#include <cstdint>
#include "oboe/Definitions.h"
#include "oboe/AudioStream.h"
#include "oboe/AudioStreamCallback.h"
 
namespace oboe {
 
class FullDuplexStream : public AudioStreamDataCallback {
public:
    FullDuplexStream() {}
    virtual ~FullDuplexStream() = default;
 
    void setInputStream(AudioStream *stream) {
        mRawInputStream = stream;
    }
 
    void setSharedInputStream(std::shared_ptr<AudioStream> &stream) {
        mSharedInputStream = stream;
    }
 
    AudioStream *getInputStream() {
        if (mSharedInputStream) {
            return mSharedInputStream.get();
        } else {
            return mRawInputStream;
        }
    }
 
    void setOutputStream(AudioStream *stream) {
        mRawOutputStream = stream;
    }
 
    void setSharedOutputStream(std::shared_ptr<AudioStream> &stream) {
        mSharedOutputStream = stream;
    }
 
    AudioStream *getOutputStream() {
        if (mSharedOutputStream) {
            return mSharedOutputStream.get();
        } else {
            return mRawOutputStream;
        }
    }
 
    virtual Result start() {
        mCountCallbacksToDrain = kNumCallbacksToDrain;
        mCountInputBurstsCushion = mNumInputBurstsCushion;
        mCountCallbacksToDiscard = kNumCallbacksToDiscard;
 
        // Determine maximum size that could possibly be called.
        int32_t bufferSize = getOutputStream()->getBufferCapacityInFrames()
                             * getOutputStream()->getChannelCount();
        if (bufferSize > mBufferSize) {
            mInputBuffer = std::make_unique<float[]>(bufferSize);
            mBufferSize = bufferSize;
        }
 
        oboe::Result result = getInputStream()->requestStart();
        if (result != oboe::Result::OK) {
            return result;
        }
        return getOutputStream()->requestStart();
    }
 
    virtual Result stop() {
        Result outputResult = Result::OK;
        Result inputResult = Result::OK;
        if (getOutputStream()) {
            outputResult = getOutputStream()->requestStop();
        }
        if (getInputStream()) {
            inputResult = getOutputStream()->requestStop();
        }
        if (outputResult != Result::OK) {
            return outputResult;
        } else {
            return inputResult;
        }
    }
 
    virtual ResultWithValue<int32_t> readInput(int32_t numFrames) {
        return getInputStream()->read(mInputBuffer.get(), numFrames, 0 /* timeout */);
    }
 
    virtual DataCallbackResult onBothStreamsReady(
            const void *inputData,
            int   numInputFrames,
            void *outputData,
            int   numOutputFrames
            ) = 0;
 
    DataCallbackResult onAudioReady(
            AudioStream * /*audioStream*/,
            void *audioData,
            int numFrames) {
        DataCallbackResult callbackResult = DataCallbackResult::Continue;
        int32_t actualFramesRead = 0;
 
        // Silence the output.
        int32_t numBytes = numFrames * getOutputStream()->getBytesPerFrame();
        memset(audioData, 0 /* value */, numBytes);
 
        if (mCountCallbacksToDrain > 0) {
            // Drain the input.
            int32_t totalFramesRead = 0;
            do {
                ResultWithValue<int32_t> result = readInput(numFrames);
                if (!result) {
                    // Ignore errors because input stream may not be started yet.
                    break;
                }
                actualFramesRead = result.value();
                totalFramesRead += actualFramesRead;
            } while (actualFramesRead > 0);
            // Only counts if we actually got some data.
            if (totalFramesRead > 0) {
                mCountCallbacksToDrain--;
            }
 
        } else if (mCountInputBurstsCushion > 0) {
            // Let the input fill up a bit so we are not so close to the write pointer.
            mCountInputBurstsCushion--;
 
        } else if (mCountCallbacksToDiscard > 0) {
            mCountCallbacksToDiscard--;
            // Ignore. Allow the input to reach to equilibrium with the output.
            ResultWithValue<int32_t> resultAvailable = getInputStream()->getAvailableFrames();
            if (!resultAvailable) {
                callbackResult = DataCallbackResult::Stop;
            } else {
                int32_t framesAvailable = resultAvailable.value();
                if (framesAvailable >= mMinimumFramesBeforeRead) {
                    ResultWithValue<int32_t> resultRead = readInput(numFrames);
                    if (!resultRead) {
                        callbackResult = DataCallbackResult::Stop;
                    }
                }
            }
        } else {
            int32_t framesRead = 0;
            ResultWithValue<int32_t> resultAvailable = getInputStream()->getAvailableFrames();
            if (!resultAvailable) {
                callbackResult = DataCallbackResult::Stop;
            } else {
                int32_t framesAvailable = resultAvailable.value();
                if (framesAvailable >= mMinimumFramesBeforeRead) {
                    // Read data into input buffer.
                    ResultWithValue<int32_t> resultRead = readInput(numFrames);
                    if (!resultRead) {
                        callbackResult = DataCallbackResult::Stop;
                    } else {
                        framesRead = resultRead.value();
                    }
                }
            }
 
            if (callbackResult == DataCallbackResult::Continue) {
                callbackResult = onBothStreamsReady(mInputBuffer.get(), framesRead,
                                                    audioData, numFrames);
            }
        }
 
        if (callbackResult == DataCallbackResult::Stop) {
            getInputStream()->requestStop();
        }
 
        return callbackResult;
    }
 
    void setNumInputBurstsCushion(int32_t numBursts) {
        mNumInputBurstsCushion = numBursts;
    }
 
    int32_t getNumInputBurstsCushion() const {
        return mNumInputBurstsCushion;
    }
 
    void setMinimumFramesBeforeRead(int32_t numFrames) {
        mMinimumFramesBeforeRead = numFrames;
    }
 
    int32_t getMinimumFramesBeforeRead() const {
        return mMinimumFramesBeforeRead;
    }
 
private:
 
    // TODO add getters and setters
    static constexpr int32_t kNumCallbacksToDrain   = 20;
    static constexpr int32_t kNumCallbacksToDiscard = 30;
 
    // let input fill back up, usually 0 or 1
    int32_t mNumInputBurstsCushion =  0;
    int32_t mMinimumFramesBeforeRead = 0;
 
    // We want to reach a state where the input buffer is empty and
    // the output buffer is full.
    // These are used in order.
    // Drain several callback so that input is empty.
    int32_t              mCountCallbacksToDrain = kNumCallbacksToDrain;
    // Let the input fill back up slightly so we don't run dry.
    int32_t              mCountInputBurstsCushion = mNumInputBurstsCushion;
    // Discard some callbacks so the input and output reach equilibrium.
    int32_t              mCountCallbacksToDiscard = kNumCallbacksToDiscard;
 
    AudioStream *mRawInputStream = nullptr;
    AudioStream *mRawOutputStream = nullptr;
    std::shared_ptr<AudioStream> mSharedInputStream;
    std::shared_ptr<AudioStream> mSharedOutputStream;
 
    int32_t              mBufferSize = 0;
    std::unique_ptr<float[]> mInputBuffer;
};
 
} // namespace oboe
 
#endif //OBOE_FULL_DUPLEX_STREAM_