Seek redesign: add implementation of Set source position (#283)

Summary: Remake Seek procedure. New method - setSourcePosition added.
Type: Feature
Test Plan: Unittests, Component Tests, YT Conformance Tests, NPLB
Jira: RIALTO-491

media/client/ipc/include/MediaPipelineIpc.h

@@ -105,6 +105,8 @@ class MediaPipelineIpc : public IMediaPipelineIpc, public IpcModule
 
     bool flush(int32_t sourceId, bool resetTime) override;
 
+    bool setSourcePosition(int32_t sourceId, int64_t position) override;
+
 private:
     /**
      * @brief The media player client ipc.

media/client/ipc/interface/IMediaPipelineIpc.h

@@ -240,6 +240,18 @@ class IMediaPipelineIpc
      * @retval true on success.
      */
     virtual bool flush(int32_t sourceId, bool resetTime) = 0;
+
+    /**
+     * @brief Set the source position in nanoseconds.
+     *
+     * This method sets the start position for a source.
+     *
+     * @param[in] sourceId  : The source id. Value should be set to the MediaSource.id returned after attachSource()
+     * @param[in] position : The position in nanoseconds.
+     *
+     * @retval true on success.
+     */
+    virtual bool setSourcePosition(int32_t sourceId, int64_t position) = 0;
 };
 
 }; // namespace firebolt::rialto::client

media/client/ipc/source/MediaPipelineIpc.cpp

@@ -552,7 +552,7 @@ bool MediaPipelineIpc::getPosition(int64_t &position)
     // check the result
     if (ipcController->Failed())
     {
-        RIALTO_CLIENT_LOG_ERROR("failed to set position due to '%s'", ipcController->ErrorText().c_str());
+        RIALTO_CLIENT_LOG_ERROR("failed to get position due to '%s'", ipcController->ErrorText().c_str());
         return false;
     }
 
@@ -777,6 +777,38 @@ bool MediaPipelineIpc::flush(int32_t sourceId, bool resetTime)
     return true;
 }
 
+bool MediaPipelineIpc::setSourcePosition(int32_t sourceId, int64_t position)
+{
+    if (!reattachChannelIfRequired())
+    {
+        RIALTO_CLIENT_LOG_ERROR("Reattachment of the ipc channel failed, ipc disconnected");
+        return false;
+    }
+
+    firebolt::rialto::SetSourcePositionRequest request;
+
+    request.set_session_id(m_sessionId);
+    request.set_source_id(sourceId);
+    request.set_position(position);
+
+    firebolt::rialto::SetSourcePositionResponse response;
+    auto ipcController = m_ipc.createRpcController();
+    auto blockingClosure = m_ipc.createBlockingClosure();
+    m_mediaPipelineStub->setSourcePosition(ipcController.get(), &request, &response, blockingClosure.get());
+
+    // wait for the call to complete
+    blockingClosure->wait();
+
+    // check the result
+    if (ipcController->Failed())
+    {
+        RIALTO_CLIENT_LOG_ERROR("failed to set source position due to '%s'", ipcController->ErrorText().c_str());
+        return false;
+    }
+
+    return true;
+}
+
 void MediaPipelineIpc::onPlaybackStateUpdated(const std::shared_ptr<firebolt::rialto::PlaybackStateChangeEvent> &event)
 {
     /* Ignore event if not for this session */

media/client/main/include/MediaPipeline.h

@@ -163,6 +163,8 @@ class MediaPipeline : public IMediaPipeline, public IMediaPipelineIpcClient, pub
 
     bool flush(int32_t sourceId, bool resetTime) override;
 
+    bool setSourcePosition(int32_t sourceId, int64_t position) override;
+
     void notifyApplicationState(ApplicationState state) override;
 
 protected:

media/client/main/source/MediaPipeline.cpp

@@ -488,6 +488,13 @@ bool MediaPipeline::flush(int32_t sourceId, bool resetTime)
     return false;
 }
 
+bool MediaPipeline::setSourcePosition(int32_t sourceId, int64_t position)
+{
+    RIALTO_CLIENT_LOG_DEBUG("entry:");
+
+    return m_mediaPipelineIpc->setSourcePosition(sourceId, position);
+}
+
 void MediaPipeline::discardNeedDataRequest(uint32_t needDataRequestId)
 {
     // Find the needDataRequest for this needDataRequestId

media/public/include/IMediaPipeline.h

@@ -1206,6 +1206,18 @@ class IMediaPipeline
      * @retval true on success.
      */
     virtual bool flush(int32_t sourceId, bool resetTime) = 0;
+
+    /**
+     * @brief Set the source position in nanoseconds.
+     *
+     * This method sets the start position for a source.
+     *
+     * @param[in] sourceId  : The source id. Value should be set to the MediaSource.id returned after attachSource()
+     * @param[in] position : The position in nanoseconds.
+     *
+     * @retval true on success.
+     */
+    virtual bool setSourcePosition(int32_t sourceId, int64_t position) = 0;
 };
 
 }; // namespace firebolt::rialto

media/server/gstplayer/CMakeLists.txt

@@ -58,6 +58,7 @@ add_library(
         source/tasks/generic/SetVideoGeometry.cpp
         source/tasks/generic/SetVolume.cpp
         source/tasks/generic/SetMute.cpp
+        source/tasks/generic/SetSourcePosition.cpp
         source/tasks/generic/Shutdown.cpp
         source/tasks/generic/Stop.cpp
         source/tasks/generic/Underflow.cpp

media/server/gstplayer/include/GenericPlayerContext.h

@@ -225,6 +225,20 @@ struct GenericPlayerContext
      * Attribute can be used only in worker thread
      */
     bool wereAllSourcesAttached{false};
+
+    /**
+     * @brief Flag used to check if FinishSetupSource is finished. It is needed to avoid need data overwriting.
+     *
+     * Attribute can be used only in worker thread
+     */
+    bool setupSourceFinished{false};
+
+    /**
+     * @brief Queued source positions. Used by SetSourcePosition task to request pushing new sample.
+     *
+     * Attribute can be used only in worker thread
+     */
+    std::map<GstElement *, std::uint64_t> initialPositions;
 };
 } // namespace firebolt::rialto::server
 

media/server/gstplayer/include/GstGenericPlayer.h

@@ -117,6 +117,7 @@ class GstGenericPlayer : public IGstGenericPlayer, public IGstGenericPlayerPriva
     bool getMute(bool &mute) override;
     void ping(std::unique_ptr<IHeartbeatHandler> &&heartbeatHandler) override;
     void flush(const MediaSourceType &mediaSourceType, bool resetTime) override;
+    void setSourcePosition(const MediaSourceType &mediaSourceType, int64_t position) override;
 
 private:
     void scheduleNeedMediaData(GstAppSrc *src) override;
@@ -225,6 +226,16 @@ class GstGenericPlayer : public IGstGenericPlayer, public IGstGenericPlayerPriva
      */
     bool setCodecData(GstCaps *caps, const std::shared_ptr<CodecData> &codecData) const;
 
+    /**
+     * @brief Pushes GstSample if playback position has changed or new segment needs to be sent.
+     *
+     * @param[in] source          : The Gst Source element, that should receive new sample
+     * @param[in] buffer          : The next GstBuffer to push
+     *
+     * @retval True if operation was performed
+     */
+    bool pushSampleIfRequired(GstElement *source, GstBuffer *buffer);
+
 private:
     /**
      * @brief The player context.

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

@@ -344,15 +344,26 @@ class IGenericPlayerTaskFactory
      * @brief Creates a Flush task.
      *
      * @param[in] context   : The GstPlayer context
-     * @param[in] player    : The GstGenericPlayer instance
      * @param[in] type      : The media source type to flush
      * @param[in] resetTime : True if time should be reset
      *
      * @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;
+    virtual std::unique_ptr<IPlayerTask>
+    createFlush(GenericPlayerContext &context, const firebolt::rialto::MediaSourceType &type, bool resetTime) const = 0;
+
+    /**
+     * @brief Creates a SetSourcePosition task.
+     *
+     * @param[in] context   : The GstPlayer context
+     * @param[in] type      : The media source type to set position
+     * @param[in] position  : The new source position
+     *
+     * @retval the new SetSourcePosition task instance.
+     */
+    virtual std::unique_ptr<IPlayerTask> createSetSourcePosition(GenericPlayerContext &context,
+                                                                 const firebolt::rialto::MediaSourceType &type,
+                                                                 std::int64_t position) const = 0;
 };
 
 } // namespace firebolt::rialto::server

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

@@ -22,7 +22,6 @@
 
 #include "GenericPlayerContext.h"
 #include "IGstGenericPlayerClient.h"
-#include "IGstGenericPlayerPrivate.h"
 #include "IGstWrapper.h"
 #include "IPlayerTask.h"
 #include <memory>
@@ -32,15 +31,14 @@ namespace firebolt::rialto::server::tasks::generic
 class Flush : public IPlayerTask
 {
 public:
-    Flush(GenericPlayerContext &context, IGstGenericPlayerPrivate &player, IGstGenericPlayerClient *client,
+    Flush(GenericPlayerContext &context, IGstGenericPlayerClient *client,
           std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> gstWrapper, const MediaSourceType &type,
           bool resetTime);
     ~Flush() override;
     void execute() const override;
 
 private:
     GenericPlayerContext &m_context;
-    IGstGenericPlayerPrivate &m_player;
     IGstGenericPlayerClient *m_gstPlayerClient;
     std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> m_gstWrapper;
     MediaSourceType m_type;

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

@@ -87,9 +87,11 @@ class GenericPlayerTaskFactory : public IGenericPlayerTaskFactory
     std::unique_ptr<IPlayerTask> createRenderFrame(GenericPlayerContext &context,
                                                    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,
+    std::unique_ptr<IPlayerTask> createFlush(GenericPlayerContext &context, const firebolt::rialto::MediaSourceType &type,
                                              bool resetTime) const override;
+    std::unique_ptr<IPlayerTask> createSetSourcePosition(GenericPlayerContext &context,
+                                                         const firebolt::rialto::MediaSourceType &type,
+                                                         std::int64_t position) const override;
 
 private:
     IGstGenericPlayerClient *m_client;

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

@@ -0,0 +1,50 @@
+/*
+ * If not stated otherwise in this file or this component's LICENSE file the
+ * following copyright and licenses apply:
+ *
+ * Copyright 2024 Sky UK
+ *
+ * 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 FIREBOLT_RIALTO_SERVER_TASKS_GENERIC_SET_SOURCE_POSITION_H_
+#define FIREBOLT_RIALTO_SERVER_TASKS_GENERIC_SET_SOURCE_POSITION_H_
+
+#include "GenericPlayerContext.h"
+#include "IGstGenericPlayerClient.h"
+#include "IGstWrapper.h"
+#include "IPlayerTask.h"
+#include <cstdint>
+#include <memory>
+
+namespace firebolt::rialto::server::tasks::generic
+{
+class SetSourcePosition : public IPlayerTask
+{
+public:
+    SetSourcePosition(GenericPlayerContext &context, IGstGenericPlayerClient *client,
+                      const std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> &gstWrapper,
+                      const MediaSourceType &type, std::int64_t position);
+    ~SetSourcePosition() override;
+    void execute() const override;
+
+private:
+    GenericPlayerContext &m_context;
+    IGstGenericPlayerClient *m_gstPlayerClient;
+    std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> m_gstWrapper;
+    MediaSourceType m_type;
+    std::int64_t m_position;
+};
+} // namespace firebolt::rialto::server::tasks::generic
+
+#endif // FIREBOLT_RIALTO_SERVER_TASKS_GENERIC_SET_SOURCE_POSITION_H_

media/server/gstplayer/interface/IGstGenericPlayer.h

@@ -260,6 +260,16 @@ class IGstGenericPlayer
      *
      */
     virtual void flush(const MediaSourceType &mediaSourceType, bool resetTime) = 0;
+
+    /**
+     * @brief Set the source position in nanoseconds.
+     *
+     * This method sets the start position for a source.
+     *
+     * @param[in] mediaSourceType : The media source type to flush.
+     * @param[in] position : The position in nanoseconds.
+     */
+    virtual void setSourcePosition(const MediaSourceType &mediaSourceType, int64_t position) = 0;
 };
 
 }; // namespace firebolt::rialto::server