sfu

package
v0.1.0 Latest Latest
Warning

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

Go to latest
Published: Aug 23, 2026 License: MIT Imports: 10 Imported by: 0

Documentation

Overview

Package sfu implements a Selective Forwarding Unit (SFU) for WebRTC media streaming. It receives video/audio from publishers and forwards to multiple subscribers with minimal latency.

Index

Constants

View Source
const (
	ErrorCodeInvalidMessage  = "invalid_message"
	ErrorCodeRoomNotFound    = "room_not_found"
	ErrorCodeRoomFull        = "room_full"
	ErrorCodePublisherExists = "publisher_exists"
	ErrorCodePeerNotFound    = "peer_not_found"
	ErrorCodeInternalError   = "internal_error"
	ErrorCodeUnauthorized    = "unauthorized"
)

Error codes for signaling errors.

Variables

This section is empty.

Functions

This section is empty.

Types

type Config

type Config struct {
	// ICEServers are STUN/TURN servers for WebRTC.
	ICEServers []webrtc.ICEServer

	// EnableTWCC enables Transport-Wide Congestion Control.
	EnableTWCC bool

	// EnableRTCPReports enables RTCP reports for congestion control.
	EnableRTCPReports bool

	// PLIInterval is the interval for sending PLI requests (in seconds).
	// PLI (Picture Loss Indication) requests keyframes from the publisher.
	// Default: 3 seconds
	PLIInterval uint16

	// MaxBufferedAmount is the max bytes to buffer for data channels.
	// Default: 1MB
	MaxBufferedAmount uint64
}

Config contains SFU configuration options.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a Config with sensible defaults.

func (*Config) Validate

func (c *Config) Validate() error

Validate validates the configuration.

func (*Config) WebRTCConfig

func (c *Config) WebRTCConfig() webrtc.Configuration

WebRTCConfig returns a webrtc.Configuration from this Config.

type ICECandidateJSON

type ICECandidateJSON struct {
	// Candidate is the candidate string.
	Candidate string `json:"candidate"`

	// SDPMid is the media stream identification tag.
	SDPMid *string `json:"sdpMid,omitempty"`

	// SDPMLineIndex is the index of the media description.
	SDPMLineIndex *uint16 `json:"sdpMLineIndex,omitempty"`

	// UsernameFragment is the ICE username fragment.
	UsernameFragment *string `json:"usernameFragment,omitempty"`
}

ICECandidateJSON represents an ICE candidate in JSON format. This matches the WebRTC ICECandidateInit structure.

type InputChannel

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

InputChannel handles forwarding input events from subscribers to the publisher. Subscribers send keyboard/mouse events via data channels, which are forwarded to the publisher for processing.

func NewInputChannel

func NewInputChannel(room *Room, logger *zap.Logger) *InputChannel

NewInputChannel creates a new InputChannel for a room.

func (*InputChannel) AddSubscriberChannel

func (ic *InputChannel) AddSubscriberChannel(peerID string, dc *webrtc.DataChannel) error

AddSubscriberChannel adds a subscriber's data channel for sending input events.

func (*InputChannel) Close

func (ic *InputChannel) Close()

Close closes all data channels.

func (*InputChannel) HasPublisherChannel

func (ic *InputChannel) HasPublisherChannel() bool

HasPublisherChannel returns whether a publisher channel is set.

func (*InputChannel) IsClosed

func (ic *InputChannel) IsClosed() bool

IsClosed returns whether the input channel is closed.

func (*InputChannel) RemoveSubscriberChannel

func (ic *InputChannel) RemoveSubscriberChannel(peerID string)

RemoveSubscriberChannel removes a subscriber's data channel.

func (*InputChannel) SetPublisherChannel

func (ic *InputChannel) SetPublisherChannel(dc *webrtc.DataChannel)

SetPublisherChannel sets the data channel connected to the publisher. Input events from subscribers will be forwarded to this channel.

func (*InputChannel) Stats

func (ic *InputChannel) Stats() InputChannelStats

Stats returns current input channel statistics.

func (*InputChannel) SubscriberCount

func (ic *InputChannel) SubscriberCount() int

SubscriberCount returns the number of subscriber channels.

type InputChannelStats

type InputChannelStats struct {
	HasPublisherChannel  bool
	SubscriberCount      int
	PublisherChannelOpen bool
}

InputChannelStats contains input channel statistics.

type Peer

type Peer struct {
	ID         string
	Role       PeerRole
	Connection *webrtc.PeerConnection
	Logger     *zap.Logger
	// contains filtered or unexported fields
}

Peer wraps a WebRTC peer connection with SFU-specific functionality.

func NewPeer

func NewPeer(config PeerConfig, conn *webrtc.PeerConnection) *Peer

NewPeer creates a new Peer with the given WebRTC connection.

func (*Peer) AddICECandidate

func (p *Peer) AddICECandidate(candidate webrtc.ICECandidateInit) error

AddICECandidate adds a remote ICE candidate.

func (*Peer) AddTrack

func (p *Peer) AddTrack(track *webrtc.TrackLocalStaticRTP) (*webrtc.RTPSender, error)

AddTrack adds a local track to the peer connection.

func (*Peer) Close

func (p *Peer) Close(_ context.Context) error

Close closes the peer connection.

func (*Peer) CreateAnswer

func (p *Peer) CreateAnswer() (*webrtc.SessionDescription, error)

CreateAnswer creates an SDP answer.

func (*Peer) CreateDataChannel

func (p *Peer) CreateDataChannel(label string, opts *webrtc.DataChannelInit) (*webrtc.DataChannel, error)

CreateDataChannel creates a new data channel.

func (*Peer) CreateOffer

func (p *Peer) CreateOffer() (*webrtc.SessionDescription, error)

CreateOffer creates an SDP offer.

func (*Peer) GetDataChannel

func (p *Peer) GetDataChannel(label string) (*webrtc.DataChannel, bool)

GetDataChannel returns a data channel by label.

func (*Peer) LocalDescription

func (p *Peer) LocalDescription() *webrtc.SessionDescription

LocalDescription returns the local SDP.

func (*Peer) OnDataChannel

func (p *Peer) OnDataChannel(handler func(*webrtc.DataChannel))

OnDataChannel sets the callback for data channels.

func (*Peer) OnICECandidate

func (p *Peer) OnICECandidate(handler func(*webrtc.ICECandidate))

OnICECandidate sets the callback for ICE candidates.

func (*Peer) OnStateChange

func (p *Peer) OnStateChange(handler func(PeerState))

OnStateChange sets the callback for state changes.

func (*Peer) OnTrack

func (p *Peer) OnTrack(handler func(*webrtc.TrackRemote, *webrtc.RTPReceiver))

OnTrack sets the callback for when a track is received.

func (*Peer) RemoteDescription

func (p *Peer) RemoteDescription() *webrtc.SessionDescription

RemoteDescription returns the remote SDP.

func (*Peer) RemoveTrack

func (p *Peer) RemoveTrack(sender *webrtc.RTPSender) error

RemoveTrack removes a local track from the peer connection.

func (*Peer) SetRemoteDescription

func (p *Peer) SetRemoteDescription(sdp webrtc.SessionDescription) error

SetRemoteDescription sets the remote SDP.

func (*Peer) State

func (p *Peer) State() PeerState

State returns the current peer state.

type PeerConfig

type PeerConfig struct {
	ID     string
	Role   PeerRole
	Logger *zap.Logger
}

PeerConfig contains configuration for creating a peer.

type PeerRole

type PeerRole string

PeerRole defines the role of a peer connection.

const (
	// PeerRolePublisher indicates a peer that sends media.
	PeerRolePublisher PeerRole = "publisher"

	// PeerRoleSubscriber indicates a peer that receives media.
	PeerRoleSubscriber PeerRole = "subscriber"
)

type PeerState

type PeerState string

PeerState represents the connection state of a peer.

const (
	PeerStateNew          PeerState = "new"
	PeerStateConnecting   PeerState = "connecting"
	PeerStateConnected    PeerState = "connected"
	PeerStateDisconnected PeerState = "disconnected"
	PeerStateFailed       PeerState = "failed"
	PeerStateClosed       PeerState = "closed"
)

type Room

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

Room represents a streaming room with one publisher and multiple subscribers.

func (*Room) AddSubscriber

func (r *Room) AddSubscriber(ctx context.Context, peerID string) (*Peer, error)

AddSubscriber adds a new subscriber to the room.

func (*Room) Close

func (r *Room) Close(ctx context.Context) error

Close closes the room and all connections.

func (*Room) GetPublisher

func (r *Room) GetPublisher() *Peer

GetPublisher returns the publisher peer.

func (*Room) GetSubscriber

func (r *Room) GetSubscriber(peerID string) (*Peer, bool)

GetSubscriber returns a subscriber by ID.

func (*Room) GetTrackInfo

func (r *Room) GetTrackInfo() []TrackInfo

GetTrackInfo returns information about all tracks in the room.

func (*Room) HasPublisher

func (r *Room) HasPublisher() bool

HasPublisher returns whether the room has a publisher.

func (*Room) IsClosed

func (r *Room) IsClosed() bool

IsClosed returns whether the room is closed.

func (*Room) RemoveSubscriber

func (r *Room) RemoveSubscriber(ctx context.Context, peerID string) error

RemoveSubscriber removes a subscriber from the room.

func (*Room) SetPublisher

func (r *Room) SetPublisher(ctx context.Context, peerID string) (*Peer, error)

SetPublisher sets the publisher for this room. Only one publisher is allowed per room.

func (*Room) Stats

func (r *Room) Stats() RoomStats

Stats returns room statistics.

func (*Room) StreamID

func (r *Room) StreamID() string

StreamID returns the room's stream ID.

func (*Room) SubscriberCount

func (r *Room) SubscriberCount() int

SubscriberCount returns the number of subscribers.

func (*Room) SubscriberIDs

func (r *Room) SubscriberIDs() []string

SubscriberIDs returns the IDs of all subscribers.

type RoomStats

type RoomStats struct {
	StreamID        string
	HasPublisher    bool
	SubscriberCount int
	TrackCount      int
}

RoomStats contains room statistics.

type SFU

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

SFU manages WebRTC peer connections for selective forwarding.

func New

func New(config Config, logger *zap.Logger) (*SFU, error)

New creates a new SFU instance.

func (*SFU) Close

func (s *SFU) Close(ctx context.Context) error

Close closes the SFU and all rooms.

func (*SFU) Config

func (s *SFU) Config() Config

Config returns the SFU configuration (copy).

func (*SFU) CreateRoom

func (s *SFU) CreateRoom(streamID string) (*Room, error)

CreateRoom creates a new room for a stream.

func (*SFU) GetOrCreateRoom

func (s *SFU) GetOrCreateRoom(streamID string) (*Room, error)

GetOrCreateRoom returns an existing room or creates a new one.

func (*SFU) GetRoom

func (s *SFU) GetRoom(streamID string) (*Room, bool)

GetRoom returns a room by stream ID.

func (*SFU) GetStats

func (s *SFU) GetStats() Stats

GetStats returns current SFU statistics.

func (*SFU) ListRooms

func (s *SFU) ListRooms() []string

ListRooms returns all active room IDs.

func (*SFU) RemoveRoom

func (s *SFU) RemoveRoom(ctx context.Context, streamID string) error

RemoveRoom removes a room and closes all connections.

func (*SFU) RoomCount

func (s *SFU) RoomCount() int

RoomCount returns the number of active rooms.

type SignalingHandler

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

SignalingHandler handles signaling messages for WebRTC connections.

func NewSignalingHandler

func NewSignalingHandler(sfu *SFU, logger *zap.Logger) *SignalingHandler

NewSignalingHandler creates a new SignalingHandler.

func (*SignalingHandler) Close

func (h *SignalingHandler) Close()

Close closes all sessions.

func (*SignalingHandler) CreateOfferForSubscriber

func (h *SignalingHandler) CreateOfferForSubscriber(ctx context.Context, streamID, peerID string) (*SignalingMessage, error)

CreateOfferForSubscriber creates an offer for a subscriber. This is useful when tracks are added after the initial connection.

func (*SignalingHandler) GetSession

func (h *SignalingHandler) GetSession(peerID string) (*peerSession, bool)

GetSession returns the session for a peer.

func (*SignalingHandler) HandleMessage

func (h *SignalingHandler) HandleMessage(ctx context.Context, msg *SignalingMessage) error

HandleMessage processes a signaling message.

func (*SignalingHandler) OnSend

func (h *SignalingHandler) OnSend(callback func(streamID, peerID string, msg *SignalingMessage))

OnSend sets the callback for sending messages to peers. The callback is invoked when the handler needs to send a message to a specific peer (identified by streamID and peerID).

func (*SignalingHandler) SessionCount

func (h *SignalingHandler) SessionCount() int

SessionCount returns the number of active sessions.

type SignalingMessage

type SignalingMessage struct {
	// Type is the message type.
	Type SignalingMessageType `json:"type"`

	// StreamID identifies the stream/room.
	StreamID string `json:"stream_id,omitempty"`

	// PeerID identifies the peer sending/receiving the message.
	PeerID string `json:"peer_id,omitempty"`

	// Role is the peer's role (publisher or subscriber).
	Role PeerRole `json:"role,omitempty"`

	// SDP contains the Session Description Protocol data for offer/answer.
	SDP string `json:"sdp,omitempty"`

	// Candidate contains ICE candidate information.
	Candidate *ICECandidateJSON `json:"candidate,omitempty"`

	// Error contains error information when Type is SignalingTypeError.
	Error string `json:"error,omitempty"`

	// ErrorCode is a machine-readable error code.
	ErrorCode string `json:"error_code,omitempty"`
}

SignalingMessage represents a signaling message exchanged between clients and the SFU server.

func NewAnswerMessage

func NewAnswerMessage(streamID, peerID, sdp string) *SignalingMessage

NewAnswerMessage creates a new answer signaling message.

func NewCandidateMessage

func NewCandidateMessage(streamID, peerID string, candidate *ICECandidateJSON) *SignalingMessage

NewCandidateMessage creates a new ICE candidate signaling message.

func NewErrorMessage

func NewErrorMessage(streamID, peerID, errorCode, errorMsg string) *SignalingMessage

NewErrorMessage creates a new error signaling message.

func NewJoinMessage

func NewJoinMessage(streamID, peerID string, role PeerRole) *SignalingMessage

NewJoinMessage creates a new join signaling message.

func NewLeaveMessage

func NewLeaveMessage(streamID, peerID string) *SignalingMessage

NewLeaveMessage creates a new leave signaling message.

func NewOfferMessage

func NewOfferMessage(streamID, peerID, sdp string) *SignalingMessage

NewOfferMessage creates a new offer signaling message.

func NewPingMessage

func NewPingMessage() *SignalingMessage

NewPingMessage creates a new ping signaling message.

func NewPongMessage

func NewPongMessage() *SignalingMessage

NewPongMessage creates a new pong signaling message.

func NewReadyMessage

func NewReadyMessage(streamID, peerID string) *SignalingMessage

NewReadyMessage creates a new ready signaling message.

func ParseSignalingMessage

func ParseSignalingMessage(data []byte) (*SignalingMessage, error)

ParseSignalingMessage parses a JSON-encoded signaling message.

func (*SignalingMessage) ToJSON

func (m *SignalingMessage) ToJSON() ([]byte, error)

ToJSON serializes the signaling message to JSON.

func (*SignalingMessage) Validate

func (m *SignalingMessage) Validate() error

Validate validates the signaling message.

type SignalingMessageType

type SignalingMessageType string

SignalingMessageType defines the type of signaling message.

const (
	// SignalingTypeOffer is sent when creating an SDP offer.
	SignalingTypeOffer SignalingMessageType = "offer"

	// SignalingTypeAnswer is sent in response to an offer.
	SignalingTypeAnswer SignalingMessageType = "answer"

	// SignalingTypeCandidate is sent when an ICE candidate is discovered.
	SignalingTypeCandidate SignalingMessageType = "candidate"

	// SignalingTypePing is a keepalive message.
	SignalingTypePing SignalingMessageType = "ping"

	// SignalingTypePong is a response to ping.
	SignalingTypePong SignalingMessageType = "pong"

	// SignalingTypeError indicates an error occurred.
	SignalingTypeError SignalingMessageType = "error"

	// SignalingTypeJoin is sent when a peer wants to join a room.
	SignalingTypeJoin SignalingMessageType = "join"

	// SignalingTypeLeave is sent when a peer leaves a room.
	SignalingTypeLeave SignalingMessageType = "leave"

	// SignalingTypeReady indicates the peer is ready to receive media.
	SignalingTypeReady SignalingMessageType = "ready"
)

type Stats

type Stats struct {
	RoomCount        int
	TotalPublishers  int
	TotalSubscribers int
}

Stats returns SFU statistics.

type TrackInfo

type TrackInfo struct {
	ID              string
	Kind            string
	Codec           string
	SubscriberCount int
}

TrackInfo contains information about a routed track.

type TrackRouter

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

TrackRouter handles forwarding tracks from publisher to subscribers.

func NewTrackRouter

func NewTrackRouter(logger *zap.Logger) *TrackRouter

NewTrackRouter creates a new TrackRouter.

func (*TrackRouter) AddPublisherTrack

func (r *TrackRouter) AddPublisherTrack(remote *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) error

AddPublisherTrack adds a track from the publisher to be forwarded.

func (*TrackRouter) AddSubscriber

func (r *TrackRouter) AddSubscriber(peer *Peer) error

AddSubscriber adds all current tracks to a subscriber.

func (*TrackRouter) Close

func (r *TrackRouter) Close()

Close stops all track forwarding.

func (*TrackRouter) GetLocalTrack

func (r *TrackRouter) GetLocalTrack(trackID string) (*webrtc.TrackLocalStaticRTP, bool)

GetLocalTrack returns the local track for a given track ID.

func (*TrackRouter) GetTrackInfo

func (r *TrackRouter) GetTrackInfo() []TrackInfo

GetTrackInfo returns information about all tracks.

func (*TrackRouter) IsClosed

func (r *TrackRouter) IsClosed() bool

IsClosed returns whether the router is closed.

func (*TrackRouter) RemoveSubscriber

func (r *TrackRouter) RemoveSubscriber(peerID string)

RemoveSubscriber removes all tracks from a subscriber.

func (*TrackRouter) TrackCount

func (r *TrackRouter) TrackCount() int

TrackCount returns the number of tracks being routed.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
⌂ 换网址