rtmpclient.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NETWORK
#include <cstdint>
 
#include <promeki/atomic.h>
#include <promeki/duration.h>
#include <promeki/error.h>
#include <promeki/flvtag.h>
#include <promeki/list.h>
#include <promeki/metadata.h>
#include <promeki/mutex.h>
#include <promeki/namespace.h>
#include <promeki/objectbase.h>
#include <promeki/queue.h>
#include <promeki/result.h>
#include <promeki/rtmpmessage.h>
#include <promeki/rtmpsession.h>
#include <promeki/string.h>
#include <promeki/uniqueptr.h>
#include <promeki/url.h>
 
#if PROMEKI_ENABLE_TLS
#include <promeki/sslcontext.h>
#endif
 
PROMEKI_NAMESPACE_BEGIN
 
class TcpSocket;
class Thread;
 
class RtmpClient : public ObjectBase {
                PROMEKI_OBJECT(RtmpClient, ObjectBase)
        public:
                static constexpr int DefaultSendQueueDepth = 64;
 
                static constexpr int DefaultReadQueueDepth = 64;
 
                static constexpr unsigned int DefaultConnectTimeoutMs = 10000;
 
                explicit RtmpClient(ObjectBase *parent = nullptr);
 
                ~RtmpClient() override;
 
#if PROMEKI_ENABLE_TLS
                void setSslContext(SslContext ctx) { _sslContext = std::move(ctx); }
#endif
 
                Error open(const Url &url,
                           const RtmpConnectOptions &opts = {},
                           unsigned int timeoutMs = DefaultConnectTimeoutMs);
 
                Error publish(const String &streamKey = String(),
                              const String &mode = "live",
                              unsigned int timeoutMs = 5000);
 
                Error play(const String &streamKey = String(),
                           unsigned int timeoutMs = 5000,
                           bool useFcSubscribe = false);
 
                Error sendVideo(const FlvVideoTag &tag, uint32_t timestampMs);
 
                Error sendAudio(const FlvAudioTag &tag, uint32_t timestampMs);
 
                Error sendMetadata(const Metadata &meta, uint32_t timestampMs = 0);
 
                Result<FlvVideoTag> takeVideo(unsigned int timeoutMs);
 
                Result<FlvAudioTag> takeAudio(unsigned int timeoutMs);
 
                Result<Metadata>    takeMetadata(unsigned int timeoutMs);
 
                Error close();
 
                const Url &url() const { return _url; }
 
                const String &app() const { return _app; }
 
                const String &streamKey() const { return _streamKey; }
 
                uint32_t streamId() const { return _streamId.value(); }
 
                bool isOpen() const { return _open.value(); }
 
                RtmpSession *session() { return _session.get(); }
 
                // ---- Stats ----
 
                int64_t bytesSent() const     { return _bytesSent.value(); }
 
                int64_t bytesReceived() const { return _bytesReceived.value(); }
 
                int64_t videoMessagesSent() const { return _videoMessagesSent.value(); }
 
                int64_t audioMessagesSent() const { return _audioMessagesSent.value(); }
 
                int64_t videoMessagesReceived() const { return _videoMessagesReceived.value(); }
 
                int64_t audioMessagesReceived() const { return _audioMessagesReceived.value(); }
 
                Duration rttEstimate() const { return Duration(); }
 
                // ---- Signals ----
 
                PROMEKI_SIGNAL(connected);
 
                PROMEKI_SIGNAL(disconnected, Error);
 
                PROMEKI_SIGNAL(metadataReceived, Metadata);
 
                static void splitPath(const Url &url, String &app, String &streamKey);
 
        private:
                class WriterThread;
                class ReaderThread;
 
                void teardown(Error reason);
                void startMediaThreads();
 
                Url                          _url;
                String                       _app;
                String                       _streamKey;
                Atomic<uint32_t>             _streamId{0};
                Atomic<bool>                 _open{false};
                Atomic<bool>                 _stopping{false};
 
                UniquePtr<TcpSocket>         _socket;       
                UniquePtr<RtmpSession>       _session;
                UniquePtr<WriterThread>      _writer;
                UniquePtr<ReaderThread>      _reader;
 
#if PROMEKI_ENABLE_TLS
                SslContext              _sslContext;
                bool                         _isTls = false;
#endif
 
                Queue<RtmpMessage>           _writeQueue;
                Queue<FlvVideoTag>           _videoRxQueue;
                Queue<FlvAudioTag>           _audioRxQueue;
                Queue<Metadata>              _metadataRxQueue;
 
                Atomic<int64_t>              _bytesSent{0};
                Atomic<int64_t>              _bytesReceived{0};
                Atomic<int64_t>              _videoMessagesSent{0};
                Atomic<int64_t>              _audioMessagesSent{0};
                Atomic<int64_t>              _videoMessagesReceived{0};
                Atomic<int64_t>              _audioMessagesReceived{0};
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK