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
- type Config
- type ICECandidateJSON
- type InputChannel
- func (ic *InputChannel) AddSubscriberChannel(peerID string, dc *webrtc.DataChannel) error
- func (ic *InputChannel) Close()
- func (ic *InputChannel) HasPublisherChannel() bool
- func (ic *InputChannel) IsClosed() bool
- func (ic *InputChannel) RemoveSubscriberChannel(peerID string)
- func (ic *InputChannel) SetPublisherChannel(dc *webrtc.DataChannel)
- func (ic *InputChannel) Stats() InputChannelStats
- func (ic *InputChannel) SubscriberCount() int
- type InputChannelStats
- type Peer
- func (p *Peer) AddICECandidate(candidate webrtc.ICECandidateInit) error
- func (p *Peer) AddTrack(track *webrtc.TrackLocalStaticRTP) (*webrtc.RTPSender, error)
- func (p *Peer) Close(_ context.Context) error
- func (p *Peer) CreateAnswer() (*webrtc.SessionDescription, error)
- func (p *Peer) CreateDataChannel(label string, opts *webrtc.DataChannelInit) (*webrtc.DataChannel, error)
- func (p *Peer) CreateOffer() (*webrtc.SessionDescription, error)
- func (p *Peer) GetDataChannel(label string) (*webrtc.DataChannel, bool)
- func (p *Peer) LocalDescription() *webrtc.SessionDescription
- func (p *Peer) OnDataChannel(handler func(*webrtc.DataChannel))
- func (p *Peer) OnICECandidate(handler func(*webrtc.ICECandidate))
- func (p *Peer) OnStateChange(handler func(PeerState))
- func (p *Peer) OnTrack(handler func(*webrtc.TrackRemote, *webrtc.RTPReceiver))
- func (p *Peer) RemoteDescription() *webrtc.SessionDescription
- func (p *Peer) RemoveTrack(sender *webrtc.RTPSender) error
- func (p *Peer) SetRemoteDescription(sdp webrtc.SessionDescription) error
- func (p *Peer) State() PeerState
- type PeerConfig
- type PeerRole
- type PeerState
- type Room
- func (r *Room) AddSubscriber(ctx context.Context, peerID string) (*Peer, error)
- func (r *Room) Close(ctx context.Context) error
- func (r *Room) GetPublisher() *Peer
- func (r *Room) GetSubscriber(peerID string) (*Peer, bool)
- func (r *Room) GetTrackInfo() []TrackInfo
- func (r *Room) HasPublisher() bool
- func (r *Room) IsClosed() bool
- func (r *Room) RemoveSubscriber(ctx context.Context, peerID string) error
- func (r *Room) SetPublisher(ctx context.Context, peerID string) (*Peer, error)
- func (r *Room) Stats() RoomStats
- func (r *Room) StreamID() string
- func (r *Room) SubscriberCount() int
- func (r *Room) SubscriberIDs() []string
- type RoomStats
- type SFU
- func (s *SFU) Close(ctx context.Context) error
- func (s *SFU) Config() Config
- func (s *SFU) CreateRoom(streamID string) (*Room, error)
- func (s *SFU) GetOrCreateRoom(streamID string) (*Room, error)
- func (s *SFU) GetRoom(streamID string) (*Room, bool)
- func (s *SFU) GetStats() Stats
- func (s *SFU) ListRooms() []string
- func (s *SFU) RemoveRoom(ctx context.Context, streamID string) error
- func (s *SFU) RoomCount() int
- type SignalingHandler
- func (h *SignalingHandler) Close()
- func (h *SignalingHandler) CreateOfferForSubscriber(ctx context.Context, streamID, peerID string) (*SignalingMessage, error)
- func (h *SignalingHandler) GetSession(peerID string) (*peerSession, bool)
- func (h *SignalingHandler) HandleMessage(ctx context.Context, msg *SignalingMessage) error
- func (h *SignalingHandler) OnSend(callback func(streamID, peerID string, msg *SignalingMessage))
- func (h *SignalingHandler) SessionCount() int
- type SignalingMessage
- func NewAnswerMessage(streamID, peerID, sdp string) *SignalingMessage
- func NewCandidateMessage(streamID, peerID string, candidate *ICECandidateJSON) *SignalingMessage
- func NewErrorMessage(streamID, peerID, errorCode, errorMsg string) *SignalingMessage
- func NewJoinMessage(streamID, peerID string, role PeerRole) *SignalingMessage
- func NewLeaveMessage(streamID, peerID string) *SignalingMessage
- func NewOfferMessage(streamID, peerID, sdp string) *SignalingMessage
- func NewPingMessage() *SignalingMessage
- func NewPongMessage() *SignalingMessage
- func NewReadyMessage(streamID, peerID string) *SignalingMessage
- func ParseSignalingMessage(data []byte) (*SignalingMessage, error)
- type SignalingMessageType
- type Stats
- type TrackInfo
- type TrackRouter
- func (r *TrackRouter) AddPublisherTrack(remote *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) error
- func (r *TrackRouter) AddSubscriber(peer *Peer) error
- func (r *TrackRouter) Close()
- func (r *TrackRouter) GetLocalTrack(trackID string) (*webrtc.TrackLocalStaticRTP, bool)
- func (r *TrackRouter) GetTrackInfo() []TrackInfo
- func (r *TrackRouter) IsClosed() bool
- func (r *TrackRouter) RemoveSubscriber(peerID string)
- func (r *TrackRouter) TrackCount() int
Constants ¶
const ( ErrorCodeInvalidMessage = "invalid_message" ErrorCodeRoomNotFound = "room_not_found" ErrorCodeRoomFull = "room_full" ErrorCodePublisherExists = "publisher_exists" ErrorCodePeerNotFound = "peer_not_found" ErrorCodeInternalError = "internal_error" )
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) 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) 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) 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 ¶
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 ¶
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.
type PeerConfig ¶
PeerConfig contains configuration for creating a peer.
type Room ¶
type Room struct {
// contains filtered or unexported fields
}
Room represents a streaming room with one publisher and multiple subscribers.
func (*Room) AddSubscriber ¶
AddSubscriber adds a new subscriber to the room.
func (*Room) GetPublisher ¶
GetPublisher returns the publisher peer.
func (*Room) GetSubscriber ¶
GetSubscriber returns a subscriber by ID.
func (*Room) GetTrackInfo ¶
GetTrackInfo returns information about all tracks in the room.
func (*Room) HasPublisher ¶
HasPublisher returns whether the room has a publisher.
func (*Room) RemoveSubscriber ¶
RemoveSubscriber removes a subscriber from the room.
func (*Room) SetPublisher ¶
SetPublisher sets the publisher for this room. Only one publisher is allowed per room.
func (*Room) SubscriberCount ¶
SubscriberCount returns the number of subscribers.
func (*Room) SubscriberIDs ¶
SubscriberIDs returns the IDs of all subscribers.
type SFU ¶
type SFU struct {
// contains filtered or unexported fields
}
SFU manages WebRTC peer connections for selective forwarding.
func (*SFU) CreateRoom ¶
CreateRoom creates a new room for a stream.
func (*SFU) GetOrCreateRoom ¶
GetOrCreateRoom returns an existing room or creates a new one.
func (*SFU) RemoveRoom ¶
RemoveRoom removes a room and closes all connections.
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) 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 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) 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.