rtmpsession.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NETWORK
#include <cstdint>
 
#include <promeki/amf0.h>
#include <promeki/atomic.h>
#include <promeki/enums.h>
#include <promeki/error.h>
#include <promeki/fourcc.h>
#include <promeki/hashmap.h>
#include <promeki/list.h>
#include <promeki/metadata.h>
#include <promeki/mutex.h>
#include <promeki/namespace.h>
#include <promeki/objectbase.h>
#include <promeki/result.h>
#include <promeki/rtmpchunkstream.h>
#include <promeki/rtmphandshake.h>
#include <promeki/rtmpmessage.h>
#include <promeki/string.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class IODevice;
 
struct RtmpConnectOptions {
                String       app;
                String       tcUrl;
                String       pageUrl;
                String       swfUrl;
                String       flashVer;
                String       type = "nonprivate";
                int          objectEncoding = 0;
                int          capabilities = 239;
                int          audioCodecs = 0x0FFF;
                int          videoCodecs = 0x00FF;
                int          videoFunction = 1;
                List<FourCC> fourCcList = { FourCC("hvc1") };
};
 
class RtmpSession : public ObjectBase {
                PROMEKI_OBJECT(RtmpSession, ObjectBase)
        public:
                static constexpr int DefaultPostConnectChunkSize = 60000;
 
                explicit RtmpSession(RtmpRole role, ObjectBase *parent = nullptr);
 
                ~RtmpSession() override;
 
                Error attach(IODevice *device);
 
                IODevice *device() const { return _device; }
 
                RtmpChunkStream *chunkStream() { return _chunk.get(); }
 
                Error performHandshake(unsigned int timeoutMs = 10000);
 
                // ----- High-level commands (client side) -----
 
                Error connect(const RtmpConnectOptions &opts, unsigned int timeoutMs = 10000);
 
                Result<uint32_t> createStream(unsigned int timeoutMs = 5000);
 
                Error publish(uint32_t streamId, const String &streamKey,
                              const String &mode = "live", unsigned int timeoutMs = 5000);
 
                Error play(uint32_t streamId, const String &streamKey,
                           double start = -2.0, double duration = -1.0,
                           unsigned int timeoutMs = 5000);
 
                Error deleteStream(uint32_t streamId);
 
                Error releaseStream(const String &streamKey);
 
                Error fcPublish(const String &streamKey);
 
                Error fcUnpublish(const String &streamKey);
 
                Error fcSubscribe(const String &streamKey);
 
                // ----- Raw access -----
 
                Error sendMessage(const RtmpMessage &m);
 
                Result<RtmpMessage> readMessage(unsigned int timeoutMs = 0);
 
                static Error onStatusToError(const String &code);
 
                // ----- Signals -----
 
                PROMEKI_SIGNAL(handshakeComplete);
 
                PROMEKI_SIGNAL(connected);
 
                PROMEKI_SIGNAL(connectionFailed, Error);
 
                PROMEKI_SIGNAL(streamCreated, uint32_t);
 
                PROMEKI_SIGNAL(publishStarted, uint32_t);
 
                PROMEKI_SIGNAL(playStarted, uint32_t);
 
                PROMEKI_SIGNAL(onStatus, Amf0Value);
 
                PROMEKI_SIGNAL(onMetaData, Metadata);
 
                PROMEKI_SIGNAL(audioMessageReceived, RtmpMessage);
 
                PROMEKI_SIGNAL(videoMessageReceived, RtmpMessage);
 
                PROMEKI_SIGNAL(disconnected);
 
        private:
                struct PendingTransaction {
                        Error    result = Error::Timeout;
                        bool     completed = false;
                        Amf0Value commandObject;   
                        Amf0Value info;             
                        uint32_t expectedMsid = 0;
                        String   commandName;
                };
 
                Error    sendCommand(uint32_t csid, uint32_t msid,
                                     const String &command, double txnId,
                                     const Amf0Value::List &args);
                Error    awaitTransaction(double txnId, unsigned int timeoutMs,
                                          PendingTransaction *outTxn = nullptr);
                Error    handleInboundCommand(const RtmpMessage &msg);
                Error    handleInboundUserControl(const RtmpMessage &msg);
                Error    handleInboundData(const RtmpMessage &msg);
                void     applyConnectFlowDefaults();
 
                Error    pumpUntilTransaction(double txnId, unsigned int timeoutMs,
                                              PendingTransaction *outTxn);
 
                double   nextTransactionId();
 
                IODevice              *_device = nullptr;
                RtmpRole               _role;
                RtmpHandshake          _handshake;
                UniquePtr<RtmpChunkStream> _chunk;
 
                Atomic<double>         _nextTxnId{1.0};
 
                Mutex                       _txnMutex;
                HashMap<double, PendingTransaction *> _pending;
 
                bool _connected = false;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK