巴延興

深圳開鴻數字產業發展有限公司

資深OS框架開發工程師

一、 簡介

多媒體播放框架主要的實現在PlayerServer服務中，這個服務提供了媒體播放框架所需要的實現環境，繼續跟蹤代碼分析發現，PlayerServer主要通過gstreamer適配層，對gstreamer進行調用。gstreamer屬于更加具體的實現，所以本篇文章主要是分析PlayerServer通過適配層調用到gstreamer的過程。

此前，我在《OpenHarmony 3.2 Beta多媒體系列-音視頻播放框架》一文中，主要分析了多媒體播放的框架層代碼，本地接口通過服務端的proxy代理類進行IPC調用，最終調用到PlayerServer服務端。本篇主要分析了多媒體gstreamer的調用，涉及到從PlayerServer到gstreamer的整體流程。

二、 目錄

  gstreamer
    ├── BUILD.gn
    ├── common
    │   ├── BUILD.gn
    │   ├── playbin_adapter
    │   │   ├── i_playbin_ctrler.h
    │   │   ├── playbin2_ctrler.cpp
    │   │   ├── playbin2_ctrler.h
    │   │   ├── playbin_ctrler_base.cpp
    │   │   ├── playbin_ctrler_base.h
    │   │   ├── playbin_msg_define.h
    │   │   ├── playbin_sink_provider.h
    │   │   ├── playbin_state.cpp
    │   │   ├── playbin_state.h
    │   │   ├── playbin_task_mgr.cpp
    │   │   └── playbin_task_mgr.h
    │   ├── state_machine
    │   │   ├── state_machine.cpp
    │   │   └── state_machine.h
    ├── factory
    │   ├── BUILD.gn
    │   └── engine_factory.cpp
    └── player
        ├── BUILD.gn
        ├── player_codec_ctrl.cpp
        ├── player_codec_ctrl.h
        ├── player_engine_gst_impl.cpp
        ├── player_engine_gst_impl.h
        ├── player_sinkprovider.cpp
        ├── player_sinkprovider.h
        ├── player_track_parse.cpp
        └── player_track_parse.h

目錄主要是多媒體子系統中的engine部分，涉及到了gstreamer的適配層，gstreamer具體的實現是在third_party/gstreamer目錄中。

三 、Gstreamer介紹

1. 簡介
Gstreamer是一個跨平臺的多媒體框架，應用程序可以通過管道（Pipeline）的方式，將多媒體處理的各個步驟串聯起來，達到預期的效果。每個步驟通過元素（Element）基于GObject對象系統通過插件（plugins）的方式實現，方便了各項功能的擴展。

2.Gstreamer幾個重要的概念
Element

Element是Gstreamer中最重要的對象類型之一。一個element實現一個功能（讀取文件，解碼，輸出等），程序需要創建多個element，并按順序將其串聯起來，構成一個完整的Pipeline。

Pad

Pad是一個element的輸入/輸出接口，分為src pad（生產數據）和sink pad（消費數據）兩種。兩個element必須通過pad才能連接起來，pad擁有當前element能處理數據類型的能力（capabilities），會在連接時通過比較src pad和sink pad中所支持的能力，來選擇最恰當的數據類型用于傳輸，如果element不支持，程序會直接退出。在element通過pad連接成功后，數據會從上一個element的src pad傳到下一個element的sink pad然后進行處理。

Bin和Pipeline

Bin是一個容器，用于管理多個element，改變bin的狀態時，bin會自動去修改所包含的element的狀態，也會轉發所收到的消息。如果沒有bin，我們需要依次操作我們所使用的element。通過bin降低了應用的復雜度。

Pipeline繼承自bin，為程序提供一個bus用于傳輸消息，并且對所有子element進行同步。當將Pipeline的狀態設置為PLAYING時，Pipeline會在一個/多個新的線程中通過element處理數據。

四、調用流程

?

左右滑動查看更多

五、源碼分析

1. PrepareAsync分析

首先，在PlayerServer的PrepareAsync中會調用OnPrepare(false)，具體是在OnPrepare(false)中實現，參數傳入false，表明調用的是異步方法。

int32_tPlayerServer::PrepareAsync()
{
    std::lock_guard<std::mutex> lock(mutex_);
    MEDIA_LOGW("KPI-TRACE: PlayerServer PrepareAsync in");


    if (lastOpStatus_ == PLAYER_INITIALIZED || lastOpStatus_ == PLAYER_STOPPED) {
        return OnPrepare(false);
    } else {
        MEDIA_LOGE("Can not Prepare, currentState is %{public}s", GetStatusDescription(lastOpStatus_).c_str());
        return MSERR_INVALID_OPERATION;
    }
}

OnPrepare方法中，先通過playerEngine_調用SerVideoSurface的方法，將surface_設置到PlayerEngineGstImpl中（producerSurface_），接著啟動一個任務，調用目前狀態的Prepare()方法。

int32_tPlayerServer::OnPrepare(boolsync)
{
    CHECK_AND_RETURN_RET_LOG(playerEngine_ != nullptr, MSERR_NO_MEMORY, "playerEngine_ is nullptr");
    int32_t ret = MSERR_OK;


#ifdef SUPPORT_VIDEO
    if (surface_ != nullptr) {
        ret = playerEngine_->SetVideoSurface(surface_);
        CHECK_AND_RETURN_RET_LOG(ret == MSERR_OK, MSERR_INVALID_OPERATION, "Engine SetVideoSurface Failed!");
    }
#endif


    lastOpStatus_ = PLAYER_PREPARED;


    auto preparedTask = std::make_sharedint32_t<>>([this]() {
        MediaTrace::TraceBegin("PlayerServer::PrepareAsync", FAKE_POINTER(this));
        auto currState = std::static_pointer_cast(GetCurrState());
        return currState->Prepare();
    });


    ret = taskMgr_.LaunchTask(preparedTask, PlayerServerTaskType::STATE_CHANGE);
    CHECK_AND_RETURN_RET_LOG(ret == MSERR_OK, ret, "Prepare launch task failed");


    if (sync) {
        (void)preparedTask->GetResult(); // wait HandlePrpare
    }
    return MSERR_OK;
}

進入Preparing狀態后，會觸發PlayerServer的HandlePrepare()方法被調用，在這個方法里會通過playerEngine_調用PrepareAsync方法，這個方法調用的是PlayerEngineGstImpl對應的PrepareAsync方法。

int32_tPlayerServer::HandlePrepare()
{
    int32_t ret = playerEngine_->PrepareAsync();
    CHECK_AND_RETURN_RET_LOG(ret == MSERR_OK, MSERR_INVALID_OPERATION, "Server Prepare Failed!");
    if (config_.leftVolume <= 1.0f || config_.rightVolume <= 1.0f) {
        ret = playerEngine_->SetVolume(config_.leftVolume, config_.rightVolume);
        MEDIA_LOGD("Prepared SetVolume leftVolume:%{public}f rightVolume:%{public}f, ret:%{public}d", 
                   config_.leftVolume, config_.rightVolume, ret);
    }
    (void)playerEngine_->SetLooping(config_.looping);


    {
        auto rateTask = std::make_sharedvoid<>>([this]() {
            auto currState = std::static_pointer_cast(GetCurrState());
            (void)currState->SetPlaybackSpeed(config_.speedMode);
        });


        (void)taskMgr_.LaunchTask(rateTask, PlayerServerTaskType::RATE_CHANGE);
    }
    return MSERR_OK;
}

首先初始化playBinCtrler_，后續的操作都是通過PlayBinCtrlerBase對象來操作的，所以PlayBinCtrlerInit()方法會創建PlayBinCtrlerBase對象（playBinCtrler_），創建好以后通過playBinCtrler_進行SetSource和SetXXXListener的設置。

int32_tPlayerEngineGstImpl::PrepareAsync()
{
    std::unique_lock<std::mutex> lock(mutex_);
    MEDIA_LOGD("Prepare in");


    int32_t ret = PlayBinCtrlerInit();
    CHECK_AND_RETURN_RET_LOG(ret == MSERR_OK, MSERR_INVALID_VAL, "PlayBinCtrlerInit failed");


    CHECK_AND_RETURN_RET_LOG(playBinCtrler_ != nullptr, MSERR_INVALID_VAL, "playBinCtrler_ is nullptr");
    ret = playBinCtrler_->PrepareAsync();
    CHECK_AND_RETURN_RET_LOG(ret == MSERR_OK, ret, "PrepareAsync failed");


    // The duration of some resources without header information cannot be obtained.
    MEDIA_LOGD("Prepared ok out");
    return MSERR_OK;
}

初始化完成以后，接下來進行playBinCtrler_的PrepareAsync的調用，PlayBinCtrlerBase中的PrepareAsync的方法間接地調用了PrepareAsyncInternal。

int32_tPlayBinCtrlerBase::PrepareAsync()
{
    MEDIA_LOGD("enter");


    std::unique_lock<std::mutex> lock(mutex_);
    return PrepareAsyncInternal();
}

PrepareAsyncInternal首先判斷當前的狀態，如果是preparingState或preparedState，那么就直接返回成功，否則繼續向下調用。接下來會調用EnterInitializedState(),這個方法中會創建playbin，設置signal的回調以及gstreamer參數的設置。最后調用目前狀態的Prepare方法，此時的狀態是InitializedState。

int32_t PlayBinCtrlerBase::PrepareAsyncInternal()
{
    if ((GetCurrState() == preparingState_) || (GetCurrState() == preparedState_)) {
        MEDIA_LOGI("already at preparing state, skip");
        return MSERR_OK;
    }


    CHECK_AND_RETURN_RET_LOG((!uri_.empty() || appsrcWrap_), MSERR_INVALID_OPERATION, "Set uri firsty!");


    int32_t ret = EnterInitializedState();
    CHECK_AND_RETURN_RET(ret == MSERR_OK, ret);


    auto currState = std::static_pointer_cast(GetCurrState());
    ret = currState->Prepare();
    CHECK_AND_RETURN_RET_LOG(ret == MSERR_OK, ret, "PrepareAsyncInternal failed");


    return MSERR_OK;
}

InitializedState的Prepare方法又通過ctrler_調回到PlayBinCtrlerBase的ChangeState方法，這個方法是在PlayBinCtrlerBase的父類StateMachine中，它是一個狀態機，管理著各種狀態的切換。

int32_t PlayBinCtrlerBase::Prepare()
{
    ctrler_.ChangeState(ctrler_.preparingState_);
    return MSERR_OK;
}

很多表示狀態的類在PlayBinCtrlerBase中進行聲明，這些子類的具體實現功能在playbin_state.cpp中。

private:
    class BaseState;
    class IdleState;
    class InitializedState;
    class PreparingState;
    class PreparedState;
    class PlayingState;
    class PausedState;
    class StoppedState;
    class StoppingState;
    class PlaybackCompletedState;

接下來看一下狀態機的ChangeState方法，可以看出切換狀態的時候，先調用切換前狀態的StateExit()方法，再調用切換后狀態的StateEnter()。如果需要一些操作，我們可以在狀態的StateEnter和StateExit中進行。

voidStateMachine::ChangeState(conststd::shared_ptr&state)
{
    ......
    if (currState_ != nullptr && currState_->GetStateName() == "stopping_state" && state->GetStateName() != "stopped_state") {
        return;
    }
    if (currState_) {
        currState_->StateExit();
    }
    currState_ = state;
    state->StateEnter();
}

因為上面切換狀態調用的是ctrler_.ChangeState(ctrler_.preparingState_)，所以接下來看一下PreparingState狀態的StateEnter方法。這個方法中首先是調用了ctrler_.ReportMessage(msg)，字面上看是用來上報msg信息的。

voidPlayBinCtrlerBase::StateEnter()
{
    PlayBinMessage msg = { PLAYBIN_MSG_SUBTYPE, PLAYBIN_SUB_MSG_BUFFERING_START, 0, {} };
    ctrler_.ReportMessage(msg);


    GstStateChangeReturn ret;
    (void)ChangePlayBinState(GST_STATE_PAUSED, ret);


    MEDIA_LOGD("PreparingState::StateEnter finished");
}

ctrler_是PlayBinCtrlerBase類型的變量，直接看PlayBinCtrlerBase的ReportMessage方法，這個方法的核心，是創建一個任務后，將任務放入消息隊列中，等待消息被處理，這里我們最想知道的是這個消息會在什么地方被處理。msgReportHandler創建了TaskHandler，這個里面會調用notifier_(msg)，這里的notifier_比較重要，我們可以順著這個變量向上分析。

voidPlayBinCtrlerBase::ReportMessage(constPlayBinMessage&msg)
{
    ......
    auto msgReportHandler = std::make_sharedvoid<>>([this, msg]() { notifier_(msg); });
    int32_t ret = msgQueue_->EnqueueTask(msgReportHandler);
    if (ret != MSERR_OK) {
        MEDIA_LOGE("async report msg failed, type: %{public}d, subType: %{public}d, code: %{public}d",
                   msg.type, msg.subType, msg.code);
    };


    if (msg.type == PlayBinMsgType::PLAYBIN_MSG_EOS) {
        ProcessEndOfStream();
    }
}

notifier_是在PlayBinCtrlerBase被創建的時候賦值的。

PlayBinCtrlerBase::PlayBinCtrlerBase(constPlayBinCreateParam&createParam)
    : renderMode_(createParam.renderMode),
    notifier_(createParam.notifier),
    sinkProvider_(createParam.sinkProvider)
{
    MEDIA_LOGD("enter ctor, instance: 0x%{public}06" PRIXPTR "", FAKE_POINTER(this));
}

在源碼分析的前期PlayerEngineGstImpl初始化PlayBinCtrlerBase的時候進行了創建notifier = std::bind(&PlayerEngineGstImpl::OnNotifyMessage, this, std::_1) notifier相當于是調用了PlayerEngineGstImpl::OnNotifyMessage方法。所以上述中的處理函數就是PlayerEngineGstImpl::OnNotifyMessage。

int32_tPlayerEngineGstImpl::PlayBinCtrlerPrepare()
{
    uint8_t renderMode = IPlayBinCtrler::DEFAULT_RENDER;
    auto notifier = std::bind(&PlayerEngineGstImpl::OnNotifyMessage, this, std::_1);


    {
        std::unique_lock<std::mutex> lk(trackParseMutex_);
        sinkProvider_ = std::make_shared(producerSurface_);
        sinkProvider_->SetAppInfo(appuid_, apppid_);
    }


    IPlayBinCtrler::PlayBinCreateParam createParam = {
        static_cast(renderMode), notifier, sinkProvider_
    };
    playBinCtrler_ = IPlayBinCtrler::PLAYBIN2, createParam);
    ......
    return MSERR_OK;
}

在OnNotifyMessage中指定了各種消息類型對應的執行函數，上述代碼中創建的Message類型是PLAYBIN_MSG_SUBTYPE，子類型為PLAYBIN_SUB_MSG_BUFFERING_START。

voidPlayerEngineGstImpl::OnNotifyMessage(constPlayBinMessage&msg)
{
    const std::unordered_map MSG_NOTIFY_FUNC_TABLE = {,>
        { PLAYBIN_MSG_ERROR, std::bind(&PlayerEngineGstImpl::HandleErrorMessage, this, std::_1) },
        { PLAYBIN_MSG_SEEKDONE, std::bind(&PlayerEngineGstImpl::HandleSeekDoneMessage, this, std::_1) },
        { PLAYBIN_MSG_SPEEDDONE, std::bind(&PlayerEngineGstImpl::HandleInfoMessage, this, std::_1) },
        { PLAYBIN_MSG_BITRATEDONE, std::bind(&PlayerEngineGstImpl::HandleInfoMessage, this, std::_1)},
        { PLAYBIN_MSG_EOS, std::bind(&PlayerEngineGstImpl::HandleInfoMessage, this, std::_1) },
        { PLAYBIN_MSG_STATE_CHANGE, std::bind(&PlayerEngineGstImpl::HandleInfoMessage, this, std::_1) },
        { PLAYBIN_MSG_SUBTYPE, std::bind(&PlayerEngineGstImpl::HandleSubTypeMessage, this, std::_1) },
        { PLAYBIN_MSG_AUDIO_SINK, std::bind(&PlayerEngineGstImpl::HandleAudioMessage, this, std::_1) },
        { PLAYBIN_MSG_POSITION_UPDATE, std::bind(&PlayerEngineGstImpl::HandlePositionUpdateMessage, this,
            std::_1) },
    };
    if (MSG_NOTIFY_FUNC_TABLE.count(msg.type) != 0) {
        MSG_NOTIFY_FUNC_TABLE.at(msg.type)(msg);
    }
}

最終的流程走到了PlayerEngineGstImpl::HandleBufferingStart()，在這個方法中，主要通過obs_將format傳給IPlayerEngineObs的OnInfo方法。

voidPlayerEngineGstImpl::HandleBufferingStart()
{
    percent_ = 0;
    Format format;
(void)format.PutIntValue(std::string(PlayerKeys::PLAYER_BUFFERING_START), 0);
    std::shared_ptr notifyObs = obs_.lock();
    if (notifyObs != nullptr) {
        notifyObs->OnInfo(INFO_TYPE_BUFFERING_UPDATE, 0, format);
    }
}

我們重點看一下obs_是哪里設置的，在PlayerServer的初始化InitPlayEngine。shared_from_this()相當于是把PlayerServer自身賦值給obs，PlayerServer也是實現了IPlayerEngineObs對應的接口。

int32_tPlayerServer::InitPlayEngine(conststd::string&url)
{
    ......
    int32_t ret = taskMgr_.Init();
    auto engineFactory = EngineFactoryRepo::SCENE_PLAYBACK, url);
    
    playerEngine_ = engineFactory->CreatePlayerEngine(appUid_, appPid_);


    if (dataSrc_ == nullptr) {
        ret = playerEngine_->SetSource(url);
    } else {
        ret = playerEngine_->SetSource(dataSrc_);
    }


    std::shared_ptr obs = shared_from_this();
    ret = playerEngine_->SetObs(obs);


    lastOpStatus_ = PLAYER_INITIALIZED;
    ChangeState(initializedState_);


    return MSERR_OK;
}

這樣我們就跟蹤到了PlayerServer的OnInfo()方法。

voidPlayerServer::OnInfo(PlayerOnInfoTypetype,int32_textra,constFormat&infoBody)
{
    std::lock_guard<std::mutex> lockCb(mutexCb_);


    int32_t ret = HandleMessage(type, extra, infoBody);
    if (playerCb_ != nullptr && ret == MSERR_OK) {
        playerCb_->OnInfo(type, extra, infoBody);
    }
}

2. Play分析

從PlayerServer開始跟蹤，調用到PlayerServer的OnPlay()方法。

int32_tPlayerServer::Play()
{
    ......
    if (lastOpStatus_ == PLAYER_PREPARED || lastOpStatus_ == PLAYER_PLAYBACK_COMPLETE ||
        lastOpStatus_ == PLAYER_PAUSED) {
        return OnPlay();
    } else {
        return MSERR_INVALID_OPERATION;
    }
}

在OnPlay中會啟動一個任務，在任務中獲取當前的狀態，然后調用當前狀態的Play()方法。

int32_tPlayerServer::OnPlay()
{
    ......
    auto playingTask = std::make_sharedvoid<>>([this]() {
        auto currState = std::static_pointer_cast(GetCurrState());
        (void)currState->Play();
    });


    int ret = taskMgr_.LaunchTask(playingTask, PlayerServerTaskType::STATE_CHANGE);


    lastOpStatus_ = PLAYER_STARTED;
    return MSERR_OK;
}

前面調用了PrepareAsync，所以當前的狀態是Prepared，調用到了PreparedState的Play()方法，這個方法還是按照之前Prepare的方式，調回到PlayerServer的HandlePlay()。

int32_tPlayerServer::Play()
{
    return server_.HandlePlay();
}

在PlayServer中通過播放引擎繼續向下調用。

int32_tPlayerServer::HandlePlay()
{
    int32_t ret = playerEngine_->Play();
    CHECK_AND_RETURN_RET_LOG(ret == MSERR_OK, MSERR_INVALID_OPERATION, "Engine Play Failed!");


    return MSERR_OK;
}

在PlayerEngineGstImpl的Play()方法會繼續調用playBinCtrler_的Play()方法。

int32_tPlayerEngineGstImpl::Play()
{
    ......
    playBinCtrler_->Play();
    return MSERR_OK;
}

PlayBinCtrlerBase的Play()方法根據當前的State，調用currSate->Play()。

int32_tPlayBinCtrlerBase::Play()
{
    ......
    auto currState =    std::static_pointer_cast(GetCurrState());
    int32_t ret = currState->Play();


    return MSERR_OK;
}

在PreparedState的Play()方法中改變了PlayBin的狀態為playing。

int32_tPlayBinCtrlerBase::Play()
{
    GstStateChangeReturn ret;
    return ChangePlayBinState(GST_STATE_PLAYING, ret);
}

ChangePlayBinState主要是調用了gst_element_set_state(GST_ELEMENT_CAST(ctrler_.playbin_),GST_STATE_PLAYING),這個直接調用了gstreamer三方庫的實現，調用完這個方法以后，gstreamer就開始進行播放了。

int32_tPlayBinCtrlerBase::ChangePlayBinState(GstStatetargetState,GstStateChangeReturn&ret)
{
    ......
    ret = gst_element_set_state(GST_ELEMENT_CAST(ctrler_.playbin_), targetState);
    if (ret == GST_STATE_CHANGE_FAILURE) {
        MEDIA_LOGE("Failed to change playbin's state to %{public}s", gst_element_state_get_name(targetState));
        return MSERR_INVALID_OPERATION;
    }


    return MSERR_OK;
}

六、總結

本篇文章主要從PlayerServer播放服務開始分析音視頻播放的流程，涉及到gstreamer引擎的調用，相對于多媒體播放框架來說，更加底層，便于熟悉從框架到gstreamer的整體流程。

深圳開鴻數字產業發展有限公司（簡稱“深開鴻”）于2021年成立于中國深圳，以數字化、智慧化改變人類的生產和生活方式為愿景，專注于自主軟件根技術的研發與持續創新，致力于打造萬物智聯核心技術、定義萬物智聯標準、引領萬物智聯時代發展。

深開鴻基于OpenHarmony，創新打造互通互聯互享的KaihongOS數字底座，上承可視可管可控的超級設備管理平臺，靈活擴展，柔性組合，聚合成全場景超級設備解決方案，實現更大范圍的以軟件定義硬件，引領智慧基建、智慧康養、智慧能源、智慧交通、智慧制造、智慧政務、智慧金融、智慧教育等多個行業變革，賦能、賦智、賦值千行百業的數智化轉型。

從開源中來，到行業中去，深開鴻以構筑行業數字化生態、培養生態人才為己任，持續突破行業邊界，立志成為萬物智聯時代的“國之重器”。

以數字化、智慧化改變人類的生產和生活方式

點擊在看和點贊，與更多的美好相遇 ↓