downlink

package
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 2, 2026 License: Apache-2.0, MIT Imports: 14 Imported by: 0

Documentation

Overview

Package downlink wraps a BusSubscription with per-subscriber telemetry: write-duration tracking via a lock-free histogram, RTCP feedback, and RTP header extension patching.

Downlink satisfies the session.BusSubscription interface so it can be stored directly in subscriptionEntry without an extra adapter.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func DiscoverAbsCaptureTimeID 

func DiscoverAbsCaptureTimeID(sdp string) uint8

DiscoverAbsCaptureTimeID parses an SDP string and returns the extension ID assigned to the abs-capture-time URI. Returns 0 if the extension is absent.

func DiscoverAbsSendTimeID 

func DiscoverAbsSendTimeID(sdp string) uint8

DiscoverAbsSendTimeID parses an SDP string and returns the extension ID assigned to the abs-send-time URI. Returns 0 if the extension is absent.

Types

type AudioCadencePacer 

type AudioCadencePacer struct {
	// contains filtered or unexported fields
}

AudioCadencePacer smooths clustered Opus arrivals according to their RTP timestamps. Its queue is bounded and enqueue is non-blocking. One worker owns the timer and writer, preserving packet order without locking the writer.

func (*AudioCadencePacer) Enqueue 

func (p *AudioCadencePacer) Enqueue(pkt *rtp.Packet) error

func (*AudioCadencePacer) EnqueueSource 

func (p *AudioCadencePacer) EnqueueSource(
	pkt *rtp.Packet,
	source session.SourceIdentity,
	captureTime time.Time,
	captureTimeKnown bool,
) error

func (*AudioCadencePacer) HandleNACK 

func (p *AudioCadencePacer) HandleNACK(seq uint16)

func (*AudioCadencePacer) RequiresPacketCopy 

func (*AudioCadencePacer) RequiresPacketCopy() bool

func (*AudioCadencePacer) SetRTT 

func (p *AudioCadencePacer) SetRTT(rtt time.Duration)

func (*AudioCadencePacer) Snapshot 

func (p *AudioCadencePacer) Snapshot() PacerSnapshot

func (*AudioCadencePacer) Stop 

func (p *AudioCadencePacer) Stop()
type Downlink struct {
	Extensions ExtensionMapper
	// contains filtered or unexported fields
}

Downlink wraps a BusSubscription with per-subscriber telemetry.

WriteRTP may be called concurrently from multiple forwardLoop goroutines when the subscriber uses BusMix (receives from all buses). writeHist uses lock-free atomics; no mutex is acquired on the write path. Snapshot reads all stats atomically and is safe from any goroutine. 

func NewForwardingDownlink(id string, inner session.PacketWriter, feedback *FeedbackReader) *Downlink

NewForwardingDownlink is the production downlink contract. It preserves ordered single-writer forwarding, reconnect-safe RTP translation, NACK recovery, and telemetry without applying a second RTP cadence schedule.

func NewPassthrough 

func NewPassthrough(id string, inner session.PacketWriter) *Downlink

NewPassthrough creates a minimal Downlink with no RTCP reader and no abs-send-time patching. Used where no RTPSender is available (WHIP ingest, unit tests).

func (*Downlink) ConfigureExtensions 

func (d *Downlink) ConfigureExtensions(localSDP string)

ConfigureExtensions records the header-extension IDs negotiated in the subscriber leg's local SDP answer. It must run before the downlink is exposed to the relay's forwarding loop.

func (*Downlink) HandleNACK 

func (d *Downlink) HandleNACK(seq uint16)

HandleNACK schedules a cached subscriber-hop packet for retransmission.

func (*Downlink) ObserveRTT 

func (d *Downlink) ObserveRTT(rtt time.Duration)

ObserveRTT stores a valid sender-side RTT sample reported by Pion's remote inbound RTP stats. RTT is derived from RTCP SR/RR timing per RFC 3550.

func (*Downlink) RequiresPacketCopy 

func (d *Downlink) RequiresPacketCopy() bool

RequiresPacketCopy reports whether WriteRTP mutates the packet header for subscriber-specific negotiated state.

func (*Downlink) SetFeedback 

func (d *Downlink) SetFeedback(fr *FeedbackReader)

SetFeedback attaches a FeedbackReader after construction. Must be called before any concurrent WriteRTP calls.

func (*Downlink) SetSenderTimeline 

func (d *Downlink) SetSenderTimeline(timeline *clocklineage.Timeline)

SetSenderTimeline attaches Pion's outgoing Sender Report timeline for this subscriber. It must be called before the downlink is published to the session.

func (*Downlink) Snapshot 

func (d *Downlink) Snapshot() session.SubscriptionSnapshot

Snapshot returns a point-in-time view of this Downlink's telemetry. BusID is left empty here; RelaySession fills it from the subscription entry. All reads are atomic; safe to call from any goroutine.

func (*Downlink) Stats 

func (d *Downlink) Stats() *SenderStats

Stats returns a pointer to this Downlink's SenderStats so callers can direct a FeedbackReader to update the same counters that Snapshot reads.

func (*Downlink) Stop 

func (d *Downlink) Stop()

Stop shuts down the FeedbackReader goroutine. Safe to call even if no FeedbackReader was attached.

Only call Stop() after the underlying peer connection has been closed, or after ReadRTCP() is guaranteed to return (e.g. due to an error). Calling Stop() with an open, idle PC will block until ReadRTCP returns.

func (*Downlink) StopForwarding 

func (d *Downlink) StopForwarding()

StopForwarding terminates the serialized forwarding worker without waiting for RTCP. The feedback reader exits when its PeerConnection closes.

func (*Downlink) WriteRTP 

func (d *Downlink) WriteRTP(pkt *rtp.Packet) error

WriteRTP satisfies session.BusSubscription.

The inner.WriteRTP call is always lock-free on the hot path. Write duration is recorded in the lock-free histogram: one compare+atomic.Add.

func (*Downlink) WriteRTPWithCaptureTime 

func (d *Downlink) WriteRTPWithCaptureTime(pkt *rtp.Packet, captureTime time.Time, known bool) error

WriteRTPWithCaptureTime attaches genuine publisher clock lineage when it is known, then transfers ownership to the configured pacer.

func (*Downlink) WriteRTPWithSource 

func (d *Downlink) WriteRTPWithSource(
	pkt *rtp.Packet,
	captureTime time.Time,
	known bool,
	source session.SourceIdentity,
) error

WriteRTPWithSource carries the publisher attachment identity into the single-writer downlink worker for reconnect-safe RTP translation.

type ExtensionMapper 

type ExtensionMapper struct {
	// contains filtered or unexported fields
}

ExtensionMapper holds the negotiated RTP header extension IDs. Zero value is safe: all IDs default to 0 (disabled).

func (*ExtensionMapper) HasAbsCaptureTime 

func (e *ExtensionMapper) HasAbsCaptureTime() bool

HasAbsCaptureTime reports whether abs-capture-time was negotiated in the SDP answer and is therefore expected to be present in forwarded packets.

func (*ExtensionMapper) MutatesPackets 

func (e *ExtensionMapper) MutatesPackets() bool

MutatesPackets reports whether this mapper patches outbound RTP packets.

func (*ExtensionMapper) PatchAbsCaptureTime 

func (e *ExtensionMapper) PatchAbsCaptureTime(pkt *rtp.Packet, captureTime time.Time) bool

PatchAbsCaptureTime writes the publisher-derived capture clock into the negotiated abs-capture-time extension. It returns true only when a packet was patched. The fixed-size payload avoids allocation on the forwarding path.

func (*ExtensionMapper) PatchAbsSendTime 

func (e *ExtensionMapper) PatchAbsSendTime(pkt *rtp.Packet)

PatchAbsSendTime writes the current wall-clock time as the middle 24 bits of an NTP timestamp (6-bit seconds, 18-bit fraction) into the packet's abs-send-time header extension. No-op if the extension ID is 0.

Hot-path cost: ~50 ns (time.Now + 2 arithmetic ops + pkt.SetExtension).

func (*ExtensionMapper) SetAbsCaptureTimeID 

func (e *ExtensionMapper) SetAbsCaptureTimeID(id uint8)

SetAbsCaptureTimeID configures the extension slot for abs-capture-time. id == 0 means the extension was not negotiated.

func (*ExtensionMapper) SetAbsSendTimeID 

func (e *ExtensionMapper) SetAbsSendTimeID(id uint8)

SetAbsSendTimeID configures the extension slot for abs-send-time patching. id == 0 disables patching (default).

type FeedbackReader 

type FeedbackReader struct {
	// contains filtered or unexported fields
}

FeedbackReader reads RTCP from an RTCPSource in a dedicated goroutine and updates a SenderStats. It also invokes an optional RecvReportCallback so callers can react to loss (e.g., emit a CODEC_HINT or ICE_RESTART).

The goroutine exits automatically when src.ReadRTCP returns an error.

func StartFeedbackReader 

func StartFeedbackReader(
	src RTCPSource,
	stats *SenderStats,
	onReport RecvReportCallback,
	onNACK NackCallback,
) *FeedbackReader

StartFeedbackReader launches the RTCP reader goroutine and returns immediately. stats must not be nil. onReport may be nil.

func (*FeedbackReader) Stop 

func (fr *FeedbackReader) Stop()

Stop signals the reader goroutine to exit and waits for it to finish. Safe to call multiple times; only the first call has effect.

type NackCallback 

type NackCallback func(seq uint16)

NackCallback receives each subscriber-hop RTP sequence number requested by RTCP NACK.

type Pacer 

type Pacer interface {
	Enqueue(pkt *rtp.Packet) error
	EnqueueSource(pkt *rtp.Packet, source session.SourceIdentity, captureTime time.Time, captureTimeKnown bool) error
	HandleNACK(seq uint16)
	SetRTT(rtt time.Duration)
	Stop()
	Snapshot() PacerSnapshot
	RequiresPacketCopy() bool
}

Pacer schedules packets for one subscriber. Enqueue must not block.

type PacerSnapshot 

type PacerSnapshot struct {
	Mode                              string
	EnqueuedCount                     uint64
	SentCount                         uint64
	QueueFullDrops                    uint64
	StaleDrops                        uint64
	QueueDepth                        uint64
	QueuePeak                         uint64
	MaxQueueAgeNs                     uint64
	QueueAgeP95Ms                     float64
	LastSpacingNs                     uint64
	SpacingP50Ms                      float64
	SpacingP95Ms                      float64
	MaxSpacingNs                      uint64
	MaxTimerWaitNs                    uint64
	MaxTimerOversleepNs               uint64
	MaxWriterDurationNs               uint64
	WriterBlockedCount                uint64
	PaddingPacketCount                uint64
	SourcePaddingStrippedCount        uint64
	LatePacketDropCount               uint64
	LatePaddingDropCount              uint64
	LateMediaDropCount                uint64
	GapTimeoutCount                   uint64
	RecoveryPacketCount               uint64
	TimelineResets                    uint64
	NackQueueDrops                    uint64
	NackCacheHits                     uint64
	NackCacheMisses                   uint64
	NackThrottled                     uint64
	RetransmitSentCount               uint64
	RetransmitErrorCount              uint64
	OutputSequenceDiscontinuityCount  uint64
	OutputTimestampDiscontinuityCount uint64
	OutputMaxSequenceDelta            uint64
	OutputMaxTimestampDeltaSamples    uint64
}

PacerSnapshot is a point-in-time view of bounded downlink worker state.

type RTCPSource 

type RTCPSource interface {
	ReadRTCP() ([]rtcp.Packet, interceptor.Attributes, error)
}

RTCPSource is the read side of RTCP for a WebRTC sender. *webrtc.RTPSender satisfies this interface directly.

type RecvReportCallback 

type RecvReportCallback func(fractionLost float64)

RecvReportCallback is invoked for each RTCP ReceptionReport block received. fractionLost is in [0.0, 1.0] (RTCP raw byte / 256).

type SenderStats 

type SenderStats struct {
	PacketsSent         atomic.Uint64
	BytesSent           atomic.Uint64
	WriteErrCount       atomic.Uint64
	NackCount           atomic.Uint64
	TwccFeedbackCount   atomic.Uint64 // TransportLayerCC packets received
	ReceiverReportCount atomic.Uint64
	RttLastUs           atomic.Int64  // microseconds; -1 if unknown
	JitterLastUs        atomic.Int64  // microseconds
	FractionLostLast    atomic.Uint32 // raw RTCP fraction 0-255
	AbsCapturePatched   atomic.Uint64
}

SenderStats holds per-subscriber delivery counters. All fields are accessed atomically so any goroutine may read a snapshot while the forwarding goroutine writes.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.
⌂ 换网址