forwarding

package
v1.9.2 Latest Latest
Warning

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

 Go to latest Published: Oct 6, 2026 License: MIT Imports: 33 Imported by: 0

Documentation

Overview

Package forwarding adapts mature RTP forwarding components without owning room authority, capture or a second congestion controller.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func PublicationRID 

func PublicationRID(layer, count int) string

Types

type EgressStats 

type EgressStats struct {
	// Drops include canceled probe padding and queued packets retired at close.
	Packets, RTPBytes, DroppedPackets uint64
}

type EncodedSource 

type EncodedSource struct {
	*Source
	// contains filtered or unexported fields
}

EncodedSource feeds complete codec outputs into the same forwarding buffers used for received RTP. Its capture/decoder owner serializes calls and fences replacement input; no network, capture or wall-clock scheduler lives here.

func NewEncodedSource 

func NewEncodedSource(options SourceOptions) (*EncodedSource, error)

func (*EncodedSource) BeginFrame 

func (input *EncodedSource) BeginFrame(pts time.Duration, at time.Time) error

func (*EncodedSource) BeginGeneration 

func (input *EncodedSource) BeginGeneration()

func (*EncodedSource) CodecBudget 

func (input *EncodedSource) CodecBudget(layer int, rtpBudget int64) int64

CodecBudget converts source-RTP allocation units to encoder AU-payload units. Actual packetization includes codec descriptors; bitrate comes from the existing tracker window so dormant time does not dilute the overhead rate. Downstream extensions and SRTP/UDP overhead are outside this source measure.

func (*EncodedSource) WriteFrame 

func (input *EncodedSource) WriteFrame(layer int, frame encoded.Frame) error

type LayerFormat 

type LayerFormat struct {
	Width, Height, Bitrate uint32
}

type Output 

type Output struct {
	*sfu.DownTrack
	// contains filtered or unexported fields
}

Output owns allocation for one video stream on one child connection. Its physical owner closes it; budget changes and upstream events do not create another scheduler. Do not attach a StreamAllocator to the same DownTrack.

func NewOutput 

func NewOutput(track *sfu.DownTrack, budget int64) *Output

func (*Output) BeginFrame 

func (output *Output) BeginFrame()

BeginFrame fences a controlled Native input before any of its outputs arrive. Raw RTP has no such boundary and keeps DownTrack's ordinary switching path.

func (*Output) Close 

func (output *Output) Close()

func (*Output) ReceiverRestart 

func (output *Output) ReceiverRestart(receiver sfu.TrackReceiver)

func (*Output) Reconcile 

func (output *Output) Reconcile()

func (*Output) Resync 

func (output *Output) Resync()

func (*Output) SetBudget 

func (output *Output) SetBudget(budget int64)

func (*Output) SetMaxSpatialLayer 

func (output *Output) SetMaxSpatialLayer(layer int32)

func (*Output) SetMaxTemporalLayer 

func (output *Output) SetMaxTemporalLayer(layer int32)

func (*Output) State 

func (output *Output) State() Snapshot

func (*Output) UpTrackBitrateAvailabilityChange 

func (output *Output) UpTrackBitrateAvailabilityChange()

func (*Output) UpTrackBitrateReport 

func (output *Output) UpTrackBitrateReport(layers []int32, rates sfu.Bitrates)

func (*Output) UpTrackLayersChange 

func (output *Output) UpTrackLayersChange()

func (*Output) UpTrackMaxPublishedLayerChange 

func (output *Output) UpTrackMaxPublishedLayerChange(layer int32)

func (*Output) UpTrackMaxTemporalLayerSeenChange 

func (output *Output) UpTrackMaxTemporalLayerSeenChange(layer int32)

func (*Output) WriteRTP 

func (output *Output) WriteRTP(packet *buffer.ExtPacket, layer int32) int32

type Publication 

type Publication struct {
	PC *webrtc.PeerConnection
	// contains filtered or unexported fields
}

Publication sends existing representations over one simulcast connection. LiveKit's allocator owns all track allocations and probes on the shared BWE.

func NewPublication 

func NewPublication(options TransportOptions) (_ *Publication, err error)

func (*Publication) Close 

func (publication *Publication) Close() error

func (*Publication) Counters 

func (publication *Publication) Counters() PublicationCounters

func (*Publication) CurrentSource 

func (publication *Publication) CurrentSource() *Source

func (*Publication) LowestLayerBudget 

func (publication *Publication) LowestLayerBudget() int64

func (*Publication) OnBindAndConnected 

func (*Publication) OnBindAndConnected()

func (*Publication) OnCodecNegotiated 

func (*Publication) OnCodecNegotiated(webrtc.RTPCodecCapability)

func (*Publication) OnDownTrackClose 

func (*Publication) OnDownTrackClose(bool)

func (*Publication) OnMaxSubscribedLayerChanged 

func (*Publication) OnMaxSubscribedLayerChanged(int32)

func (*Publication) OnRttUpdate 

func (publication *Publication) OnRttUpdate(rtt uint32)

func (*Publication) OnStatsUpdate 

func (publication *Publication) OnStatsUpdate(stats *livekit.AnalyticsStat)

func (*Publication) OnStreamStarted 

func (*Publication) OnStreamStarted(time.Duration)

func (*Publication) ReplaceSource 

func (publication *Publication) ReplaceSource(next *Source) error

ReplaceSource keeps every RID, allocator, counter and sender on this PC.

func (*Publication) RequiredActiveCount 

func (publication *Publication) RequiredActiveCount() int

func (*Publication) SetActiveCount 

func (publication *Publication) SetActiveCount(count int) error

func (*Publication) SetAudioBitrate 

func (publication *Publication) SetAudioBitrate(bitrate uint32)

func (*Publication) SetConnected 

func (publication *Publication) SetConnected() error

func (*Publication) VideoBudget 

func (publication *Publication) VideoBudget() int64

type PublicationCounters 

type PublicationCounters struct {
	Frames                 []uint64
	VideoBytes, AudioBytes uint64
	TargetBitrate          int64
	Observed, Limited      bool
}

type Snapshot 

type Snapshot struct {
	Prepare, Target, Requested, Current int32
	Paused                              bool
	// VideoBudget is the current connection estimate after audio reservation.
	// It is codec input, not a claim that this bitrate is available or delivered.
	VideoBudget int64
}

Snapshot is read by the parent at its input-frame boundary. Current means accepted for forwarding, not remotely decoded. Prepare admits codec work, not network delivery, from the latest supplied budget (initial or observed). Its caller gates observed-only decisions with Transport.TargetBitrate. -1 means no selected layer.

type Source 

type Source struct {
	*sfu.ReceiverBase
	// contains filtered or unexported fields
}

Source owns finite RTP layers and their mature forwarding buffers. The caller owns input generations, codec clocks and PeerConnections. Feedback callbacks run without Source's mutex and may retire the source.

func NewSource 

func NewSource(options SourceOptions) (*Source, error)

func (*Source) BindLayer 

func (source *Source) BindLayer(layer int, ssrc uint32, parameters webrtc.RTPParameters) error

BindLayer installs a single source identity. A different SSRC requires the caller's source-generation replacement, not silent buffer reuse.

func (*Source) Close 

func (source *Source) Close()

func (*Source) RateControlled 

func (source *Source) RateControlled(layer int32) bool

func (*Source) ReferenceTime 

func (source *Source) ReferenceTime(layer int, timestamp uint32) (time.Time, bool)

ReferenceTime keeps derived outputs on the received source's RTP/NTP clock. The bounded input history makes the nearest 32-bit timestamp delta unambiguous.

func (*Source) SenderReport 

func (source *Source) SenderReport(layer int, report *rtcp.SenderReport) error

func (*Source) SetCorrelation 

func (source *Source) SetCorrelation(layer int, correlation *livekit.RTCPSenderReportState) error

SetCorrelation receives the local source owner's RTP/NTP mapping unchanged. The layer must already have received RTP; the media core owns report ordering.

func (*Source) SetLowestLayerRateControlled 

func (source *Source) SetLowestLayerRateControlled(controlled bool)

The live codec owner sets this after startup and clears it at retirement.

func (*Source) TrackInfo 

func (source *Source) TrackInfo() *livekit.TrackInfo

func (*Source) UpdateDimensions 

func (source *Source) UpdateDimensions(layer int, width, height uint32) error

UpdateDimensions publishes actual encoder dimensions without UpdateTrackInfo's mute propagation, which resets every active layer's tracker and bitrate.

func (*Source) UpdateFormats 

func (source *Source) UpdateFormats(formats []LayerFormat) error

UpdateFormats changes one source owner's declared layer metadata without replacing its buffers or packet identities. Zero values remain unknown. The caller serializes updates with its capture/source generation changes.

func (*Source) WriteRTP 

func (source *Source) WriteRTP(layer int, packet *rtp.Packet) error

type SourceOptions 

type SourceOptions struct {
	ID, StreamID     string
	Codec            webrtc.RTPCodecParameters
	Formats          []LayerFormat
	HeaderExtensions []webrtc.RTPHeaderExtensionParameter
	MaxPackets       int
	OnRTCP           func(layer int, packets []rtcp.Packet)
}

type Transport 

type Transport struct {
	PC          *webrtc.PeerConnection
	Output      *Output
	VideoSender *webrtc.RTPSender
	AudioSender *webrtc.RTPSender
	// contains filtered or unexported fields
}

Transport is outgoing-only. Its caller owns signaling, ICE events and the shared UDP mux. Video RTCP is callback-driven; never read VideoSender's RTCP.

func NewTransport 

func NewTransport(options TransportOptions) (_ *Transport, err error)

func (*Transport) BWEType 

func (*Transport) BWEType() bwe.BWEType

func (*Transport) Close 

func (transport *Transport) Close() error

func (*Transport) CurrentSource 

func (transport *Transport) CurrentSource() *Source

func (*Transport) Egress 

func (transport *Transport) Egress() EgressStats

Egress counts successful RTP writes, including headers but excluding SRTP/UDP overhead. It is local delivery to Pion, not proof of remote receipt or decode.

func (*Transport) IsBWEEnabled 

func (*Transport) IsBWEEnabled(*sfu.DownTrack) bool

func (*Transport) IsSubscribeMutable 

func (*Transport) IsSubscribeMutable(*sfu.DownTrack) bool

func (*Transport) OnAvailableLayersChanged 

func (*Transport) OnAvailableLayersChanged(*sfu.DownTrack)

func (*Transport) OnBindAndConnected 

func (*Transport) OnBindAndConnected()

func (*Transport) OnBitrateAvailabilityChanged 

func (*Transport) OnBitrateAvailabilityChanged(*sfu.DownTrack)

func (*Transport) OnCodecNegotiated 

func (*Transport) OnCodecNegotiated(webrtc.RTPCodecCapability)

func (*Transport) OnCongestionStateChange 

func (transport *Transport) OnCongestionStateChange(_, _ bwe.CongestionState, capacity int64)

HandleTWCCFeedback invokes this synchronously under the transport's event lock. Store the framework result without reentering that lock.

func (*Transport) OnDownTrackClose 

func (*Transport) OnDownTrackClose(bool)

func (*Transport) OnMaxPublishedSpatialChanged 

func (*Transport) OnMaxPublishedSpatialChanged(*sfu.DownTrack)

func (*Transport) OnMaxPublishedTemporalChanged 

func (*Transport) OnMaxPublishedTemporalChanged(*sfu.DownTrack)

func (*Transport) OnMaxSubscribedLayerChanged 

func (*Transport) OnMaxSubscribedLayerChanged(int32)

func (*Transport) OnPacerProbeObserverClusterComplete 

func (transport *Transport) OnPacerProbeObserverClusterComplete(id ccutils.ProbeClusterId)

func (*Transport) OnProbeClusterSwitch 

func (*Transport) OnProbeClusterSwitch(ccutils.ProbeClusterInfo)

AddCluster returns the exact cluster synchronously. Initializing it there avoids reentering the Prober while its own cluster-start mutex is held.

func (*Transport) OnREMB 

func (*Transport) OnREMB(*sfu.DownTrack, *rtcp.ReceiverEstimatedMaximumBitrate)

func (*Transport) OnResume 

func (*Transport) OnResume(*sfu.DownTrack)

func (*Transport) OnRttUpdate 

func (transport *Transport) OnRttUpdate(rtt uint32)

func (*Transport) OnSendProbe 

func (transport *Transport) OnSendProbe(bytesToSend int)

func (*Transport) OnStatsUpdate 

func (transport *Transport) OnStatsUpdate(stats *livekit.AnalyticsStat)

func (*Transport) OnStreamStarted 

func (*Transport) OnStreamStarted(time.Duration)

func (*Transport) OnSubscribedLayerChanged 

func (*Transport) OnSubscribedLayerChanged(*sfu.DownTrack, buffer.VideoLayer)

func (*Transport) OnSubscriptionChanged 

func (transport *Transport) OnSubscriptionChanged(*sfu.DownTrack)

func (*Transport) OnTransportCCFeedback 

func (transport *Transport) OnTransportCCFeedback(_ *sfu.DownTrack, packet *rtcp.TransportLayerCC)

LiveKit delivers feedback without a ReadRTCP loop on its callback reader.

func (*Transport) ReplaceSource 

func (transport *Transport) ReplaceSource(next *Source) error

ReplaceSource retains the connection and its congestion state. The caller prepares recovery in next and serializes group lifetime with this handoff.

func (*Transport) RequiredActiveCount 

func (transport *Transport) RequiredActiveCount() int

func (*Transport) SetAudioBitrate 

func (transport *Transport) SetAudioBitrate(bitrate uint32)

SetAudioBitrate reserves the current audio owner's contribution to this PC's shared estimate. Zero means no audio demand; the transport invents no profile.

func (*Transport) SetConnected 

func (transport *Transport) SetConnected() error

func (*Transport) TargetBitrate 

func (transport *Transport) TargetBitrate() (int, bool)

type TransportOptions 

type TransportOptions struct {
	Source         *Source
	Settings       webrtc.SettingEngine
	ConnectionID   string
	InitialBitrate int
	Audio          webrtc.TrackLocal
	// OnDemandChanged must only signal its owner without blocking or reentering
	// Transport. Callers can coalesce signals in their existing bounded owner.
	OnDemandChanged func()
}

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.
⌂ 换网址