fix: flag fix (#775)

Author: mdydek

packages/react-native-audio-api/common/cpp/audioapi/core/sources/StreamerNode.cpp

@@ -76,20 +76,10 @@ bool StreamerNode::initialize(const std::string &input_url) {
   receiver_ = std::move(receiver);
 
   streamingThread_ = std::thread(&StreamerNode::streamAudio, this);
-  streamFlag.store(true, std::memory_order_release);
   isInitialized_ = true;
   return true;
 }
 
-void StreamerNode::stop(double when) {
-  AudioScheduledSourceNode::stop(when);
-  streamFlag.store(false, std::memory_order_release);
-  StreamingData dummy;
-  while (receiver_.try_receive(dummy) ==
-         channels::spsc::ResponseStatus::SUCCESS)
-    ; // clear the receiver
-}
-
 bool StreamerNode::setupResampler() {
   // Allocate resampler context
   swrCtx_ = swr_alloc();
@@ -126,7 +116,7 @@ bool StreamerNode::setupResampler() {
 }
 
 void StreamerNode::streamAudio() {
-  while (streamFlag.load(std::memory_order_acquire)) {
+  while (!isNodeFinished_.load(std::memory_order_acquire)) {
     if (av_read_frame(fmtCtx_, pkt_) < 0) {
       return;
     }
@@ -143,6 +133,10 @@ void StreamerNode::streamAudio() {
     }
     av_packet_unref(pkt_);
   }
+  StreamingData dummy;
+  while (receiver_.try_receive(dummy) ==
+         channels::spsc::ResponseStatus::SUCCESS)
+    ; // clear the receiver
 }
 
 std::shared_ptr<AudioBus> StreamerNode::processNode(
@@ -151,6 +145,7 @@ std::shared_ptr<AudioBus> StreamerNode::processNode(
   size_t startOffset = 0;
   size_t offsetLength = 0;
   updatePlaybackInfo(processingBus, framesToProcess, startOffset, offsetLength);
+  isNodeFinished_.store(isFinished(), std::memory_order_release);
 
   if (!isPlaying() && !isStopScheduled()) {
     processingBus->zero();
@@ -171,10 +166,14 @@ std::shared_ptr<AudioBus> StreamerNode::processNode(
       alreadyProcessed += bufferRemaining;
     }
     StreamingData data;
-    receiver_.try_receive(data);
-    bufferedBus_ = std::make_shared<AudioBus>(std::move(data.bus));
-    bufferedBusSize_ = data.size;
-    processedSamples_ = 0;
+    auto res = receiver_.try_receive(data);
+    if (res == channels::spsc::ResponseStatus::SUCCESS) {
+      bufferedBus_ = std::make_shared<AudioBus>(std::move(data.bus));
+      bufferedBusSize_ = data.size;
+      processedSamples_ = 0;
+    } else {
+      bufferedBus_ = nullptr;
+    }
   }
   if (bufferedBus_ != nullptr) {
     for (int ch = 0; ch < processingBus->getNumberOfChannels(); ch++) {
@@ -223,7 +222,7 @@ bool StreamerNode::processFrameWithResampler(AVFrame *frame) {
   }
 
   // if we would like to finish dont copy anything
-  if (!streamFlag.load(std::memory_order_acquire)) {
+  if (this->isFinished()) {
     return true;
   }
   auto bus = AudioBus(
@@ -281,7 +280,7 @@ bool StreamerNode::setupDecoder() {
 }
 
 void StreamerNode::cleanup() {
-  streamFlag.store(false, std::memory_order_release);
+  this->playbackState_ = PlaybackState::FINISHED;
   // cleanup cannot be called from the streaming thread so there is no need to
   // check if we are in the same thread
   streamingThread_.join();

packages/react-native-audio-api/common/cpp/audioapi/core/sources/StreamerNode.h

@@ -71,7 +71,6 @@ class StreamerNode : public AudioScheduledSourceNode {
    * @brief Initialize all necessary ffmpeg components for streaming audio
   */
   bool initialize(const std::string& inputUrl);
-  void stop(double when) override;
 
  protected:
   std::shared_ptr<AudioBus> processNode(const std::shared_ptr<AudioBus>& processingBus, int framesToProcess) override;
@@ -94,7 +93,7 @@ class StreamerNode : public AudioScheduledSourceNode {
   size_t processedSamples_;
 
   std::thread streamingThread_;
-  std::atomic<bool> streamFlag; // Flag to control the streaming thread
+  std::atomic<bool> isNodeFinished_; // Flag to control the streaming thread
   static constexpr int INITIAL_MAX_RESAMPLED_SAMPLES = 8192; // Initial size for resampled data
   channels::spsc::Sender<StreamingData, STREAMER_NODE_SPSC_OVERFLOW_STRATEGY, STREAMER_NODE_SPSC_WAIT_STRATEGY> sender_;
   channels::spsc::Receiver<StreamingData, STREAMER_NODE_SPSC_OVERFLOW_STRATEGY, STREAMER_NODE_SPSC_WAIT_STRATEGY> receiver_;