oboe: fix sample rate converter for input streams

The callbackCount was not handled correctly.
This could cause a stream to stall and stop calling callbacks.

Fixes #987

5 files changed, 20 insertions, 16 deletions

diff --git a/src/common/DataConversionFlowGraph.cpp b/src/common/DataConversionFlowGraph.cpp
index 7ddc0e60..6f0ee5b8 100644
--- a/src/common/DataConversionFlowGraph.cpp
+++ b/src/common/DataConversionFlowGraph.cpp
@@ -98,8 +98,8 @@ Result DataConversionFlowGraph::configure(AudioStream *sourceStream, AudioStream
                                 ? sourceStream->getFramesPerBurst()
                                 : sourceStream->getFramesPerCallback();
     // Source
-    // If OUTPUT and using a callback then call back to the app using a SourceCaller.
-    // If INPUT and NOT using a callback then read from the child stream using a SourceCaller.
+    // IF OUTPUT and using a callback then call back to the app using a SourceCaller.
+    // OR IF INPUT and NOT using a callback then read from the child stream using a SourceCaller.
     if ((sourceStream->getCallback() != nullptr && isOutput)
         || (sourceStream->getCallback() == nullptr && isInput)) {
         switch (sourceFormat) {
@@ -118,8 +118,8 @@ Result DataConversionFlowGraph::configure(AudioStream *sourceStream, AudioStream
         mSourceCaller->setStream(sourceStream);
         lastOutput = &mSourceCaller->output;
     } else {
-        // If OUTPUT and NOT using a callback then write to the child stream using a BlockWriter.
-        // If INPUT and using a callback then write to the app using a BlockWriter.
+        // IF OUTPUT and NOT using a callback then write to the child stream using a BlockWriter.
+        // OR IF INPUT and using a callback then write to the app using a BlockWriter.
         switch (sourceFormat) {
             case AudioFormat::Float:
                 mSource = std::make_unique<SourceFloat>(sourceChannelCount);
@@ -200,8 +200,6 @@ Result DataConversionFlowGraph::configure(AudioStream *sourceStream, AudioStream
     }
     lastOutput->connect(&mSink->input);
 
-    mFramePosition = 0;
-
     return Result::OK;
 }
 
@@ -210,7 +208,6 @@ int32_t DataConversionFlowGraph::read(void *buffer, int32_t numFrames, int64_t t
         mSourceCaller->setTimeoutNanos(timeoutNanos);
     }
     int32_t numRead = mSink->read(buffer, numFrames);
-    mFramePosition += numRead;
     return numRead;
 }
 
@@ -221,7 +218,6 @@ int32_t DataConversionFlowGraph::write(void *inputBuffer, int32_t numFrames) {
     while (true) {
         // Pull and read some data in app format into a small buffer.
         int32_t framesRead = mSink->read(mAppBuffer.get(), flowgraph::kDefaultBufferSize);
-        mFramePosition += framesRead;
         if (framesRead <= 0) break;
         // Write to a block adapter, which will call the destination whenever it has enough data.
         int32_t bytesRead = mBlockWriter.write(mAppBuffer.get(),
diff --git a/src/common/DataConversionFlowGraph.h b/src/common/DataConversionFlowGraph.h
index 5b8d3f66..0cde1f35 100644
--- a/src/common/DataConversionFlowGraph.h
+++ b/src/common/DataConversionFlowGraph.h
@@ -80,8 +80,6 @@ private:
     DataCallbackResult                                 mCallbackResult = DataCallbackResult::Continue;
     AudioStream                                       *mFilterStream = nullptr;
     std::unique_ptr<uint8_t[]>                         mAppBuffer;
-
-    int64_t mFramePosition = 0;
 };
 
 }
diff --git a/src/flowgraph/FlowGraphNode.cpp b/src/flowgraph/FlowGraphNode.cpp
index c7e3ff9d..9a62d7d8 100644
--- a/src/flowgraph/FlowGraphNode.cpp
+++ b/src/flowgraph/FlowGraphNode.cpp
@@ -111,4 +111,4 @@ float *FlowGraphPortFloatInput::getBuffer() {
 
 int32_t FlowGraphSink::pullData(int32_t numFrames) {
     return FlowGraphNode::pullData(numFrames, getLastCallCount() + 1);
-}
\ No newline at end of file
+}
diff --git a/src/flowgraph/SampleRateConverter.cpp b/src/flowgraph/SampleRateConverter.cpp
index 0c92d7fd..b1ae4bd8 100644
--- a/src/flowgraph/SampleRateConverter.cpp
+++ b/src/flowgraph/SampleRateConverter.cpp
@@ -25,11 +25,17 @@ SampleRateConverter::SampleRateConverter(int32_t channelCount, MultiChannelResam
     setDataPulledAutomatically(false);
 }
 
+void SampleRateConverter::reset() {
+    FlowGraphNode::reset();
+    mInputCursor = kInitialCallCount;
+}
+
 // Return true if there is a sample available.
 bool SampleRateConverter::isInputAvailable() {
+    // If we have consumed all of the input data then go out and get some more.
     if (mInputCursor >= mNumValidInputFrames) {
-        mNumValidInputFrames = input.pullData(mInputFramePosition, input.getFramesPerBuffer());
-        mInputFramePosition += mNumValidInputFrames;
+        mInputCallCount++;
+        mNumValidInputFrames = input.pullData(mInputCallCount, input.getFramesPerBuffer());
         mInputCursor = 0;
     }
     return (mInputCursor < mNumValidInputFrames);
diff --git a/src/flowgraph/SampleRateConverter.h b/src/flowgraph/SampleRateConverter.h
index 5fb5c650..534df49a 100644
--- a/src/flowgraph/SampleRateConverter.h
+++ b/src/flowgraph/SampleRateConverter.h
@@ -38,6 +38,8 @@ public:
         return "SampleRateConverter";
     }
 
+    void reset() override;
+
 private:
 
     // Return true if there is a sample available.
@@ -48,9 +50,11 @@ private:
 
     resampler::MultiChannelResampler &mResampler;
 
-    int32_t mInputCursor = 0;
-    int32_t mNumValidInputFrames = 0;
-    int64_t mInputFramePosition = 0; // monotonic counter of input frames used for pullData
+    int32_t mInputCursor = 0;         // offset into the input port buffer
+    int32_t mNumValidInputFrames = 0; // number of valid frames currently in the input port buffer
+    // We need our own callCount for upstream calls because calls occur at a different rate.
+    // This means we cannot have cyclic graphs or merges that contain an SRC.
+    int64_t mInputCallCount = 0;
 
 };