webrtc

package
v0.11.0 Latest Latest
Warning

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

 Go to latest Published: Oct 17, 2026 License: MIT Imports: 18 Imported by: 0

Documentation

Overview

Package webrtc provides WebRTC abstractions for the Slive project. It wraps the Pion WebRTC library to provide thread-safe, domain-specific interfaces for managing peer connections, media tracks, session descriptions, and ICE candidates.

The package is designed to be used by the core engine to enable real-time audio/video communication between participants in a room.

Key abstractions:

  • PeerConnection: Manages WebRTC peer connections with thread-safe operations.
  • WebRTCTrack: Extends the domain Track model with WebRTC-specific functionality.
  • SessionDescription: Handles SDP offers/answers for signaling.
  • ICECandidate: Manages ICE candidate exchange for NAT traversal.

All abstractions are thread-safe and can be safely used concurrently.

Connection lifecycle

A PeerConnection starts in state New when created via NewPeerConnection and moves through Connecting/Connected as pion reports connection-state changes via OnConnectionStateChange. Disconnected marks a recoverable drop (NeedsReconnect reports true for Disconnected and Failed); Closed is terminal. Close is idempotent and safe to call from any goroutine; it stops internal RTCP pumps and closes the underlying pion peer connection.

Operations attempted on a Closed or Failed connection return ErrPeerConnectionClosed instead of panicking or blocking.

PeerConnectionState.Usable reports whether a state can still carry media (everything except Closed and Failed; Disconnected is usable because ICE may self-heal). The signaling layer reuses usable connections across reconnects and replaces unusable ones.

Structured logging

Peer connections emit structured slog events for lifecycle transitions (connection_state), failed automatic offer pushes (offer_create_failed), signaling send failures (signaling_send_failed) and ICE retry exhaustion (ice_retry_failed), each tagged with a participant_id where known. Choose the destination per connection via PeerConnectionConfig.Logger; when nil, slog.Default() is used. The signaling handler accepts its own logger via WithLogger and propagates it to every connection it creates, so one logger configured at startup covers the whole server.

Retry policy (ICE yes, SDP no)

AddICECandidateWithRetry retries transient ICE candidate failures a bounded number of times (defaultICERetryAttempts with linear defaultICERetryDelay backoff) and reports exhaustion as ErrICEFailed wrapping the last error. Both knobs are package vars so tests can shrink them.

SDP operations are deliberately never retried. CreateOffer, CreateAnswer and SetRemoteDescription mutate the negotiation state machine: a blind retry after a wrong-state failure produces another wrong-state failure at best and a duplicate offer or desynchronized description at worst. Those failures are surfaced immediately to the signaling layer, which maps them to negotiation_failed / peer_connection_closed for the remote client.

Negotiation flow (SFU-style: the server answers on behalf of its peers)

Slive's server owns one PeerConnection per participant. Signaling messages carry a target_participant_id and are applied to that participant's local PeerConnection by the signaling layer:

  • An offer from client A targeted at B is applied to B's peer connection with CreateAnswer. CreateAnswer first installs the remote offer (SetRemoteDescription), then generates the answer, waits for ICE gathering to complete so the SDP is complete, and returns it; the signaling layer relays it back to A.
  • An answer from client A targeted at B is installed on B's peer connection via SetRemoteDescription.
  • ICE candidates from any client are added to the target peer connection with AddICECandidateWithRetry, which retries transient failures a bounded number of times before giving up.

Outbound events flow the opposite way: whenever pion fires negotiation-needed, or produces a local ICE candidate, the peer connection pushes a "webrtc:offer" / "webrtc:ice-candidate" message through the configured SignalingSender. The sender can be swapped at any time with UpdateSignalingSender — the signaling handler does this on every reconnect so events follow the newest WebSocket of a participant.

Note that wrong-state SetRemoteDescription failures are protocol misuse by the remote side, not transient errors: they are surfaced directly and are deliberately not retried.

Event callbacks

OnNegotiationNeeded, OnICECandidate and OnTrack register application callbacks that are stored on the peer connection and invoked from the pion event handlers:

  • OnNegotiationNeeded callbacks run asynchronously so user code never executes on pion's dispatch goroutine.
  • OnICECandidate receives every gathered local candidate wrapped in an ICECandidate (the trailing nil gathering-complete sentinel is not forwarded).
  • OnTrack receives the WebRTCTrack wrapper registered in the remote-track registry; callbacks run outside the internal lock and may re-enter PeerConnection methods.

Registering a negotiation-needed callback does not disable the automatic offer push described above; both happen when a sender is configured.

Configuration injection points

PeerConnectionConfig carries everything NewPeerConnection needs: ICE servers and SDP semantics. DefaultPeerConnectionConfig returns the production default (public STUN, unified plan with fallback). Applications build their own configuration without importing application-level config packages by using ICEServersFromURLs together with ICETurnServer, which map plain STUN URL lists and TURN credentials onto pion's ICEServer type. The signaling handler accepts a PeerConnectionConfig via WithPeerConnectionConfig and uses it for every peer connection it creates.

For deterministic offline tests use a STUN-free PeerConnectionConfig (only SDPSemantics set) plus an AddTransceiverFromKind(recvonly audio) so offers contain a media section with ICE credentials.

Example usage: 

// Create a new peer connection
config := webrtc.DefaultPeerConnectionConfig()
pc, err := webrtc.NewPeerConnection(config, participant, nil)
if err != nil {
    // handle error
}
defer pc.Close()

// Add a local track
track := webrtc.NewWebRTCTrack(domainTrack, pionTrack, codec)
if err := pc.AddTrack(track); err != nil {
    // handle error
}

// Create an offer
offer, err := pc.CreateOffer()
if err != nil {
    // handle error
}

// Send offer via signaling channel
// ...

// Set remote answer
if err := pc.SetRemoteDescription(answer); err != nil {
    // handle error
}

Package webrtc implements SFU forwarding and peer connection management.

Lock hierarchy (must be respected to avoid deadlocks): 

PeerConnection.mu > TrackForwarder.mu > WebRTCTrack.mu
Handler.trackForwardersMutex > TrackForwarder.lifecycleMu > TrackForwarder.mu

TrackForwarder must never call Handler while holding mu, and callbacks registered via OnLocalTrackAdded / OnTrack must never acquire TrackForwarder.mu while holding PeerConnection.mu.

Index

Constants

View Source 
const DefaultQueueSize = 64

DefaultQueueSize is the default per-subscriber RTP queue capacity.

Variables

View Source 
var (
	// ErrTrackNotReady is returned when a track is not ready for operations.
	ErrTrackNotReady = errors.New("track not ready")

	// ErrTrackNotFound is returned when a track with the given ID is not found.
	ErrTrackNotFound = errors.New("track not found")

	// ErrPeerConnectionClosed is returned when operations are attempted on a closed peer connection.
	ErrPeerConnectionClosed = errors.New("peer connection closed")

	// ErrInvalidSDP is returned when an SDP string is invalid.
	ErrInvalidSDP = errors.New("invalid SDP")

	// ErrInvalidICECandidate is returned when an ICE candidate string is invalid.
	ErrInvalidICECandidate = errors.New("invalid ICE candidate")

	// ErrNoPeerConnection is returned when no peer connection is available.
	ErrNoPeerConnection = errors.New("no peer connection")

	// ErrSignalingError is returned when there is an error during signaling.
	ErrSignalingError = errors.New("signaling error")

	// ErrNegotiationFailed is returned when SDP negotiation fails.
	ErrNegotiationFailed = errors.New("negotiation failed")

	// ErrICEFailed is returned when adding a remote ICE candidate kept
	// failing until retries were exhausted; it wraps the last underlying
	// error so callers can inspect the root cause with errors.Is.
	ErrICEFailed = errors.New("ice candidate failed")

	// ErrNoPendingOffer is returned when a browser answer arrives but the
	// PC holds no outstanding server offer (not in have-local-offer).
	// Callers treat duplicate retransmits as idempotent and anything else
	// as a mis-targeted answer. It never wraps a pion
	// InvalidModificationError: the pion state is pre-checked, so reaching
	// pion with an incompatible state is impossible by construction.
	ErrNoPendingOffer = errors.New("answer with no outstanding server offer")

	// ErrUnexpectedOffer is returned when a browser offer arrives while a
	// browser negotiation is already in progress on the PC (e.g. a second
	// outstanding publisher offer). Each negotiation cycle has exactly one
	// offerer; a concurrent second browser offer is a client protocol
	// violation, reported without touching pion state.
	ErrUnexpectedOffer = errors.New("unexpected browser offer while negotiation in progress")
)

Common errors for the webrtc package.

View Source 
var ConnectionMetrics = &connectionMetrics{}

ConnectionMetrics is the package-level metrics registry for peer connections.

Functions

func ICEServersFromURLs 

func ICEServersFromURLs(stunURLs []string, turnServers []ICETurnServer) []webrtc.ICEServer

ICEServersFromURLs builds the pion ICE server list from STUN URLs and TURN servers. It lets callers translate application configuration into PeerConnectionConfig without this package depending on internal/config.

All STUN URLs are grouped into a single credential-free ICEServer entry; every TURN server becomes its own entry carrying its credentials. Empty inputs are skipped, so callers may pass optional configuration verbatim.

func SDPSummary added in v0.9.0 

func SDPSummary(sdp string) string

SDPSummary compresses an SDP into one grep-able line per m-section: "m=<kind> <PTs> | mid:<mid> | <direction> | rtpmap:<PT> <codec> ...". DIAG-E2E: lets a live two-browser test compare the SFU subscriber offer against the browser answer without dumping full SDP.

func StartTime 

func StartTime() time.Time

StartTime returns the process start time used for UptimeSeconds.

Types

type ForwarderConfig 

type ForwarderConfig struct {
	QueueSize int
}

ForwarderConfig holds tunable parameters for TrackForwarder.

type ICECandidate 

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

ICECandidate represents a WebRTC ICE candidate. It provides thread-safe access to the underlying Pion WebRTC ICE candidate.

func NewICECandidate 

func NewICECandidate(candidate *webrtc.ICECandidate) *ICECandidate

NewICECandidate creates a new ICECandidate from a Pion WebRTC ICE candidate.

func NewICECandidateFromString 

func NewICECandidateFromString(candidate string) (*ICECandidate, error)

NewICECandidateFromString creates a new ICECandidate from a candidate string.

func (*ICECandidate) Candidate 

func (c *ICECandidate) Candidate() string

Candidate returns the candidate string.

func (*ICECandidate) MarshalJSON 

func (c *ICECandidate) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler for ICECandidate.

func (*ICECandidate) PionICECandidateInit 

func (c *ICECandidate) PionICECandidateInit() webrtc.ICECandidateInit

PionICECandidateInit returns the underlying Pion WebRTC ICE candidate init.

func (*ICECandidate) SDPMLineIndex 

func (c *ICECandidate) SDPMLineIndex() uint16

SDPMLineIndex returns the SDP m-line index.

func (*ICECandidate) SDPMid 

func (c *ICECandidate) SDPMid() string

SDPMid returns the SDP mid attribute.

func (*ICECandidate) SetCandidate 

func (c *ICECandidate) SetCandidate(candidate string) error

SetCandidate updates the candidate string.

func (*ICECandidate) SetSDPMLineIndex 

func (c *ICECandidate) SetSDPMLineIndex(index uint16)

SetSDPMLineIndex updates the SDP m-line index.

func (*ICECandidate) SetSDPMid 

func (c *ICECandidate) SetSDPMid(sdpMid string)

SetSDPMid updates the SDP mid attribute.

func (*ICECandidate) String 

func (c *ICECandidate) String() string

String returns a string representation of the ICECandidate.

func (*ICECandidate) UnmarshalJSON 

func (c *ICECandidate) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler for ICECandidate.

type ICETurnServer 

type ICETurnServer struct {
	// URLs of the TURN server, e.g. "turn:turn.example.com:3478".
	URLs []string
	// Username used to authenticate against the TURN server.
	Username string
	// Credential used to authenticate against the TURN server.
	Credential string
}

ICETurnServer describes a TURN relay server used during ICE gathering.

type MetricsSnapshot 

type MetricsSnapshot struct {
	ConnectionAttemptsTotal uint64 `json:"connection_attempts_total"`
	ConnectionFailuresTotal uint64 `json:"connection_failures_total"`
	ForwarderSubscribers    int    `json:"forwarder_subscribers"`
	ForwarderDroppedTotal   uint64 `json:"forwarder_dropped_total"`
	ForwarderQueueDepth     int    `json:"forwarder_queue_depth"`
	RoomsActive             int    `json:"rooms_active"`
	ParticipantsActive      int    `json:"participants_active"`
	TracksPublished         int    `json:"tracks_published"`
	GCReapedTotal           uint64 `json:"gc_reaped_total"`
	UptimeSeconds           int64  `json:"uptime_seconds"`
	Goroutines              int    `json:"goroutines"`
}

MetricsSnapshot is a point-in-time copy of all observable counters and gauges. It is safe to encode and write without holding any handler or forwarder locks.

type NegotiationState added in v0.11.0 

type NegotiationState int

NegotiationState is the observable state of the unified-plan negotiation controller. Pion's SignalingState remains authoritative for legal SDP operations; this state names the controller's intent and deferred work. 

const (
	NegotiationStateStable NegotiationState = iota
	NegotiationStateBrowserOfferInProgress
	NegotiationStateServerOfferInProgress
	NegotiationStateWaitingForBrowserAnswer
	NegotiationStateAnsweringBrowserOffer
	NegotiationStatePendingServerNegotiation
	NegotiationStateDeferredBrowserOffer
)

func (NegotiationState) String added in v0.11.0 

func (s NegotiationState) String() string

type PeerConnection 

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

PeerConnection manages a WebRTC peer connection. It provides thread-safe access to the underlying Pion WebRTC peer connection.

func NewPeerConnection 

func NewPeerConnection(config PeerConnectionConfig, participant *domain.Participant, sendSignaling SignalingSender) (*PeerConnection, error)

NewPeerConnection creates a new PeerConnection with the given configuration.

func (*PeerConnection) AddICECandidate 

func (pc *PeerConnection) AddICECandidate(candidate *ICECandidate) error

AddICECandidate adds a remote ICE candidate.

func (*PeerConnection) AddICECandidateWithRetry 

func (pc *PeerConnection) AddICECandidateWithRetry(candidate *ICECandidate) error

AddICECandidateWithRetry adds a remote ICE candidate, retrying transient failures a bounded number of times before giving up.

LiveKit-style trickle ICE: candidates received before the remote description is set are queued (logged as "queueing ICE candidate") and flushed after SetRemoteDescription moves to stable. This prevents the "ICE candidate before remote description" race from failing.

func (*PeerConnection) AddTrack 

func (pc *PeerConnection) AddTrack(track *WebRTCTrack) error

AddTrack adds a local track to the peer connection.

Lock hierarchy: PeerConnection.mu is held only to update localTracks and snapshot the OnLocalTrackAdded callback; the callback is invoked outside mu via a goroutine with panic recovery so user code never runs on pion's dispatch path and does not invert PeerConnection.mu > TrackForwarder.mu. Callbacks must not acquire TrackForwarder.mu while holding pc.mu.

func (*PeerConnection) Close 

func (pc *PeerConnection) Close() error

Close closes the peer connection.

func (*PeerConnection) Config 

func (pc *PeerConnection) Config() PeerConnectionConfig

Config returns the configuration of the peer connection.

func (*PeerConnection) CreateAnswer 

func (pc *PeerConnection) CreateAnswer(offer *SessionDescription) (*SessionDescription, error)

CreateAnswer creates an SDP answer for the given offer. LiveKit-style: after setting remote offer, flush any queued ICE candidates.

func (*PeerConnection) CreateOffer 

func (pc *PeerConnection) CreateOffer() (*SessionDescription, error)

CreateOffer creates an SDP offer.

func (*PeerConnection) EgressSSRCMap added in v0.9.0 

func (pc *PeerConnection) EgressSSRCMap() map[uint32]string

EgressSSRCMap returns the subscriber-egress media SSRCs currently bound on this PC, mapped to their track IDs. Pion assigns each RTPSender a fixed SSRC at AddTrack time and rewrites every forwarded packet to the per-binding SSRC/payload-type on WriteRTP, so a subscriber PLI's MediaSSRC (egress SSRC) differs from the publisher-ingress SSRC. The keyframe relay uses this map to resolve an incoming PLI/FIR MediaSSRC to the subscribed track without trusting the reader-closure track ID alone.

func (*PeerConnection) FlushDeferredNegotiation added in v0.9.0 

func (pc *PeerConnection) FlushDeferredNegotiation()

FlushDeferredNegotiation drains coalesced negotiation work (a stored inbound browser offer, then one subscriber offer) once the caller has finished delivering the message that preceded the flush. Callers that send an answer produced by ProcessBrowserOffer MUST call this only after that send is enqueued: conn.Send and the signaling sender share one FIFO send channel, so flushing first would let a subscriber offer overtake the answer to the browser's own publisher offer (Chrome answers the early offer, implicitly rolls back its publish exchange, and the real answer lands in stable and is lost).

func (*PeerConnection) GetLocalTrack 

func (pc *PeerConnection) GetLocalTrack(trackID string) *WebRTCTrack

GetLocalTrack retrieves a local track by ID.

func (*PeerConnection) GetRemoteTrack 

func (pc *PeerConnection) GetRemoteTrack(trackID string) *WebRTCTrack

GetRemoteTrack retrieves a remote track by ID.

func (*PeerConnection) ICEState added in v0.9.0 

func (pc *PeerConnection) ICEState() string

ICEState returns the last observed ICE connection state.

func (*PeerConnection) InstanceID added in v0.9.0 

func (pc *PeerConnection) InstanceID() string

InstanceID returns the stable per-object ID (pc-<n>) of this PeerConnection. Immutable after construction; safe without locking.

func (*PeerConnection) NeedsReconnect 

func (pc *PeerConnection) NeedsReconnect() bool

NeedsReconnect returns true when the peer connection is in a recoverable disconnected state.

func (*PeerConnection) NegotiationState added in v0.11.0 

func (pc *PeerConnection) NegotiationState() NegotiationState

NegotiationState reports the explicit controller state for diagnostics and tests. It is derived from the gate flags plus Pion's authoritative SignalingState, so callers cannot observe a synthetic state that would make an illegal SetRemoteDescription or SetLocalDescription appear valid.

func (*PeerConnection) OnICECandidate 

func (pc *PeerConnection) OnICECandidate(callback func(*ICECandidate))

OnICECandidate registers a callback invoked for every local ICE candidate produced during gathering. The trailing nil sentinel emitted at the end of gathering by the WebRTC spec is not forwarded. Passing nil clears a previously registered callback.

func (*PeerConnection) OnLocalTrackAdded 

func (pc *PeerConnection) OnLocalTrackAdded(callback func(*WebRTCTrack))

OnLocalTrackAdded registers a callback invoked when a local track is added via AddTrack. The callback receives the WebRTCTrack that was just added. Passing nil clears a previously registered callback. Callbacks are invoked outside PeerConnection.mu via a goroutine with panic recovery; they must not acquire TrackForwarder.mu while holding pc.mu (see lock hierarchy).

func (*PeerConnection) OnNegotiationNeeded 

func (pc *PeerConnection) OnNegotiationNeeded(callback func())

OnNegotiationNeeded registers a callback invoked whenever the underlying peer connection signals that renegotiation is needed. Registering a callback does not disable the automatic offer push performed when a signaling sender is configured (see handleNegotiationNeeded). Passing nil clears a previously registered callback.

func (*PeerConnection) OnTrack 

func (pc *PeerConnection) OnTrack(callback func(*WebRTCTrack))

OnTrack registers a callback invoked when the remote peer starts streaming a media track towards us. The wrapper WebRTCTrack passed to the callback is the same instance stored in the remote-track registry. Passing nil clears a previously registered callback.

func (*PeerConnection) Participant 

func (pc *PeerConnection) Participant() *domain.Participant

Participant returns the participant associated with this peer connection.

func (*PeerConnection) PionPeerConnection 

func (pc *PeerConnection) PionPeerConnection() *webrtc.PeerConnection

PionPeerConnection returns the underlying Pion WebRTC peer connection.

func (*PeerConnection) ProcessBrowserAnswer added in v0.9.0 

func (pc *PeerConnection) ProcessBrowserAnswer(answer *SessionDescription) error

ProcessBrowserAnswer installs a browser answer to our outstanding subscriber offer (Flow B completion). The PC must hold our offer (have-local-offer); otherwise ErrNoPendingOffer is returned and the caller applies duplicate/mis-targeted handling. On success the PC is stable and deferred work flushes in priority order: a stored inbound browser offer first, then one coalesced server offer.

func (*PeerConnection) ProcessBrowserOffer added in v0.9.0 

func (pc *PeerConnection) ProcessBrowserOffer(offer *SessionDescription) (*SessionDescription, bool, error)

ProcessBrowserOffer answers a browser publisher offer (Flow A: browser is the offerer, Slive the answerer) through the negotiation gate.

  • stable: the offer is answered inline and the answer is returned. The caller must deliver the answer and then call FlushDeferredNegotiation, which flushes any coalesced server negotiation as exactly one subscriber offer after the answer is enqueued. Returns (answer, false, nil).
  • have-local-offer (our subscriber offer outstanding): the inbound offer is stored, never dropped, and answered with priority once the PC is stable again. Returns (nil, true, nil); the caller must not report an error — the answer is delivered later via the signaling sender. This replaces the old InvalidModificationError failure. A second inbound offer while one is stored does not error and does not open a second lifecycle: the browser's newest offer supersedes the stored one (its older transaction is dead — a browser has one signaling state machine), so the slot is replaced with a loud supersession log.
  • any other state: returns ErrUnexpectedOffer without touching pion state (unreachable through the gate; guards legacy callers).

func (*PeerConnection) RemoveTrack 

func (pc *PeerConnection) RemoveTrack(trackID string) error

RemoveTrack removes a local track from the peer connection.

func (*PeerConnection) RequestSubscriberOffer added in v0.9.0 

func (pc *PeerConnection) RequestSubscriberOffer() bool

RequestSubscriberOffer drives one server subscriber offer synchronously through the gate (used by handleSubscribeTrack right after AddTrack, so a first subscription never depends solely on pion's asynchronous negotiation-needed callback). If the gate is closed the need stays coalesced in pendingNegotiation for the async callback and completion flushes; returns true when an offer was sent.

func (*PeerConnection) SetRTCPObserver added in v0.9.0 

func (pc *PeerConnection) SetRTCPObserver(fn RTCPObserver)

SetRTCPObserver registers a hook invoked for every RTCP packet read from this PC's senders (see RTCPObserver). Passing nil clears it. Additive instrumentation only; the RTCP discard behavior is unchanged.

func (*PeerConnection) SetRemoteDescription 

func (pc *PeerConnection) SetRemoteDescription(sdp *SessionDescription) error

SetRemoteDescription sets the remote session description. On success, if a negotiation was queued while an offer was pending, it flushes the coalesced request after moving to stable. LiveKit-style ICE queuing: any candidates queued before remote description are flushed here.

func (*PeerConnection) State 

func (pc *PeerConnection) State() PeerConnectionState

State returns the current state of the peer connection.

func (*PeerConnection) SwapEgressTrack added in v0.9.0 

func (pc *PeerConnection) SwapEgressTrack(trackID string, next *WebRTCTrack) (removedErr error, err error)

SwapEgressTrack replaces one egress sender (codec reconcile: remove the stale TrackLocal, attach the fresh one) under the negotiation gate. The churn must never interleave with an in-flight exchange: a sender attached after CreateOffer but before SetRemoteDescription(answer) is started by pion against the stale negotiated m-line, and if the browser answered that m-line without the new codec the whole answer fails with ErrUnsupportedCodec (observed live: a reconciled H264 egress bound against a VP8-era answer, deadlocking the subscriber's video). Holding negMu means the swap happens either fully before CreateOffer or fully after the answer is applied; the generation bump from AddTrack then drives exactly one fresh subscriber offer that carries the new codec. Caller must NOT hold f.mu-style forwarder locks that conflict with PC callbacks (same contract as RemoveTrack/AddTrack).

func (*PeerConnection) UpdateSignalingSender 

func (pc *PeerConnection) UpdateSignalingSender(sendSignaling SignalingSender)

UpdateSignalingSender updates the signaling sender function.

func (*PeerConnection) WriteRTCP added in v0.9.0 

func (pc *PeerConnection) WriteRTCP(pkts []rtcp.Packet) error

WriteRTCP sends RTCP packets (e.g. relayed PLI/FIR) to this PC's remote. It is the upstream half of the keyframe-feedback relay: the signaling layer calls it on the PUBLISHER's PC with a PLI/FIR addressed to the publisher-ingress media SSRC. Packets are passed to pion unchanged; no RTP state is touched.

type PeerConnectionConfig 

type PeerConnectionConfig struct {
	// ICEServers is a list of ICE servers to use for NAT traversal.
	ICEServers []webrtc.ICEServer
	// SDPSemantics is the SDP semantics to use (e.g., "unified-plan").
	SDPSemantics webrtc.SDPSemantics
	// Logger receives structured lifecycle events for connections created
	// with this config. A nil Logger resolves to slog.Default() inside
	// NewPeerConnection, so the zero value stays usable.
	Logger *slog.Logger
}

PeerConnectionConfig holds configuration for creating a new PeerConnection.

func DefaultPeerConnectionConfig 

func DefaultPeerConnectionConfig() PeerConnectionConfig

DefaultPeerConnectionConfig returns a default PeerConnectionConfig.

type PeerConnectionState 

type PeerConnectionState int

PeerConnectionState represents the current state of a peer connection. 

const (
	PeerConnectionStateNew PeerConnectionState = iota
	PeerConnectionStateConnecting
	PeerConnectionStateConnected
	PeerConnectionStateDisconnected
	PeerConnectionStateFailed
	PeerConnectionStateClosed
)

func (PeerConnectionState) String 

func (s PeerConnectionState) String() string

func (PeerConnectionState) Usable 

func (s PeerConnectionState) Usable() bool

Usable reports whether a connection in this state can still carry media. Only the terminal Closed state and the unrecoverable Failed state are unusable; Disconnected is deliberately usable because ICE may self-heal (NeedsReconnect describes that recoverable-drop case instead). The signaling layer replaces unusable connections on reconnect and reuses usable ones.

type RTCPObserver added in v0.9.0 

type RTCPObserver func(trackID string, packet rtcp.Packet)

RTCPObserver is a hook invoked for every RTCP packet read from a subscriber sender (see AddTrack). The signaling layer uses it to relay subscriber PLI/FIR upstream to the publisher. It runs on the RTCP reader goroutine without holding PeerConnection.mu; implementations must not block and must not call back into PeerConnection methods that take mu while holding their own locks in conflicting order. A nil observer is a no-op. The packet is still discarded afterwards.

type SessionDescription 

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

SessionDescription represents a WebRTC session description (SDP offer/answer). It provides thread-safe access to the underlying Pion WebRTC session description.

func NewSessionDescription 

func NewSessionDescription(sdp *webrtc.SessionDescription) *SessionDescription

NewSessionDescription creates a new SessionDescription from a Pion WebRTC session description.

func NewSessionDescriptionFromString 

func NewSessionDescriptionFromString(sdp string, sdpType webrtc.SDPType) (*SessionDescription, error)

NewSessionDescriptionFromString creates a new SessionDescription from an SDP string and type.

func (*SessionDescription) MarshalJSON 

func (s *SessionDescription) MarshalJSON() ([]byte, error)

MarshalJSON implements json.Marshaler for SessionDescription.

func (*SessionDescription) PionSessionDescription 

func (s *SessionDescription) PionSessionDescription() *webrtc.SessionDescription

PionSessionDescription returns the underlying Pion WebRTC session description.

func (*SessionDescription) SDP 

func (s *SessionDescription) SDP() string

SDP returns the SDP string.

func (*SessionDescription) SetSDP 

func (s *SessionDescription) SetSDP(sdp string)

SetSDP updates the SDP string.

func (*SessionDescription) SetType 

func (s *SessionDescription) SetType(sdpType webrtc.SDPType)

SetType updates the SDP type.

func (*SessionDescription) String 

func (s *SessionDescription) String() string

String returns a string representation of the SessionDescription.

func (*SessionDescription) Type 

func (s *SessionDescription) Type() webrtc.SDPType

Type returns the SDP type (offer, answer, etc.).

func (*SessionDescription) UnmarshalJSON 

func (s *SessionDescription) UnmarshalJSON(data []byte) error

UnmarshalJSON implements json.Unmarshaler for SessionDescription.

type SignalingSender 

type SignalingSender func(msgType string, data interface{}) error

SignalingSender is a function type for sending signaling messages. It takes a message type and data, and sends it via the signaling connection.

type TrackForwarder 

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

TrackForwarder fans RTP packets from a publisher's WebRTCTrack out to multiple subscriber PeerConnections.

Each subscriber receives its own TrackLocalStaticRTP with the same codec as the publisher; Pion assigns a unique SSRC per PeerConnection binding during negotiation (Bind), so no manual SSRC management is required.

Thread-safe: all public methods may be called concurrently.

func NewTrackForwarder 

func NewTrackForwarder(publisher *WebRTCTrack) (*TrackForwarder, error)

NewTrackForwarder creates a forwarder for the given publisher track. publisher must be non-nil.

func NewTrackForwarderWithConfig 

func NewTrackForwarderWithConfig(publisher *WebRTCTrack, cfg ForwarderConfig) (*TrackForwarder, error)

NewTrackForwarderWithConfig creates a forwarder with explicit queue sizing.

func (*TrackForwarder) AddSubscriber 

func (f *TrackForwarder) AddSubscriber(pc *PeerConnection) error

AddSubscriber registers pc as a destination for forwarded RTP. It creates a TrackLocalStaticRTP that mirrors the publisher's codec, wraps it in a WebRTCTrack, stores it, and adds it to pc via AddTrack.

When the publisher is still a placeholder (codec unknown), the egress codec is inferred provisionally from kind; UpdatePublisher rebuilds the subscriber track once the real TrackRemote codec arrives if the MIME mismatches, so no RTP ever flows long-term through a wrong-codec binding.

Lock hierarchy: must call pc.AddTrack outside f.mu (f.mu > pc.mu would invert PeerConnection.mu > TrackForwarder.mu). Snapshot codec/trackID under RLock, release, call pc.AddTrack, then re-lock with CAS for idempotency (double-checked after pc call).

If pc is already subscribed the call is idempotent and returns nil. Returns an error if pc is nil or closed/failed.

func (*TrackForwarder) BytesForwarded added in v0.10.0 

func (f *TrackForwarder) BytesForwarded() uint64

func (*TrackForwarder) BytesReceived added in v0.9.0 

func (f *TrackForwarder) BytesReceived() uint64

func (*TrackForwarder) DiagSnapshot added in v0.9.0 

func (f *TrackForwarder) DiagSnapshot() (recv, fwd, written, dropped uint64, seq uint16, ts uint32, ssrc uint32, pt uint8)

DiagSnapshot returns a log-ready snapshot of forwarder media counters.

func (*TrackForwarder) DroppedCount 

func (f *TrackForwarder) DroppedCount(pc *PeerConnection) uint64

DroppedCount returns the number of packets dropped for the given subscriber due to queue full. Returns 0 if pc is not a subscriber.

func (*TrackForwarder) ForwardRTPPacket 

func (f *TrackForwarder) ForwardRTPPacket(pkt *rtp.Packet) error

ForwardRTPPacket is an alias for WriteRTP kept for readability in tests and for callers that already hold a parsed packet.

func (*TrackForwarder) HasSubscriber added in v0.9.0 

func (f *TrackForwarder) HasSubscriber(pc *PeerConnection) bool

HasSubscriber reports whether pc is already attached to this forwarder. It lets signaling distinguish an idempotent client subscribe from a new egress binding without exposing the subscriber map.

func (*TrackForwarder) IsRunning 

func (f *TrackForwarder) IsRunning() bool

IsRunning reports whether the forwarding goroutine is active.

func (*TrackForwarder) LastPayloadType added in v0.9.0 

func (f *TrackForwarder) LastPayloadType() uint8

func (*TrackForwarder) LastSSRC added in v0.9.0 

func (f *TrackForwarder) LastSSRC() uint32

func (*TrackForwarder) LastSeq added in v0.9.0 

func (f *TrackForwarder) LastSeq() uint16

func (*TrackForwarder) LastTimestamp added in v0.9.0 

func (f *TrackForwarder) LastTimestamp() uint32

func (*TrackForwarder) MaxQueueDepth 

func (f *TrackForwarder) MaxQueueDepth() int

MaxQueueDepth returns the maximum queue depth across all subscribers.

func (*TrackForwarder) NextFIRSequenceNumber added in v0.9.0 

func (f *TrackForwarder) NextFIRSequenceNumber() uint8

NextFIRSequenceNumber mints the next FIR entry sequence number for this track. Relayed FIRs share one per-track sequence space (starting at 1) across all subscribers so the publisher can dedupe requests.

func (*TrackForwarder) PacketsDropped added in v0.9.0 

func (f *TrackForwarder) PacketsDropped() uint64

func (*TrackForwarder) PacketsForwarded added in v0.9.0 

func (f *TrackForwarder) PacketsForwarded() uint64

func (*TrackForwarder) PacketsReceived added in v0.9.0 

func (f *TrackForwarder) PacketsReceived() uint64

Media counter accessors. Read-only snapshots.

func (*TrackForwarder) PacketsWritten added in v0.9.0 

func (f *TrackForwarder) PacketsWritten() uint64

func (*TrackForwarder) PublisherTrack 

func (f *TrackForwarder) PublisherTrack() *WebRTCTrack

PublisherTrack returns the publisher track this forwarder was created for.

func (*TrackForwarder) QueueDepth 

func (f *TrackForwarder) QueueDepth(pc *PeerConnection) int

QueueDepth returns the current queue depth for the given subscriber. Returns 0 if pc is not a subscriber.

func (*TrackForwarder) RemoveSubscriber 

func (f *TrackForwarder) RemoveSubscriber(pc *PeerConnection) error

RemoveSubscriber unregisters pc. It stops the per-subscriber writer, removes the forwarded track from the subscriber's PeerConnection when possible and deletes the entry. Returns ErrTrackNotFound if pc was not a subscriber. Drains queue best-effort before removing track.

Lock hierarchy: calls pc.RemoveTrack outside f.mu.

func (*TrackForwarder) ResetDropped 

func (f *TrackForwarder) ResetDropped()

ResetDropped clears dropped counters for all subscribers. Intended for tests.

func (*TrackForwarder) Start 

func (f *TrackForwarder) Start() error

Start launches the forwarding goroutine. It is idempotent: calling Start on an already-running forwarder returns nil. Returns an error if the publisher track is nil.

Dual modes: TrackLocal placeholder → no goroutine, running stays false, forwarding only via WriteRTP. TrackRemote → run pumps Read → WriteRTP.

func (*TrackForwarder) Stop 

func (f *TrackForwarder) Stop() error

Stop terminates the forwarding goroutine and all per-subscriber writers, then removes the forwarded track from every subscriber. It is idempotent and safe to call concurrently with other methods. For TrackLocal publishers where IsRunning is false, Stop still drains writers and removes tracks.

func (*TrackForwarder) SubscriberCount 

func (f *TrackForwarder) SubscriberCount() int

SubscriberCount returns the number of subscribers currently registered.

func (*TrackForwarder) TotalDropped 

func (f *TrackForwarder) TotalDropped() uint64

TotalDropped returns the sum of dropped packets across all subscribers.

func (*TrackForwarder) UpdatePublisher 

func (f *TrackForwarder) UpdatePublisher(publisher *WebRTCTrack) error

UpdatePublisher swaps the publisher track and restarts the forwarding goroutine if the forwarder is running. It is used to replace a placeholder TrackLocal publisher (created eagerly in handlePublishTrack, codec unknown) with the real TrackRemote that arrives later via PeerConnection.OnTrack. The forwarder keeps its existing subscribers across the swap.

Codec reconciliation: subscribers attached before the real codec arrived built their egress TrackLocal from a provisional inference. When the authoritative codec arrives (IsRemote), every subscriber whose egress MIME mismatches is rebuilt from the real codec (remove stale sender/TrackLocal, add fresh TrackLocal, existing negotiation-needed flow renegotiates). Pion cannot safely re-codec a bound TrackLocal, hence rebuild. The run loop restarts only after reconciliation so post-swap packets never sit in stale-codec queues.

Lock ordering: lifecycleMu > mu (Handler.trackForwardersMutex > lifecycleMu > mu). PC calls and writer shutdown happen outside f.mu; reconcileSubscriber takes only brief f.mu sections for map updates. Dual modes: TrackLocal → TrackRemote transitions running false→true and launches run; TrackRemote → TrackLocal stops run and sets running=false.

func (*TrackForwarder) Write 

func (f *TrackForwarder) Write(b []byte) (int, error)

Write forwards a raw RTP buffer to every subscriber by unmarshaling it into a packet and calling WriteRTP. See WriteRTP for backpressure semantics.

func (*TrackForwarder) WriteRTP 

func (f *TrackForwarder) WriteRTP(pkt *rtp.Packet) error

WriteRTP forwards a single RTP packet to every subscriber via bounded per-subscriber queues. It clones the packet per subscriber so Pion's per-binding SSRC/payload-type rewrite does not race. Slow subscribers never block fast ones: enqueue is non-blocking; on full queue the packet is dropped and DroppedCount increments.

Hardened: single RLock snapshot for entries+trackID (was double RLock), and queue-full check before clone to avoid alloc on dropped packets. Clone uses packetPool; dropped clones are returned to pool.

type WebRTCTrack 

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

WebRTCTrack wraps a domain.Track with WebRTC-specific functionality. It provides thread-safe access to the underlying WebRTC track and its metadata.

func NewWebRTCTrack 

func NewWebRTCTrack(domainTrack *domain.Track, pionTrack interface{}, codec webrtc.RTPCodecParameters) *WebRTCTrack

NewWebRTCTrack creates a new WebRTCTrack from a domain.Track and a Pion WebRTC track.

func (*WebRTCTrack) Close 

func (t *WebRTCTrack) Close() error

Close closes the underlying WebRTC track.

func (*WebRTCTrack) Codec 

func (t *WebRTCTrack) Codec() webrtc.RTPCodecParameters

Codec returns the codec parameters for this track.

func (*WebRTCTrack) DomainTrack 

func (t *WebRTCTrack) DomainTrack() *domain.Track

DomainTrack returns the underlying domain.Track.

func (*WebRTCTrack) ID 

func (t *WebRTCTrack) ID() string

ID returns the track's unique identifier.

func (*WebRTCTrack) IsRemote 

func (t *WebRTCTrack) IsRemote() bool

IsRemote reports whether the underlying Pion track is a TrackRemote (i.e. a real publisher track arriving via PeerConnection.OnTrack). A TrackLocalStaticRTP placeholder returns false.

func (*WebRTCTrack) Kind 

func (t *WebRTCTrack) Kind() domain.TrackKind

Kind returns the type of the track (audio or video).

func (*WebRTCTrack) PionTrack 

func (t *WebRTCTrack) PionTrack() interface{}

PionTrack returns the underlying Pion WebRTC track.

func (*WebRTCTrack) Publish 

func (t *WebRTCTrack) Publish()

Publish transitions the track to the published state.

func (*WebRTCTrack) Publisher 

func (t *WebRTCTrack) Publisher() *domain.Participant

Publisher returns the participant who published this track.

func (*WebRTCTrack) Read 

func (t *WebRTCTrack) Read(b []byte) (int, error)

Read reads RTP packets from the track.

The underlying reader reference is copied while holding the lock and the lock is released before any blocking I/O: TrackRemote.Read blocks until the next packet arrives, and holding t.mu across that window would deadlock a concurrent Close (which needs the write lock).

func (*WebRTCTrack) RemoteSSRC added in v0.9.0 

func (t *WebRTCTrack) RemoteSSRC() (uint32, bool)

RemoteSSRC returns the publisher-ingress media SSRC when the underlying track is a real TrackRemote. The keyframe-feedback relay addresses the upstream PLI/FIR to this SSRC. Placeholders (TrackLocal) report ok=false.

func (*WebRTCTrack) SetCodec 

func (t *WebRTCTrack) SetCodec(codec webrtc.RTPCodecParameters)

SetCodec updates the codec parameters for this track.

func (*WebRTCTrack) SetPublisher 

func (t *WebRTCTrack) SetPublisher(p *domain.Participant)

SetPublisher sets the participant who published this track.

func (*WebRTCTrack) Source 

func (t *WebRTCTrack) Source() domain.TrackSource

Source returns the source of the track.

func (*WebRTCTrack) State 

func (t *WebRTCTrack) State() domain.TrackState

State returns the current state of the track.

func (*WebRTCTrack) Unpublish 

func (t *WebRTCTrack) Unpublish()

Unpublish transitions the track to the unpublished state.

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