Asure correct data & flush order during postponed flushes (#436)

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

Author: skywojciechowskim

media/server/gstplayer/include/FlushOnPrerollController.h

@@ -21,6 +21,7 @@
 #define FIREBOLT_RIALTO_SERVER_FLUSH_ONPREROLL_CONTROLLER_H_
 
 #include "IFlushOnPrerollController.h"
+#include <condition_variable>
 #include <mutex>
 #include <optional>
 #include <set>
@@ -37,13 +38,16 @@ class FlushOnPrerollController : public IFlushOnPrerollController
     FlushOnPrerollController() = default;
     ~FlushOnPrerollController() override = default;
 
-    bool shouldPostponeFlush(const MediaSourceType &type) const override;
-    void setFlushing(const MediaSourceType &type, const GstState &currentPipelineState) override;
+    void waitIfRequired(const MediaSourceType &type) override;
+    void setFlushing(const MediaSourceType &type) override;
+    void setPrerolling() override;
     void stateReached(const GstState &newPipelineState) override;
+    void setTargetState(const GstState &state) override;
     void reset() override;
 
 private:
-    mutable std::mutex m_mutex{};
+    std::mutex m_mutex{};
+    std::condition_variable m_conditionVariable{};
     std::set<MediaSourceType> m_flushingSources{};
     std::optional<GstState> m_targetState{std::nullopt};
     bool m_isPrerolled{false};

media/server/gstplayer/include/GenericPlayerContext.h

@@ -270,7 +270,7 @@ struct GenericPlayerContext
     /**
      * @brief Workaround for the gstreamer flush issue
      */
-    FlushOnPrerollController flushOnPrerollController;
+    std::shared_ptr<IFlushOnPrerollController> flushOnPrerollController{std::make_shared<FlushOnPrerollController>()};
 };
 } // namespace firebolt::rialto::server
 

media/server/gstplayer/include/GstDispatcherThread.h

@@ -34,13 +34,15 @@ class GstDispatcherThreadFactory : public IGstDispatcherThreadFactory
     ~GstDispatcherThreadFactory() override = default;
     std::unique_ptr<IGstDispatcherThread>
     createGstDispatcherThread(IGstDispatcherThreadClient &client, GstElement *pipeline,
+                              const std::shared_ptr<IFlushOnPrerollController> &flushOnPrerollController,
                               const std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> &gstWrapper) const override;
 };
 
 class GstDispatcherThread : public IGstDispatcherThread
 {
 public:
     GstDispatcherThread(IGstDispatcherThreadClient &client, GstElement *pipeline,
+                        const std::shared_ptr<IFlushOnPrerollController> &flushOnPrerollController,
                         const std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> &gstWrapper);
     ~GstDispatcherThread() override;
 
@@ -58,6 +60,11 @@ class GstDispatcherThread : public IGstDispatcherThread
      */
     IGstDispatcherThreadClient &m_client;
 
+    /**
+     * @brief The flush on preroll controller.
+     */
+    std::shared_ptr<IFlushOnPrerollController> m_flushOnPrerollController;
+
     /**
      * @brief The gstreamer wrapper object.
      */

media/server/gstplayer/include/GstGenericPlayer.h

@@ -195,8 +195,6 @@ class GstGenericPlayer : public IGstGenericPlayer, public IGstGenericPlayerPriva
     GstElement *getSink(const MediaSourceType &mediaSourceType) const override;
     void setSourceFlushed(const MediaSourceType &mediaSourceType) override;
     bool isAsync(const MediaSourceType &mediaSourceType) const;
-    void postponeFlush(const MediaSourceType &mediaSourceType, bool resetTime) override;
-    void executePostponedFlushes() override;
     void notifyPlaybackInfo() override;
 
 private:
@@ -426,11 +424,6 @@ class GstGenericPlayer : public IGstGenericPlayer, public IGstGenericPlayerPriva
      */
     std::unique_ptr<IFlushWatcher> m_flushWatcher;
 
-    /**
-     * @brief The postponed flush tasks
-     */
-    std::vector<std::pair<MediaSourceType, bool>> m_postponedFlushes{};
-
     /**
      * @brief The ongoing state change operations counter
      */

media/server/gstplayer/include/IFlushOnPrerollController.h

@@ -34,9 +34,11 @@ class IFlushOnPrerollController
 public:
     virtual ~IFlushOnPrerollController() = default;
 
-    virtual bool shouldPostponeFlush(const MediaSourceType &type) const = 0;
-    virtual void setFlushing(const MediaSourceType &type, const GstState &currentPipelineState) = 0;
+    virtual void waitIfRequired(const MediaSourceType &type) = 0;
+    virtual void setFlushing(const MediaSourceType &type) = 0;
+    virtual void setPrerolling() = 0;
     virtual void stateReached(const GstState &newPipelineState) = 0;
+    virtual void setTargetState(const GstState &state) = 0;
     virtual void reset() = 0;
 };
 } // namespace firebolt::rialto::server

media/server/gstplayer/include/IGstDispatcherThread.h

@@ -20,6 +20,7 @@
 #ifndef FIREBOLT_RIALTO_SERVER_I_GST_DISPATCHER_THREAD_H_
 #define FIREBOLT_RIALTO_SERVER_I_GST_DISPATCHER_THREAD_H_
 
+#include "IFlushOnPrerollController.h"
 #include "IGstDispatcherThreadClient.h"
 #include "IGstWrapper.h"
 #include <memory>
@@ -36,6 +37,7 @@ class IGstDispatcherThreadFactory
 
     virtual std::unique_ptr<IGstDispatcherThread>
     createGstDispatcherThread(IGstDispatcherThreadClient &client, GstElement *pipeline,
+                              const std::shared_ptr<IFlushOnPrerollController> &flushOnPrerollController,
                               const std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> &gstWrapper) const = 0;
 };
 

media/server/gstplayer/include/IGstGenericPlayerPrivate.h

@@ -321,19 +321,6 @@ class IGstGenericPlayerPrivate
      */
     virtual void setSourceFlushed(const MediaSourceType &mediaSourceType) = 0;
 
-    /**
-     * @brief Postpones flush for the given source type
-     *
-     * @param[in] mediaSourceType : the source type that has been flushed
-     * @param[in] resetTime       : whether to reset the time after flush
-     */
-    virtual void postponeFlush(const MediaSourceType &mediaSourceType, bool resetTime) = 0;
-
-    /**
-     * @brief Queues postponed flushes for execution
-     */
-    virtual void executePostponedFlushes() = 0;
-
     /**
      * @brief Sends PlaybackInfo notification. Called by the worker thread.
      */

media/server/gstplayer/include/tasks/IGenericPlayerTaskFactory.h

@@ -429,12 +429,13 @@ class IGenericPlayerTaskFactory
      * @param[in] context   : The GstPlayer context
      * @param[in] type      : The media source type to flush
      * @param[in] resetTime : True if time should be reset
+     * @param[in] isAsync     : True if flushed source is asynchronous
      *
      * @retval the new Flush task instance.
      */
     virtual std::unique_ptr<IPlayerTask> createFlush(GenericPlayerContext &context, IGstGenericPlayerPrivate &player,
-                                                     const firebolt::rialto::MediaSourceType &type,
-                                                     bool resetTime) const = 0;
+                                                     const firebolt::rialto::MediaSourceType &type, bool resetTime,
+                                                     bool isAsync) const = 0;
 
     /**
      * @brief Creates a SetSourcePosition task.

media/server/gstplayer/include/tasks/generic/Flush.h

@@ -33,8 +33,8 @@ class Flush : public IPlayerTask
 {
 public:
     Flush(GenericPlayerContext &context, IGstGenericPlayerPrivate &player, IGstGenericPlayerClient *client,
-          std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> gstWrapper, const MediaSourceType &type,
-          bool resetTime);
+          const std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> &gstWrapper, const MediaSourceType &type,
+          bool resetTime, bool isAsync);
     ~Flush() override;
     void execute() const override;
 
@@ -45,6 +45,7 @@ class Flush : public IPlayerTask
     std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> m_gstWrapper;
     MediaSourceType m_type;
     bool m_resetTime;
+    bool m_isAsync;
 };
 } // namespace firebolt::rialto::server::tasks::generic
 

media/server/gstplayer/include/tasks/generic/GenericPlayerTaskFactory.h

@@ -108,8 +108,8 @@ class GenericPlayerTaskFactory : public IGenericPlayerTaskFactory
                                                    IGstGenericPlayerPrivate &player) const override;
     std::unique_ptr<IPlayerTask> createPing(std::unique_ptr<IHeartbeatHandler> &&heartbeatHandler) const override;
     std::unique_ptr<IPlayerTask> createFlush(GenericPlayerContext &context, IGstGenericPlayerPrivate &player,
-                                             const firebolt::rialto::MediaSourceType &type,
-                                             bool resetTime) const override;
+                                             const firebolt::rialto::MediaSourceType &type, bool resetTime,
+                                             bool isAsync) const override;
     std::unique_ptr<IPlayerTask> createSetSourcePosition(GenericPlayerContext &context,
                                                          const firebolt::rialto::MediaSourceType &type,
                                                          std::int64_t position, bool resetTime, double appliedRate,