Asure correct data & flush order during multiple flushes (#194)

Summary: Asure correct data & flush order during multiple flushes
Type: Fix
Test Plan: UT/CT, Fullstack
Jira: NO-JIRA

Author: skywojciechowskim

source/CMakeLists.txt

@@ -48,6 +48,7 @@ add_library(gstrialtosinks
         Timer.cpp
         BufferParser.cpp
         LogToGstHandler.cpp
+        FlushAndDataSynchronizer.cpp
         )
 
 target_include_directories(gstrialtosinks

source/FlushAndDataSynchronizer.cpp

@@ -0,0 +1,101 @@
+/*
+ * Copyright (C) 2026 Sky UK
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation;
+ * version 2.1 of the License.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#include "FlushAndDataSynchronizer.h"
+#include "GStreamerMSEMediaPlayerClient.h"
+#include "GstreamerCatLog.h"
+#include <algorithm>
+#include <gst/gst.h>
+
+#define GST_CAT_DEFAULT rialtoGStreamerCat
+
+void FlushAndDataSynchronizer::addSource(int32_t sourceId)
+{
+    std::unique_lock lock(m_mutex);
+    m_sourceStates[sourceId] = {FlushState::IDLE, DataState::NO_DATA};
+    GST_INFO("Added source %d to FlushAndDataSynchronizer", sourceId);
+}
+
+void FlushAndDataSynchronizer::removeSource(int32_t sourceId)
+{
+    std::unique_lock lock(m_mutex);
+    m_sourceStates.erase(sourceId);
+    m_cv.notify_all();
+    GST_INFO("Removed source %d from FlushAndDataSynchronizer", sourceId);
+}
+
+void FlushAndDataSynchronizer::notifyFlushStarted(int32_t sourceId)
+{
+    std::unique_lock lock(m_mutex);
+    m_sourceStates[sourceId].flushState = FlushState::FLUSHING;
+    m_sourceStates[sourceId].dataState = DataState::NO_DATA;
+    GST_INFO("FlushAndDataSynchronizer: Flush started for source %d", sourceId);
+}
+
+void FlushAndDataSynchronizer::notifyFlushCompleted(int32_t sourceId)
+{
+    std::unique_lock lock(m_mutex);
+    m_sourceStates[sourceId].flushState = FlushState::FLUSHED;
+    m_cv.notify_all();
+    GST_INFO("FlushAndDataSynchronizer: Flush completed for source %d", sourceId);
+}
+
+void FlushAndDataSynchronizer::notifyDataReceived(int32_t sourceId)
+{
+    std::unique_lock lock(m_mutex);
+    if (m_sourceStates[sourceId].dataState == DataState::NO_DATA)
+    {
+        m_sourceStates[sourceId].dataState = DataState::DATA_RECEIVED;
+        GST_INFO("FlushAndDataSynchronizer: Data received for source %d", sourceId);
+    }
+}
+
+void FlushAndDataSynchronizer::notifyDataPushed(int32_t sourceId)
+{
+    std::unique_lock lock(m_mutex);
+    m_sourceStates[sourceId].dataState = DataState::DATA_PUSHED;
+    m_sourceStates[sourceId].flushState = FlushState::IDLE;
+    m_cv.notify_all();
+    GST_INFO("FlushAndDataSynchronizer: Data pushed for source %d", sourceId);
+}
+
+void FlushAndDataSynchronizer::waitIfRequired(int32_t sourceId)
+{
+    std::unique_lock lock(m_mutex);
+    GST_INFO("FlushAndDataSynchronizer: waitIfRequired enter for source %d", sourceId);
+    m_cv.wait(lock,
+              [&]()
+              {
+                  return std::none_of(m_sourceStates.begin(), m_sourceStates.end(),
+                                      [&](const auto &state)
+                                      {
+                                          return (state.first == sourceId &&
+                                                  state.second.flushState == FlushState::FLUSHING) ||
+                                                 (state.second.flushState != FlushState::IDLE &&
+                                                  state.second.dataState == DataState::DATA_RECEIVED);
+                                      });
+              });
+    GST_INFO("FlushAndDataSynchronizer: waitIfRequired exit for source %d", sourceId);
+}
+
+bool FlushAndDataSynchronizer::isAnySourceFlushing() const
+{
+    std::unique_lock lock(m_mutex);
+    return std::any_of(m_sourceStates.begin(), m_sourceStates.end(),
+                       [](const auto &state) { return state.second.flushState == FlushState::FLUSHING; });
+}

source/FlushAndDataSynchronizer.h

@@ -0,0 +1,75 @@
+/*
+ * Copyright (C) 2026 Sky UK
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation;
+ * version 2.1 of the License.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#ifndef FLUSH_AND_DATA_SYNCHRONIZER_H_
+#define FLUSH_AND_DATA_SYNCHRONIZER_H_
+
+#include "IFlushAndDataSynchronizer.h"
+#include <condition_variable>
+#include <map>
+#include <mutex>
+#include <vector>
+
+class FlushAndDataSynchronizer : public IFlushAndDataSynchronizer
+{
+    enum class FlushState
+    {
+        IDLE,
+        FLUSHING,
+        FLUSHED
+    };
+
+    enum class DataState
+    {
+        NO_DATA,
+        DATA_RECEIVED,
+        DATA_PUSHED
+    };
+
+    struct SourceState
+    {
+        FlushState flushState;
+        DataState dataState;
+    };
+
+    struct FlushData
+    {
+        int32_t sourceId;
+        bool resetTime;
+    };
+
+public:
+    FlushAndDataSynchronizer() = default;
+    ~FlushAndDataSynchronizer() override = default;
+
+    void addSource(int32_t sourceId) override;
+    void removeSource(int32_t sourceId) override;
+    void notifyFlushStarted(int32_t sourceId) override;
+    void notifyFlushCompleted(int32_t sourceId) override;
+    void notifyDataReceived(int32_t sourceId) override;
+    void notifyDataPushed(int32_t sourceId) override;
+    void waitIfRequired(int32_t sourceId) override;
+    bool isAnySourceFlushing() const override;
+
+private:
+    mutable std::mutex m_mutex;
+    std::condition_variable m_cv;
+    std::map<int32_t, SourceState> m_sourceStates;
+};
+
+#endif // FLUSH_AND_DATA_SYNCHRONIZER_H_

source/GStreamerMSEMediaPlayerClient.cpp

@@ -254,7 +254,8 @@ StateChangeResult GStreamerMSEMediaPlayerClient::play(int32_t sourceId)
                 return;
             }
 
-            if (m_serverPlaybackState == firebolt::rialto::PlaybackState::PLAYING)
+            if (m_serverPlaybackState == firebolt::rialto::PlaybackState::PLAYING ||
+                (m_serverPlaybackState == firebolt::rialto::PlaybackState::END_OF_STREAM && wasPlayingBeforeEos))
             {
                 GST_INFO("Server is already playing");
                 sourceIt->second.m_state = ClientState::PLAYING;
@@ -296,7 +297,8 @@ StateChangeResult GStreamerMSEMediaPlayerClient::play(int32_t sourceId)
             }
             else
             {
-                GST_WARNING("Not in PAUSED state in %u state", static_cast<uint32_t>(m_clientState));
+                GST_WARNING("Not in PAUSED state in client state %u state; server playback state: %u",
+                            static_cast<uint32_t>(m_clientState), static_cast<uint32_t>(m_serverPlaybackState));
             }
 
             result = StateChangeResult::SUCCESS_ASYNC;
@@ -387,9 +389,11 @@ void GStreamerMSEMediaPlayerClient::setPlaybackRate(double rate)
 
 void GStreamerMSEMediaPlayerClient::flush(int32_t sourceId, bool resetTime)
 {
+    m_flushAndDataSynchronizer.notifyFlushStarted(sourceId);
     m_backendQueue->callInEventLoop(
         [&]()
         {
+            wasPlayingBeforeEos = false;
             bool async{true};
             auto sourceIt = m_attachedSources.find(sourceId);
             if (sourceIt == m_attachedSources.end())
@@ -519,6 +523,7 @@ bool GStreamerMSEMediaPlayerClient::attachSource(std::unique_ptr<firebolt::rialt
                     m_attachedSources.emplace(source->getId(),
                                               AttachedSource(rialtoSink, bufferPuller, delegate, source->getType()));
                     delegate->setSourceId(source->getId());
+                    m_flushAndDataSynchronizer.addSource(source->getId());
                     bufferPuller->start();
                 }
             }
@@ -566,6 +571,7 @@ void GStreamerMSEMediaPlayerClient::removeSource(int32_t sourceId)
                 GST_WARNING("Remove source %d failed", sourceId);
             }
             m_attachedSources.erase(sourceId);
+            m_flushAndDataSynchronizer.removeSource(sourceId);
         });
 }
 
@@ -575,12 +581,14 @@ void GStreamerMSEMediaPlayerClient::handlePlaybackStateChange(firebolt::rialto::
     m_backendQueue->callInEventLoop(
         [&]()
         {
+            const auto kPreviousState{m_serverPlaybackState};
             m_serverPlaybackState = state;
             switch (state)
             {
             case firebolt::rialto::PlaybackState::PAUSED:
             case firebolt::rialto::PlaybackState::PLAYING:
             {
+                wasPlayingBeforeEos = false;
                 if (state == firebolt::rialto::PlaybackState::PAUSED && m_clientState == ClientState::AWAITING_PAUSED)
                 {
                     m_clientState = ClientState::PAUSED;
@@ -616,6 +624,10 @@ void GStreamerMSEMediaPlayerClient::handlePlaybackStateChange(firebolt::rialto::
             }
             case firebolt::rialto::PlaybackState::END_OF_STREAM:
             {
+                if (!wasPlayingBeforeEos && firebolt::rialto::PlaybackState::PLAYING == kPreviousState)
+                {
+                    wasPlayingBeforeEos = true;
+                }
                 for (const auto &source : m_attachedSources)
                 {
                     source.second.m_delegate->handleEos();
@@ -629,6 +641,7 @@ void GStreamerMSEMediaPlayerClient::handlePlaybackStateChange(firebolt::rialto::
             }
             case firebolt::rialto::PlaybackState::FAILURE:
             {
+                wasPlayingBeforeEos = false;
                 for (const auto &source : m_attachedSources)
                 {
                     source.second.m_delegate->handleError("Rialto server playback failed");
@@ -669,6 +682,7 @@ void GStreamerMSEMediaPlayerClient::handleSourceFlushed(int32_t sourceId)
             }
             sourceIt->second.m_isFlushing = false;
             sourceIt->second.m_delegate->handleFlushCompleted();
+            m_flushAndDataSynchronizer.notifyFlushCompleted(sourceId);
         });
 }
 
@@ -925,6 +939,11 @@ bool GStreamerMSEMediaPlayerClient::switchSource(const std::unique_ptr<firebolt:
     return result;
 }
 
+IFlushAndDataSynchronizer &GStreamerMSEMediaPlayerClient::getFlushAndDataSynchronizer()
+{
+    return m_flushAndDataSynchronizer;
+}
+
 bool GStreamerMSEMediaPlayerClient::checkIfAllAttachedSourcesInStates(const std::vector<ClientState> &states)
 {
     return std::all_of(m_attachedSources.begin(), m_attachedSources.end(), [states](const auto &source)
@@ -1177,6 +1196,11 @@ void PullBufferMessage::handle()
         status = firebolt::rialto::MediaSourceStatus::NO_AVAILABLE_SAMPLES;
     }
 
+    if (firebolt::rialto::MediaSourceStatus::OK == status || firebolt::rialto::MediaSourceStatus::EOS == status)
+    {
+        m_player->getFlushAndDataSynchronizer().notifyDataPushed(m_sourceId);
+    }
+
     m_player->m_backendQueue->postMessage(
         std::make_shared<HaveDataMessage>(status, m_sourceId, m_needDataRequestId, m_player));
 }

source/GStreamerMSEMediaPlayerClient.h

@@ -35,6 +35,7 @@
 
 #include "BufferParser.h"
 #include "Constants.h"
+#include "FlushAndDataSynchronizer.h"
 #include "IMediaPipeline.h"
 #include "IMessageQueue.h"
 #include "IPullModePlaybackDelegate.h"
@@ -322,6 +323,7 @@ class GStreamerMSEMediaPlayerClient : public firebolt::rialto::IMediaPipelineCli
     void setUseBuffering(bool useBuffering);
     bool getUseBuffering();
     bool switchSource(const std::unique_ptr<firebolt::rialto::IMediaPipeline::MediaSource> &source);
+    IFlushAndDataSynchronizer &getFlushAndDataSynchronizer();
 
 private:
     bool areAllStreamsAttached();
@@ -340,6 +342,8 @@ class GStreamerMSEMediaPlayerClient : public firebolt::rialto::IMediaPipelineCli
     int32_t m_videoStreams;
     int32_t m_subtitleStreams;
     firebolt::rialto::PlaybackInfo m_playbackInfo{-1, 1.0};
+    FlushAndDataSynchronizer m_flushAndDataSynchronizer;
+    bool wasPlayingBeforeEos{false};
 
     struct Rectangle
     {

source/IFlushAndDataSynchronizer.h

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2026 Sky UK
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation;
+ * version 2.1 of the License.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#ifndef IFLUSH_AND_DATA_SYNCHRONIZER_H_
+#define IFLUSH_AND_DATA_SYNCHRONIZER_H_
+
+#include <cstdint>
+
+class IFlushAndDataSynchronizer
+{
+public:
+    virtual ~IFlushAndDataSynchronizer() = default;
+
+    virtual void addSource(int32_t sourceId) = 0;
+    virtual void removeSource(int32_t sourceId) = 0;
+    virtual void notifyFlushStarted(int32_t sourceId) = 0;
+    virtual void notifyFlushCompleted(int32_t sourceId) = 0;
+    virtual void notifyDataReceived(int32_t sourceId) = 0;
+    virtual void notifyDataPushed(int32_t sourceId) = 0;
+    virtual void waitIfRequired(int32_t sourceId) = 0;
+    virtual bool isAnySourceFlushing() const = 0;
+};
+
+#endif // IFLUSH_AND_DATA_SYNCHRONIZER_H_