IbmLiveStreaming.h

Go to the documentation of this file.

// Copyright Epic Games, Inc. All Rights Reserved.
 
#pragma once
 
#include "CoreMinimal.h"
#include "GameplayMediaEncoder.h"
#include "Interfaces/IHttpRequest.h"
#include "GameplayMediaEncoderCommon.h"
 
#if defined(WITH_IBMRTMPINGEST) && LIVESTREAMING
 
// IBM RTMP Ingest
THIRD_PARTY_INCLUDES_START
    extern "C" {
        #include "rtmp_c/core/init.h"
        #include "rtmp_c/net/rtmp/rtmpclient.h"
        #include "rtmp_c/net/rtmp/modules/rtmpmodule_connect.h"
        #include "rtmp_c/net/rtmp/modules/rtmpmodule_broadcaster.h"
        #include "rtmp_c/core/rawdata.h"
        #include "rtmp_c/core/platform/threadfunc.h"
        #include "rtmp_c/codec/codecs.h"
    }
THIRD_PARTY_INCLUDES_END
 
class FJsonObject;
 
DECLARE_LOG_CATEGORY_EXTERN(IbmLiveStreaming, Log, All);
 
class FIbmLiveStreaming final : private IGameplayMediaEncoderListener
{
public:
    using FBandwidthProbeCallback = TFunction<void(uint32, bool)>;
 
    FIbmLiveStreaming();
    ~FIbmLiveStreaming();
 
    static FIbmLiveStreaming* Get();
 
    bool Start();
    bool Start(const FString& ClientId, const FString& ClientSecret, const FString& Channel, uint32 AudioSampleRate, uint32 AudioNumChannels, uint32 AudioBitrate);
    void Stop();
 
    static void StartCmd()
    {
        // We call Get(), so it creates the singleton
        Get()->Start();
    }
 
    static void StopCmd()
    {
        if (Singleton)
        {
            Singleton->Stop();
        }
    }
 
private:
    enum class EState
    {
        None,
        GettingAccessToken, // See https://developers.video.ibm.com/channel-api/getting-started.html#token-endpoint_8
        GettingIngestSettings, // See https://developers.video.ibm.com/channel-api/channel.html#ingest-settings_77
        Connecting,
        Connected,
        Stopping
    };
 
    struct FRtmpUrl
    {
        FString Host;
        FString Application;
        FString Channel;
        FString ApplicationName;
        FString StreamName;
        FString StreamNamePrefix;
    };
 
    struct FIngest
    {
        FString StreamingKey;
        FRtmpUrl RtmpUrl;
    };
 
    struct FFormData
    {
    public:
        FFormData();
        void AddField(const FString& Name, const FString& Value);
 
        // Call this in the end, to generate the HttpRequest
        template<typename T1, typename T2>
        TSharedRef<IHttpRequest> CreateHttpPostRequest(const FString& URL, T1&& TargetObj, T2&& TargetObjHandler)
        {
            auto HttpRequest = CreateHttpPostRequestImpl(URL);
            HttpRequest->OnProcessRequestComplete().BindRaw(std::forward<T1>(TargetObj), std::forward<T2>(TargetObjHandler));
            return HttpRequest;
        }
 
    private:
        static TArray<uint8> FStringToUint8(const FString& InString);
 
        FString BoundaryLabel;
        FString BoundaryBegin;
        FString BoundaryEnd;
        FString Data;
 
        TSharedRef<IHttpRequest> CreateHttpPostRequestImpl(const FString& URL);
    };
 
    // IGameplayMediaEncoderListener interface
    void OnMediaSample(const FGameplayMediaEncoderSample& Sample) override;
 
    void OnProcessRequestComplete(FHttpRequestPtr SourceHttpRequest, FHttpResponsePtr HttpResponse, bool bSucceeded);
 
    static bool GetJsonField(const TSharedPtr<FJsonObject>& JsonObject, const FString& FieldName, FString& DestStr);
    static bool GetJsonField(const TSharedPtr<FJsonObject>& JsonObject, const FString& FieldName, const TSharedPtr<FJsonObject>*& DestObj);
 
    void Connect();
 
    // Called to finalize a previous call to `Stop()`.  This is called once
    // the IBM library tells us the stream stopped and was deleted
    void FinishStopOnGameThread();
    void StopFromSocketThread();
 
    //
    // Callbacks for IBM RTMP C Ingest
    //
    static void OnConnectionError(RTMPModuleConnect* Module, RTMPEvent Evt, void* RejectInfoObj);
    static void OnConnectionSuccess(RTMPModuleConnect* Module);
    static void OnStreamPublished(RTMPModuleBroadcaster* Module);
    static void OnStreamDeleted(RTMPModuleBroadcaster* Module);
    static void OnStreamError(RTMPModuleBroadcaster* Module);
    static void OnStopPublish(RTMPModuleBroadcaster* Module);
    static void OnStreamBandwidthChanged(RTMPModuleBroadcaster* Module, unsigned long Bandwidth, int32 QueueWasEmpty);
    void OnConnectionErrorImpl(RTMPModuleConnect* Module, RTMPEvent Evt, void* RejectInfoObj);
    void OnConnectionSuccessImpl(RTMPModuleConnect* Module);
    void OnStreamPublishedImpl(RTMPModuleBroadcaster* Module);
    void OnStreamDeletedImpl(RTMPModuleBroadcaster* Module);
    void OnStreamErrorImpl(RTMPModuleBroadcaster* Module);
    void OnStopPublishImpl(RTMPModuleBroadcaster* Module);
    void OnStreamBandwidthChangedImpl(RTMPModuleBroadcaster* Module, uint32 Bandwidth, bool bQueueWasEmpty);
 
    static RawData* GetAvccHeader(const TArrayView<uint8>& DataView, const uint8** OutPpsEnd);
    static RawData* GetVideoPacket(const TArrayView<uint8>& Data, bool bVideoKeyframe, const uint8* DataBegin);
 
    void InjectVideo(uint32 TimestampMs, const TArrayView<uint8>& DataView, bool bIsKeyFrame);
    void InjectAudio(uint32 TimestampMs, const TArrayView<uint8>& DataView);
 
    template<typename T>
    static const uint8* Bytes(T& t)
    {
        return reinterpret_cast<const uint8*>(&t);
    }
 
    // Custom log function for the IBM librayr
    static void CustomLogMsg(const char* Msg);
 
    // Utility functions to log an error if we are in the wrong state
    bool CheckState(const wchar_t* FuncName, EState ExpectedState);
    bool CheckState(const wchar_t* FuncName, EState ExpectedStateMin, EState ExpectedStateMax);
 
    static FIbmLiveStreaming* Singleton;
 
    // Initialization parameters
    struct 
    {
        FString ClientId;
        FString ClientSecret;
        FString Channel;
        uint32 AudioSampleRate;
        uint32 AudioNumChannels;
        uint32 AudioBitrate;
    } Config;
 
    FIngest Ingest;
 
    TAtomic<EState> State = EState::None;
    TAtomic<uint32> AudioPacketsSent = 0;
    TAtomic<uint32> VideoPacketsSent = 0;
    FTimespan LiveStreamStartTimespan;
    uint32 FpsCalculationStartVideoPackets = 0;
    double FpsCalculationStartTime = 0;
 
    struct 
    {
        RTMPClient* Client = nullptr;
        RTMPModuleBroadcaster* BroadcasterModule = nullptr;
        RTMPModuleConnect* ConnectModule = nullptr;
    } Ctx;
    FCriticalSection CtxMutex;
 
    // Helper function that locks Ctx and make the IBM call
    void QueueFrame(RTMPContentType FrameType, RawData* Pkt, uint32 TimestampMs);
};
 
#endif // WITH_IBMRTMPINGEST