pcapflowrouter.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_NETWORK
#include <cstdint>
#include <utility>
#include <promeki/namespace.h>
#include <promeki/ancpacket.h>
#include <promeki/buffer.h>
#include <promeki/datetime.h>
#include <promeki/duration.h>
#include <promeki/error.h>
#include <promeki/function.h>
#include <promeki/list.h>
#include <promeki/packetdemux.h>
#include <promeki/pcapsdpmap.h>
#include <promeki/rtppacket.h>
#include <promeki/rtpseqtracker.h>
#include <promeki/rxpayloadbundle.h>
#include <promeki/socketaddress.h>
#include <promeki/string.h>
#include <promeki/uniqueptr.h>
 
PROMEKI_NAMESPACE_BEGIN
 
class SdpSession;
 
class PcapFlowRouter {
        public:
                struct RoutedAncFrame {
                                SocketAddress src;
                                SocketAddress dst;
                                uint32_t ssrc = 0;
                                DateTime captureTime;
                                RxAncFrame anc;
                };
 
                using AncFrameCallback = Function<void(const RoutedAncFrame &)>;
 
                struct FlowStat {
                                SocketAddress dst;      
                                uint32_t ssrc = 0;      
                                uint8_t payloadType = 0;
                                PcapFlowKind kind = PcapFlowKind::Unknown; 
                                uint64_t packets = 0;   
                                uint64_t bytes = 0;     
                                // -- RFC 3550 health (via RtpSeqTracker), cumulative over this SSRC --
                                uint64_t lostPackets = 0;       
                                uint64_t duplicatePackets = 0;  
                                uint64_t reorderedPackets = 0;  
                                uint32_t timestampRegressions = 0; 
                                Duration maxJitter = Duration::zero(); 
                };
 
                struct RtpAnomaly {
                                enum class Kind {
                                        SsrcChange,          
                                        PayloadTypeChange,   
                                        PacketLoss,          
                                        Reorder,             
                                        Duplicate,           
                                        TimestampRegression, 
                                        JitterExceeded,      
                                };
                                Kind          kind = Kind::SsrcChange; 
                                PcapFlowKind  flowKind = PcapFlowKind::Unknown; 
                                SocketAddress dst;             
                                uint32_t      ssrc = 0;        
                                uint32_t      previous = 0;
                                uint32_t      count = 0;       
                                Duration      jitter = Duration::zero(); 
                                DateTime      captureTime;     
                };
 
                using RtpAnomalyCallback = Function<void(const RtpAnomaly &)>;
 
                PcapFlowRouter() = default;
 
                Error setSdp(const SdpSession &sdp);
 
                void addAncFlow(const SocketAddress &dst, int payloadType = -1) {
                        _map.addAncFlow(dst.address(), dst.port(), payloadType);
                }
 
                void onAncFrame(AncFrameCallback cb) { _ancCb = std::move(cb); }
 
                void onRtpAnomaly(RtpAnomalyCallback cb) { _anomalyCb = std::move(cb); }
 
                void setJitterWarnThreshold(const Duration &threshold) { _jitterWarn = threshold; }
 
                Error processFile(const String &path);
 
                Error processBuffer(const Buffer &buf);
 
                const List<FlowStat> &flowStats() const { return _stats; }
 
                const PcapSdpMap &sdpMap() const { return _map; }
 
                void reset();
 
        private:
                struct AncReasm {
                                SocketAddress dst;
                                SocketAddress src;
                                uint32_t ssrc = 0;
                                bool haveSsrc = false;
                                bool haveTs = false;
                                uint32_t timestamp = 0;
                                uint8_t payloadType = 0;
                                AncDesc desc;
                                RtpPacket::List packets;
                                DateTime captureTime;
                };
 
                struct FlowHealth {
                                SocketAddress dst;
                                uint32_t      ssrc = 0;
                                bool          haveSsrc = false;
                                uint8_t       payloadType = 0;
                                bool          havePt = false;
                                uint32_t      lastTimestamp = 0;
                                bool          haveTs = false;
                                uint32_t      lastExtendedSeq = 0;
                                bool          haveSeq = false;
                                // §A.8 jitter state (computed directly in ns; see trackRtpHealth).
                                int64_t       prevArrivalNs = 0;
                                uint32_t      prevRtpTs = 0;
                                int64_t       jitterNs = 0;
                                bool          haveJitterPrev = false;
                                Duration      maxJitter = Duration::zero();
                                bool          jitterOver = false; 
                                uint32_t      timestampRegressions = 0;
                                RtpSeqTracker tracker;
                };
 
                void handleDatagram(const UdpDatagram &dg, const DateTime &captureTime);
                void routeAnc(const UdpDatagram &dg, const PcapFlow &flow, const RtpPacket &pkt, const DateTime &captureTime);
                void flushAnc(AncReasm &r);
                FlowStat &statFor(const SocketAddress &dst, uint32_t ssrc, uint8_t pt, PcapFlowKind kind);
                AncReasm &ancReasmFor(const UdpDatagram &dg, const PcapFlow &flow);
                FlowHealth &healthFor(const SocketAddress &dst);
                void trackRtpHealth(const SocketAddress &dst, PcapFlowKind kind, const RtpPacket &pkt, FlowStat &st,
                                    const DateTime &captureTime);
                void emitAnomaly(RtpAnomaly::Kind kind, PcapFlowKind flowKind, const SocketAddress &dst, uint32_t ssrc,
                                 uint32_t previous, uint32_t count, const DateTime &captureTime,
                                 const Duration &jitter = Duration::zero());
 
                Error runReader(class PcapReader &reader);
 
                PcapSdpMap _map;
                PacketDemux _demux;
                AncFrameCallback _ancCb;
                RtpAnomalyCallback _anomalyCb;
                Duration _jitterWarn = Duration::zero(); 
                List<FlowStat> _stats;
                List<AncReasm> _ancFlows;
                List<UniquePtr<FlowHealth>> _health;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_NETWORK