rtmpchunkstream.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/error.h>
#include <promeki/hashmap.h>
#include <promeki/mutex.h>
#include <promeki/namespace.h>
#include <promeki/objectbase.h>
#include <promeki/result.h>
#include <promeki/rtmpmessage.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class IODevice;
 
class RtmpChunkStream : public ObjectBase {
                PROMEKI_OBJECT(RtmpChunkStream, ObjectBase)
        public:
                static constexpr int DefaultChunkSize = 128;
 
                static constexpr int MinChunkSize = 128;
 
                static constexpr int MaxChunkSize = 65535;
 
                static constexpr int DefaultWindowAckSize = 5'000'000;
 
                explicit RtmpChunkStream(IODevice *device, ObjectBase *parent = nullptr);
 
                ~RtmpChunkStream() override;
 
                // ---- Write side ----
 
                Error writeMessage(const RtmpMessage &msg);
 
                // ---- Read side ----
 
                Result<RtmpMessage> readMessage(unsigned int timeoutMs = 0);
 
                // ---- Configuration ----
 
                int  localChunkSize() const  { return _localChunkSize.value(); }
 
                Error setLocalChunkSize(int bytes);
 
                int  peerChunkSize() const { return _peerChunkSize.value(); }
 
                int  localWindowAckSize() const { return _localWindowAckSize.value(); }
 
                Error setLocalWindowAckSize(int bytes);
 
                int  peerWindowAckSize() const { return _peerWindowAckSize.value(); }
 
                // ---- Counters ----
 
                int64_t bytesSent() const     { return _bytesSent.value(); }
 
                int64_t bytesReceived() const { return _bytesReceived.value(); }
 
                int64_t lastAckBytesAcked() const { return _lastAckBytesSent.value(); }
 
                // ---- Signals ----
 
                PROMEKI_SIGNAL(controlMessageReceived, RtmpMessage);
 
                PROMEKI_SIGNAL(peerChunkSizeChanged, int);
 
                PROMEKI_SIGNAL(peerAck, uint32_t);
 
        private:
                struct WriteState {
                        uint32_t timestamp = 0;     
                        uint32_t delta = 0;         
                        uint32_t messageLength = 0;
                        uint8_t  messageTypeId = 0;
                        uint32_t messageStreamId = 0;
                        bool     established = false;  
                };
 
                struct ReadState {
                        uint32_t timestamp = 0;
                        uint32_t delta = 0;
                        uint32_t messageLength = 0;
                        uint8_t  messageTypeId = 0;
                        uint32_t messageStreamId = 0;
                        bool     extendedTimestamp = false;  
                        bool     established = false;
                        Buffer   reassembly;        
                        uint32_t reassemblyBytes = 0;  
                };
 
                // ---- Internal helpers ----
 
                Error writeBytes(const uint8_t *data, size_t len);
                Error readBytesExact(uint8_t *data, size_t len, unsigned int timeoutMs);
                Error sendSetChunkSize(int bytes);
                Error sendWindowAckSize(int bytes);
                Error sendAck(uint32_t cumulative);
                void  handleControl(const RtmpMessage &msg);
 
                int  defaultCsidForType(RtmpMessage::Type type) const;
 
                IODevice *_device;
 
                Atomic<int32_t> _localChunkSize{DefaultChunkSize};
                Atomic<int32_t> _peerChunkSize{DefaultChunkSize};
                Atomic<int32_t> _localWindowAckSize{DefaultWindowAckSize};
                Atomic<int32_t> _peerWindowAckSize{0};
 
                Atomic<int64_t> _bytesSent{0};
                Atomic<int64_t> _bytesReceived{0};
                Atomic<int64_t> _lastAckBytesSent{0};
 
                Mutex _writeMutex;      
                Mutex _readMutex;       
                HashMap<uint32_t, WriteState> _writeStates;
                HashMap<uint32_t, ReadState>  _readStates;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK