feat(maitred): Update maitred - hookup to the API (#198)

## Description
We are attempting to hookup maitred to the API
Maitred duties will be:
- [ ] Hookup to the API
- [ ]  Wait for signal (from the API) to start Steam
- [ ] Stop signal to stop the gaming session, clean up Steam... and
maybe do the backup

## Summary by CodeRabbit

- **New Features**
- Introduced Docker-based deployment configurations for both the main
and relay applications.
- Added new API endpoints enabling real-time machine messaging and
enhanced IoT operations.
- Expanded database schema and actor types to support improved machine
tracking.

- **Improvements**
- Enhanced real-time communication and relay management with streamlined
room handling.
- Upgraded dependencies, logging, and error handling for greater
stability and performance.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: DatCaptainHorse <DatCaptainHorse@users.noreply.github.com>
Co-authored-by: Kristian Ollikainen <14197772+DatCaptainHorse@users.noreply.github.com>
This commit is contained in:
Wanjohi
2025-04-07 23:23:53 +03:00
committed by GitHub
parent 6990494b34
commit de80f3e6ab
84 changed files with 7357 additions and 1331 deletions

View File

@@ -1,10 +1,13 @@
package relay
package common
import (
"fmt"
"github.com/libp2p/go-reuseport"
"github.com/pion/ice/v4"
"github.com/pion/interceptor"
"github.com/pion/webrtc/v4"
"log"
"log/slog"
"strconv"
)
var globalWebRTCAPI *webrtc.API
@@ -21,6 +24,19 @@ func InitWebRTCAPI() error {
// Media engine
mediaEngine := &webrtc.MediaEngine{}
// Register additional header extensions to reduce latency
// Playout Delay
if err := mediaEngine.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{
URI: ExtensionPlayoutDelay,
}, webrtc.RTPCodecTypeVideo); err != nil {
return err
}
if err := mediaEngine.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{
URI: ExtensionPlayoutDelay,
}, webrtc.RTPCodecTypeAudio); err != nil {
return err
}
// Default codecs cover most of our needs
err = mediaEngine.RegisterDefaultCodecs()
if err != nil {
@@ -66,17 +82,23 @@ func InitWebRTCAPI() error {
muxPort := GetFlags().UDPMuxPort
if muxPort > 0 {
mux, err := ice.NewMultiUDPMuxFromPort(muxPort)
// Use reuseport to allow multiple listeners on the same port
pktListener, err := reuseport.ListenPacket("udp", ":"+strconv.Itoa(muxPort))
if err != nil {
return err
return fmt.Errorf("failed to create UDP listener: %w", err)
}
mux := ice.NewMultiUDPMuxDefault(ice.NewUDPMuxDefault(ice.UDPMuxParams{
UDPConn: pktListener,
}))
slog.Info("Using UDP Mux for WebRTC", "port", muxPort)
settingEngine.SetICEUDPMux(mux)
} else {
// Set the UDP port range used by WebRTC
err = settingEngine.SetEphemeralUDPPortRange(uint16(flags.WebRTCUDPStart), uint16(flags.WebRTCUDPEnd))
if err != nil {
return err
}
}
// Set the UDP port range used by WebRTC
err = settingEngine.SetEphemeralUDPPortRange(uint16(flags.WebRTCUDPStart), uint16(flags.WebRTCUDPEnd))
if err != nil {
return err
}
settingEngine.SetIncludeLoopbackCandidate(true) // Just in case
@@ -107,7 +129,7 @@ func CreatePeerConnection(onClose func()) (*webrtc.PeerConnection, error) {
connectionState == webrtc.PeerConnectionStateClosed {
err = pc.Close()
if err != nil {
log.Printf("Error closing PeerConnection: %s\n", err.Error())
slog.Error("Failed to close PeerConnection", "err", err)
}
onClose()
}

View File

@@ -0,0 +1,19 @@
package common
import (
"crypto/rand"
"crypto/sha256"
"github.com/oklog/ulid/v2"
"time"
)
func NewULID() (ulid.ULID, error) {
return ulid.New(ulid.Timestamp(time.Now()), ulid.Monotonic(rand.Reader, 0))
}
// Helper function to generate PSK from token
func GeneratePSKFromToken(token string) ([]byte, error) {
// Simple hash-based PSK generation (32 bytes for libp2p)
hash := sha256.Sum256([]byte(token))
return hash[:], nil
}

View File

@@ -0,0 +1,11 @@
package common
const (
ExtensionPlayoutDelay string = "http://www.webrtc.org/experiments/rtp-hdrext/playout-delay"
)
// ExtensionMap maps URIs to their IDs based on registration order
// IMPORTANT: This must match the order in which extensions are registered in common.go!
var ExtensionMap = map[string]uint8{
ExtensionPlayoutDelay: 1,
}

View File

@@ -1,9 +1,9 @@
package relay
package common
import (
"flag"
"github.com/pion/webrtc/v4"
"log"
"log/slog"
"net"
"os"
"strconv"
@@ -13,9 +13,10 @@ import (
var globalFlags *Flags
type Flags struct {
Verbose bool // Verbose mode - log more information to console
Debug bool // Debug mode - log deeper debug information to console
Verbose bool // Log everything to console
Debug bool // Enable debug mode, implies Verbose
EndpointPort int // Port for HTTP/S and WS/S endpoint (TCP)
MeshPort int // Port for Mesh connections (TCP)
WebRTCUDPStart int // WebRTC UDP port range start - ignored if UDPMuxPort is set
WebRTCUDPEnd int // WebRTC UDP port range end - ignored if UDPMuxPort is set
STUNServer string // WebRTC STUN server
@@ -24,23 +25,25 @@ type Flags struct {
NAT11IPs []string // WebRTC NAT 1 to 1 IP(s) - allows specifying host IP(s) if behind NAT
TLSCert string // Path to TLS certificate
TLSKey string // Path to TLS key
ControlSecret string // Shared secret for this relay's control endpoint
}
func (flags *Flags) DebugLog() {
log.Println("Relay Flags:")
log.Println("> Verbose: ", flags.Verbose)
log.Println("> Debug: ", flags.Debug)
log.Println("> Endpoint Port: ", flags.EndpointPort)
log.Println("> WebRTC UDP Range Start: ", flags.WebRTCUDPStart)
log.Println("> WebRTC UDP Range End: ", flags.WebRTCUDPEnd)
log.Println("> WebRTC STUN Server: ", flags.STUNServer)
log.Println("> WebRTC UDP Mux Port: ", flags.UDPMuxPort)
log.Println("> Auto Add Local IP: ", flags.AutoAddLocalIP)
for i, ip := range flags.NAT11IPs {
log.Printf("> WebRTC NAT 1 to 1 IP (%d): %s\n", i, ip)
}
log.Println("> Path to TLS Cert: ", flags.TLSCert)
log.Println("> Path to TLS Key: ", flags.TLSKey)
slog.Info("Relay flags",
"verbose", flags.Verbose,
"debug", flags.Debug,
"endpointPort", flags.EndpointPort,
"meshPort", flags.MeshPort,
"webrtcUDPStart", flags.WebRTCUDPStart,
"webrtcUDPEnd", flags.WebRTCUDPEnd,
"stunServer", flags.STUNServer,
"webrtcUDPMux", flags.UDPMuxPort,
"autoAddLocalIP", flags.AutoAddLocalIP,
"webrtcNAT11IPs", strings.Join(flags.NAT11IPs, ","),
"tlsCert", flags.TLSCert,
"tlsKey", flags.TLSKey,
"controlSecret", flags.ControlSecret,
)
}
func getEnvAsInt(name string, defaultVal int) int {
@@ -76,6 +79,7 @@ func InitFlags() {
flag.BoolVar(&globalFlags.Verbose, "verbose", getEnvAsBool("VERBOSE", false), "Verbose mode")
flag.BoolVar(&globalFlags.Debug, "debug", getEnvAsBool("DEBUG", false), "Debug mode")
flag.IntVar(&globalFlags.EndpointPort, "endpointPort", getEnvAsInt("ENDPOINT_PORT", 8088), "HTTP endpoint port")
flag.IntVar(&globalFlags.MeshPort, "meshPort", getEnvAsInt("MESH_PORT", 8089), "Mesh connections TCP port")
flag.IntVar(&globalFlags.WebRTCUDPStart, "webrtcUDPStart", getEnvAsInt("WEBRTC_UDP_START", 10000), "WebRTC UDP port range start")
flag.IntVar(&globalFlags.WebRTCUDPEnd, "webrtcUDPEnd", getEnvAsInt("WEBRTC_UDP_END", 20000), "WebRTC UDP port range end")
flag.StringVar(&globalFlags.STUNServer, "stunServer", getEnvAsString("STUN_SERVER", "stun.l.google.com:19302"), "WebRTC STUN server")
@@ -83,12 +87,20 @@ func InitFlags() {
flag.BoolVar(&globalFlags.AutoAddLocalIP, "autoAddLocalIP", getEnvAsBool("AUTO_ADD_LOCAL_IP", true), "Automatically add local IP to NAT 1 to 1 IPs")
// String with comma separated IPs
nat11IPs := ""
flag.StringVar(&nat11IPs, "webrtcNAT11IPs", getEnvAsString("WEBRTC_NAT_IPS", ""), "WebRTC NAT 1 to 1 IP(s)")
flag.StringVar(&nat11IPs, "webrtcNAT11IPs", getEnvAsString("WEBRTC_NAT_IPS", ""), "WebRTC NAT 1 to 1 IP(s), comma delimited")
flag.StringVar(&globalFlags.TLSCert, "tlsCert", getEnvAsString("TLS_CERT", ""), "Path to TLS certificate")
flag.StringVar(&globalFlags.TLSKey, "tlsKey", getEnvAsString("TLS_KEY", ""), "Path to TLS key")
flag.StringVar(&globalFlags.ControlSecret, "controlSecret", getEnvAsString("CONTROL_SECRET", ""), "Shared secret for control endpoint")
// Parse flags
flag.Parse()
// If debug is enabled, verbose is also enabled
if globalFlags.Debug {
globalFlags.Verbose = true
// If Debug is enabled, set ControlSecret to 1234
globalFlags.ControlSecret = "1234"
}
// ICE STUN servers
globalWebRTCConfig.ICEServers = []webrtc.ICEServer{
{

View File

@@ -1,4 +1,4 @@
package relay
package common
import (
"fmt"

View File

@@ -0,0 +1,48 @@
package common
import (
"context"
"fmt"
"log/slog"
"os"
"strings"
)
type CustomHandler struct {
Handler slog.Handler
}
func (h *CustomHandler) Enabled(_ context.Context, level slog.Level) bool {
return h.Handler.Enabled(nil, level)
}
func (h *CustomHandler) Handle(_ context.Context, r slog.Record) error {
// Format the timestamp as "2006/01/02 15:04:05"
timestamp := r.Time.Format("2006/01/02 15:04:05")
// Convert level to uppercase string (e.g., "INFO")
level := strings.ToUpper(r.Level.String())
// Build the message
msg := fmt.Sprintf("%s %s %s", timestamp, level, r.Message)
// Handle additional attributes if they exist
var attrs []string
r.Attrs(func(a slog.Attr) bool {
attrs = append(attrs, fmt.Sprintf("%s=%v", a.Key, a.Value))
return true
})
if len(attrs) > 0 {
msg += " " + strings.Join(attrs, " ")
}
// Write the formatted message to stdout
_, err := fmt.Fprintln(os.Stdout, msg)
return err
}
func (h *CustomHandler) WithAttrs(attrs []slog.Attr) slog.Handler {
return &CustomHandler{Handler: h.Handler.WithAttrs(attrs)}
}
func (h *CustomHandler) WithGroup(name string) slog.Handler {
return &CustomHandler{Handler: h.Handler.WithGroup(name)}
}

View File

@@ -0,0 +1,101 @@
package common
import (
"errors"
"reflect"
"sync"
)
var (
ErrKeyNotFound = errors.New("key not found")
ErrValueNotPointer = errors.New("value is not a pointer")
ErrFieldNotFound = errors.New("field not found")
ErrTypeMismatch = errors.New("type mismatch")
)
// SafeMap is a generic thread-safe map with its own mutex
type SafeMap[K comparable, V any] struct {
mu sync.RWMutex
m map[K]V
}
// NewSafeMap creates a new SafeMap instance
func NewSafeMap[K comparable, V any]() *SafeMap[K, V] {
return &SafeMap[K, V]{
m: make(map[K]V),
}
}
// Get retrieves a value from the map
func (sm *SafeMap[K, V]) Get(key K) (V, bool) {
sm.mu.RLock()
defer sm.mu.RUnlock()
v, ok := sm.m[key]
return v, ok
}
// Set adds or updates a value in the map
func (sm *SafeMap[K, V]) Set(key K, value V) {
sm.mu.Lock()
defer sm.mu.Unlock()
sm.m[key] = value
}
// Delete removes a key from the map
func (sm *SafeMap[K, V]) Delete(key K) {
sm.mu.Lock()
defer sm.mu.Unlock()
delete(sm.m, key)
}
// Len returns the number of items in the map
func (sm *SafeMap[K, V]) Len() int {
sm.mu.RLock()
defer sm.mu.RUnlock()
return len(sm.m)
}
// Copy creates a shallow copy of the map and returns it
func (sm *SafeMap[K, V]) Copy() map[K]V {
sm.mu.RLock()
defer sm.mu.RUnlock()
copied := make(map[K]V, len(sm.m))
for k, v := range sm.m {
copied[k] = v
}
return copied
}
// Update updates a specific field in the value data
func (sm *SafeMap[K, V]) Update(key K, fieldName string, newValue any) error {
sm.mu.Lock()
defer sm.mu.Unlock()
v, ok := sm.m[key]
if !ok {
return ErrKeyNotFound
}
// Use reflect to update the field
rv := reflect.ValueOf(v)
if rv.Kind() != reflect.Ptr {
return ErrValueNotPointer
}
rv = rv.Elem()
// Check if the field exists
field := rv.FieldByName(fieldName)
if !field.IsValid() || !field.CanSet() {
return ErrFieldNotFound
}
newRV := reflect.ValueOf(newValue)
if newRV.Type() != field.Type() {
return ErrTypeMismatch
}
field.Set(newRV)
sm.m[key] = v
return nil
}

View File

@@ -1,9 +1,9 @@
package relay
package connections
import (
"github.com/pion/webrtc/v4"
"google.golang.org/protobuf/proto"
"log"
"log/slog"
gen "relay/internal/proto"
)
@@ -30,7 +30,7 @@ func NewNestriDataChannel(dc *webrtc.DataChannel) *NestriDataChannel {
// Decode message
var base gen.ProtoMessageInput
if err := proto.Unmarshal(msg.Data, &base); err != nil {
log.Printf("Failed to decode binary DataChannel message, reason: %s\n", err)
slog.Error("failed to decode binary DataChannel message", "err", err)
return
}

View File

@@ -1,17 +1,15 @@
package relay
package connections
import (
"github.com/pion/webrtc/v4"
"relay/internal/common"
"time"
)
// OnMessageCallback is a callback for messages of given type
type OnMessageCallback func(data []byte)
// MessageBase is the base type for WS/DC messages.
type MessageBase struct {
PayloadType string `json:"payload_type"`
Latency *LatencyTracker `json:"latency,omitempty"`
PayloadType string `json:"payload_type"`
Latency *common.LatencyTracker `json:"latency,omitempty"`
}
// MessageLog represents a log message.

View File

@@ -0,0 +1,119 @@
package connections
import (
"github.com/pion/webrtc/v4"
"google.golang.org/protobuf/proto"
gen "relay/internal/proto"
)
// SendMeshHandshake sends a handshake message to another relay.
func (ws *SafeWebSocket) SendMeshHandshake(relayID, publicKey string) error {
msg := &gen.MeshMessage{
Type: &gen.MeshMessage_Handshake{
Handshake: &gen.Handshake{
RelayId: relayID,
DhPublicKey: publicKey,
},
},
}
data, err := proto.Marshal(msg)
if err != nil {
return err
}
return ws.SendBinary(data)
}
// SendMeshHandshakeResponse sends a handshake response to a relay.
func (ws *SafeWebSocket) SendMeshHandshakeResponse(relayID, dhPublicKey string, approvals map[string]string) error {
msg := &gen.MeshMessage{
Type: &gen.MeshMessage_HandshakeResponse{
HandshakeResponse: &gen.HandshakeResponse{
RelayId: relayID,
DhPublicKey: dhPublicKey,
Approvals: approvals,
},
},
}
data, err := proto.Marshal(msg)
if err != nil {
return err
}
return ws.SendBinary(data)
}
// SendMeshForwardSDP sends a forwarded SDP message to another relay
func (ws *SafeWebSocket) SendMeshForwardSDP(roomName, participantID string, sdp webrtc.SessionDescription) error {
msg := &gen.MeshMessage{
Type: &gen.MeshMessage_ForwardSdp{
ForwardSdp: &gen.ForwardSDP{
RoomName: roomName,
ParticipantId: participantID,
Sdp: sdp.SDP,
Type: sdp.Type.String(),
},
},
}
data, err := proto.Marshal(msg)
if err != nil {
return err
}
return ws.SendBinary(data)
}
// SendMeshForwardICE sends a forwarded ICE candidate to another relay
func (ws *SafeWebSocket) SendMeshForwardICE(roomName, participantID string, candidate webrtc.ICECandidateInit) error {
var sdpMLineIndex uint32
if candidate.SDPMLineIndex != nil {
sdpMLineIndex = uint32(*candidate.SDPMLineIndex)
}
msg := &gen.MeshMessage{
Type: &gen.MeshMessage_ForwardIce{
ForwardIce: &gen.ForwardICE{
RoomName: roomName,
ParticipantId: participantID,
Candidate: &gen.ICECandidateInit{
Candidate: candidate.Candidate,
SdpMid: candidate.SDPMid,
SdpMLineIndex: &sdpMLineIndex,
UsernameFragment: candidate.UsernameFragment,
},
},
},
}
data, err := proto.Marshal(msg)
if err != nil {
return err
}
return ws.SendBinary(data)
}
func (ws *SafeWebSocket) SendMeshForwardIngest(roomName string) error {
msg := &gen.MeshMessage{
Type: &gen.MeshMessage_ForwardIngest{
ForwardIngest: &gen.ForwardIngest{
RoomName: roomName,
},
},
}
data, err := proto.Marshal(msg)
if err != nil {
return err
}
return ws.SendBinary(data)
}
func (ws *SafeWebSocket) SendMeshStreamRequest(roomName string) error {
msg := &gen.MeshMessage{
Type: &gen.MeshMessage_StreamRequest{
StreamRequest: &gen.StreamRequest{
RoomName: roomName,
},
},
}
data, err := proto.Marshal(msg)
if err != nil {
return err
}
return ws.SendBinary(data)
}

View File

@@ -1,26 +1,37 @@
package relay
package connections
import (
"encoding/json"
"github.com/gorilla/websocket"
"log"
"log/slog"
"sync"
)
// OnMessageCallback is a callback for messages of given type
type OnMessageCallback func(data []byte)
// SafeWebSocket is a websocket with a mutex
type SafeWebSocket struct {
*websocket.Conn
sync.Mutex
closeCallback func() // OnClose callback
callbacks map[string]OnMessageCallback // MessageBase type -> callback
closed bool
closeCallback func() // Callback to call on close
closeChan chan struct{} // Channel to signal closure
callbacks map[string]OnMessageCallback // MessageBase type -> callback
binaryCallback OnMessageCallback // Binary message callback
sharedSecret []byte
}
// NewSafeWebSocket creates a new SafeWebSocket from *websocket.Conn
func NewSafeWebSocket(conn *websocket.Conn) *SafeWebSocket {
ws := &SafeWebSocket{
Conn: conn,
closeCallback: nil,
callbacks: make(map[string]OnMessageCallback),
Conn: conn,
closed: false,
closeCallback: nil,
closeChan: make(chan struct{}),
callbacks: make(map[string]OnMessageCallback),
binaryCallback: nil,
sharedSecret: nil,
}
// Launch a goroutine to handle messages
@@ -30,14 +41,12 @@ func NewSafeWebSocket(conn *websocket.Conn) *SafeWebSocket {
kind, data, err := ws.Conn.ReadMessage()
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure, websocket.CloseNoStatusReceived) {
// If unexpected close error, break
if GetFlags().Verbose {
log.Printf("Unexpected WebSocket close error, reason: %s\n", err)
}
slog.Debug("WebSocket closed unexpectedly", "err", err)
break
} else if websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway, websocket.CloseAbnormalClosure, websocket.CloseNoStatusReceived) {
break
} else if err != nil {
log.Printf("Failed to read WebSocket message, reason: %s\n", err)
slog.Error("Failed reading WebSocket message", "err", err)
break
}
@@ -46,32 +55,48 @@ func NewSafeWebSocket(conn *websocket.Conn) *SafeWebSocket {
// Decode message
var msg MessageBase
if err = json.Unmarshal(data, &msg); err != nil {
log.Printf("Failed to decode text WebSocket message, reason: %s\n", err)
slog.Error("Failed decoding WebSocket message", "err", err)
continue
}
// Handle message type callback
if callback, ok := ws.callbacks[msg.PayloadType]; ok {
callback(data)
} // TODO: Log unknown message type?
} // TODO: Log unknown message payload type?
break
case websocket.BinaryMessage:
// Handle binary message callback
if ws.binaryCallback != nil {
ws.binaryCallback(data)
}
break
default:
log.Printf("Unknown WebSocket message type: %d\n", kind)
slog.Warn("Unknown WebSocket message type", "type", kind)
break
}
}
// Call close callback
// Signal closure to callback first
if ws.closeCallback != nil {
ws.closeCallback()
}
close(ws.closeChan)
ws.closed = true
}()
return ws
}
// SetSharedSecret sets the shared secret for the websocket
func (ws *SafeWebSocket) SetSharedSecret(secret []byte) {
ws.sharedSecret = secret
}
// GetSharedSecret returns the shared secret for the websocket
func (ws *SafeWebSocket) GetSharedSecret() []byte {
return ws.sharedSecret
}
// SendJSON writes JSON to a websocket with a mutex
func (ws *SafeWebSocket) SendJSON(v interface{}) error {
ws.Lock()
@@ -88,31 +113,46 @@ func (ws *SafeWebSocket) SendBinary(data []byte) error {
// RegisterMessageCallback sets the callback for binary message of given type
func (ws *SafeWebSocket) RegisterMessageCallback(msgType string, callback OnMessageCallback) {
ws.Lock()
defer ws.Unlock()
if ws.callbacks == nil {
ws.callbacks = make(map[string]OnMessageCallback)
}
ws.callbacks[msgType] = callback
}
// RegisterBinaryMessageCallback sets the callback for all binary messages
func (ws *SafeWebSocket) RegisterBinaryMessageCallback(callback OnMessageCallback) {
ws.binaryCallback = callback
}
// UnregisterMessageCallback removes the callback for binary message of given type
func (ws *SafeWebSocket) UnregisterMessageCallback(msgType string) {
ws.Lock()
defer ws.Unlock()
if ws.callbacks != nil {
delete(ws.callbacks, msgType)
}
}
// UnregisterBinaryMessageCallback removes the callback for all binary messages
func (ws *SafeWebSocket) UnregisterBinaryMessageCallback() {
ws.binaryCallback = nil
}
// RegisterOnClose sets the callback for websocket closing
func (ws *SafeWebSocket) RegisterOnClose(callback func()) {
ws.closeCallback = func() {
// Clear our callbacks
ws.Lock()
ws.callbacks = nil
ws.Unlock()
ws.binaryCallback = nil
// Call the callback
callback()
}
}
// Closed returns a channel that closes when the WebSocket connection is terminated
func (ws *SafeWebSocket) Closed() <-chan struct{} {
return ws.closeChan
}
// IsClosed returns true if the WebSocket connection is closed
func (ws *SafeWebSocket) IsClosed() bool {
return ws.closed
}

View File

@@ -1,26 +1,26 @@
package relay
package internal
import (
"context"
"encoding/json"
"github.com/pion/webrtc/v4"
"google.golang.org/protobuf/proto"
"log"
"log/slog"
"relay/internal/common"
"relay/internal/connections"
gen "relay/internal/proto"
)
func participantHandler(participant *Participant, room *Room) {
// Callback for closing PeerConnection
func ParticipantHandler(participant *Participant, room *Room, relay *Relay) {
onPCClose := func() {
if GetFlags().Verbose {
log.Printf("Closed PeerConnection for participant: '%s'\n", participant.ID)
}
slog.Debug("Participant PeerConnection closed", "participant", participant.ID, "room", room.Name)
room.removeParticipantByID(participant.ID)
}
var err error
participant.PeerConnection, err = CreatePeerConnection(onPCClose)
participant.PeerConnection, err = common.CreatePeerConnection(onPCClose)
if err != nil {
log.Printf("Failed to create PeerConnection for participant: '%s' - reason: %s\n", participant.ID, err)
slog.Error("Failed to create participant PeerConnection", "participant", participant.ID, "room", room.Name, "err", err)
return
}
@@ -32,67 +32,32 @@ func participantHandler(participant *Participant, room *Room) {
MaxRetransmits: &settingMaxRetransmits,
})
if err != nil {
log.Printf("Failed to create data channel for participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
slog.Error("Failed to create data channel for participant", "participant", participant.ID, "room", room.Name, "err", err)
return
}
participant.DataChannel = NewNestriDataChannel(dc)
participant.DataChannel = connections.NewNestriDataChannel(dc)
// Register channel opening handling
participant.DataChannel.RegisterOnOpen(func() {
if GetFlags().Verbose {
log.Printf("DataChannel open for participant: %s\n", participant.ID)
}
slog.Debug("DataChannel opened for participant", "participant", participant.ID, "room", room.Name)
})
// Register channel closing handling
participant.DataChannel.RegisterOnClose(func() {
if GetFlags().Verbose {
log.Printf("DataChannel closed for participant: %s\n", participant.ID)
}
slog.Debug("DataChannel closed for participant", "participant", participant.ID, "room", room.Name)
})
// Register text message handling
participant.DataChannel.RegisterMessageCallback("input", func(data []byte) {
// Send to room if it has a DataChannel
if room.DataChannel != nil {
// If debug mode, decode and add our timestamp, otherwise just send to room
if GetFlags().Debug {
var inputMsg gen.ProtoMessageInput
if err = proto.Unmarshal(data, &inputMsg); err != nil {
log.Printf("Failed to decode input message from participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
return
}
protoLat := inputMsg.GetMessageBase().GetLatency()
if protoLat != nil {
lat := LatencyTrackerFromProto(protoLat)
lat.AddTimestamp("relay_to_node")
protoLat = lat.ToProto()
}
// Marshal and send
if data, err = proto.Marshal(&inputMsg); err != nil {
log.Printf("Failed to marshal input message for participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
return
}
}
if err = room.DataChannel.SendBinary(data); err != nil {
log.Printf("Failed to send input message to room: '%s' - reason: %s\n", room.Name, err)
}
}
ForwardParticipantDataChannelMessage(participant, room, data)
})
participant.PeerConnection.OnICECandidate(func(candidate *webrtc.ICECandidate) {
if candidate == nil {
return
}
if GetFlags().Verbose {
log.Printf("ICE candidate for participant: '%s' in room: '%s'\n", participant.ID, room.Name)
}
err = participant.WebSocket.SendICECandidateMessageWS(candidate.ToJSON())
if err != nil {
log.Printf("Failed to send ICE candidate for participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
if err := participant.WebSocket.SendICECandidateMessageWS(candidate.ToJSON()); err != nil {
slog.Error("Failed to send ICE candidate to participant", "participant", participant.ID, "room", room.Name, "err", err)
}
})
@@ -100,35 +65,32 @@ func participantHandler(participant *Participant, room *Room) {
// ICE callback
participant.WebSocket.RegisterMessageCallback("ice", func(data []byte) {
var iceMsg MessageICECandidate
var iceMsg connections.MessageICECandidate
if err = json.Unmarshal(data, &iceMsg); err != nil {
log.Printf("Failed to decode ICE message from participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
slog.Error("Failed to decode ICE candidate message from participant", "participant", participant.ID, "room", room.Name, "err", err)
return
}
candidate := webrtc.ICECandidateInit{
Candidate: iceMsg.Candidate.Candidate,
}
if participant.PeerConnection.RemoteDescription() != nil {
if err = participant.PeerConnection.AddICECandidate(candidate); err != nil {
log.Printf("Failed to add ICE candidate from participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
if err = participant.PeerConnection.AddICECandidate(iceMsg.Candidate); err != nil {
slog.Error("Failed to add ICE candidate for participant", "participant", participant.ID, "room", room.Name, "err", err)
}
// Add held ICE candidates
for _, heldCandidate := range iceHolder {
if err = participant.PeerConnection.AddICECandidate(heldCandidate); err != nil {
log.Printf("Failed to add held ICE candidate from participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
slog.Error("Failed to add held ICE candidate for participant", "participant", participant.ID, "room", room.Name, "err", err)
}
}
iceHolder = nil
} else {
iceHolder = append(iceHolder, candidate)
iceHolder = append(iceHolder, iceMsg.Candidate)
}
})
// SDP answer callback
participant.WebSocket.RegisterMessageCallback("sdp", func(data []byte) {
var sdpMsg MessageSDP
var sdpMsg connections.MessageSDP
if err = json.Unmarshal(data, &sdpMsg); err != nil {
log.Printf("Failed to decode SDP message from participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
slog.Error("Failed to decode SDP message from participant", "participant", participant.ID, "room", room.Name, "err", err)
return
}
handleParticipantSDP(participant, sdpMsg)
@@ -136,9 +98,9 @@ func participantHandler(participant *Participant, room *Room) {
// Log callback
participant.WebSocket.RegisterMessageCallback("log", func(data []byte) {
var logMsg MessageLog
var logMsg connections.MessageLog
if err = json.Unmarshal(data, &logMsg); err != nil {
log.Printf("Failed to decode log message from participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
slog.Error("Failed to decode log message from participant", "participant", participant.ID, "room", room.Name, "err", err)
return
}
// TODO: Handle log message sending to metrics server
@@ -150,38 +112,36 @@ func participantHandler(participant *Participant, room *Room) {
})
participant.WebSocket.RegisterOnClose(func() {
if GetFlags().Verbose {
log.Printf("WebSocket closed for participant: '%s' in room: '%s'\n", participant.ID, room.Name)
}
slog.Debug("WebSocket closed for participant", "participant", participant.ID, "room", room.Name)
// Remove from Room
room.removeParticipantByID(participant.ID)
})
log.Printf("Participant: '%s' in room: '%s' is now ready, sending an OK\n", participant.ID, room.Name)
if err = participant.WebSocket.SendAnswerMessageWS(AnswerOK); err != nil {
log.Printf("Failed to send OK answer for participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
slog.Info("Participant ready, sending OK answer", "participant", participant.ID, "room", room.Name)
if err := participant.WebSocket.SendAnswerMessageWS(connections.AnswerOK); err != nil {
slog.Error("Failed to send OK answer", "participant", participant.ID, "room", room.Name, "err", err)
}
// If room is already online, send also offer
// If room is online, also send offer
if room.Online {
if room.AudioTrack != nil {
if err = participant.addTrack(&room.AudioTrack); err != nil {
log.Printf("Failed to add audio track for participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
}
if err = room.signalParticipantWithTracks(participant); err != nil {
slog.Error("Failed to signal participant with tracks", "participant", participant.ID, "room", room.Name, "err", err)
}
if room.VideoTrack != nil {
if err = participant.addTrack(&room.VideoTrack); err != nil {
log.Printf("Failed to add video track for participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
} else {
active, provider := relay.IsRoomActive(room.ID)
if active {
slog.Debug("Room active remotely, requesting stream", "room", room.Name, "provider", provider)
if _, err := relay.requestStream(context.Background(), room.Name, room.ID, provider); err != nil {
slog.Error("Failed to request stream", "room", room.Name, "err", err)
} else {
slog.Debug("Stream requested successfully", "room", room.Name, "provider", provider)
}
}
if err = participant.signalOffer(); err != nil {
log.Printf("Failed to signal offer for participant: '%s' in room: '%s' - reason: %s\n", participant.ID, room.Name, err)
}
}
}
// SDP answer handler for participants
func handleParticipantSDP(participant *Participant, answerMsg MessageSDP) {
func handleParticipantSDP(participant *Participant, answerMsg connections.MessageSDP) {
// Get SDP offer
sdpAnswer := answerMsg.SDP.SDP
@@ -191,6 +151,37 @@ func handleParticipantSDP(participant *Participant, answerMsg MessageSDP) {
SDP: sdpAnswer,
})
if err != nil {
log.Printf("Failed to set remote description for participant: '%s' - reason: %s\n", participant.ID, err)
slog.Error("Failed to set remote SDP answer for participant", "participant", participant.ID, "err", err)
}
}
func ForwardParticipantDataChannelMessage(participant *Participant, room *Room, data []byte) {
// Debug mode: Add latency timestamp
if common.GetFlags().Debug {
var inputMsg gen.ProtoMessageInput
if err := proto.Unmarshal(data, &inputMsg); err != nil {
slog.Error("Failed to decode input message from participant", "participant", participant.ID, "room", room.Name, "err", err)
return
}
protoLat := inputMsg.GetMessageBase().GetLatency()
if protoLat != nil {
lat := common.LatencyTrackerFromProto(protoLat)
lat.AddTimestamp("relay_to_node")
protoLat = lat.ToProto()
}
if newData, err := proto.Marshal(&inputMsg); err != nil {
slog.Error("Failed to marshal input message from participant", "participant", participant.ID, "room", room.Name, "err", err)
return
} else {
// Update data with the modified message
data = newData
}
}
// Forward to local room DataChannel if it exists (e.g., local ingest)
if room.DataChannel != nil {
if err := room.DataChannel.SendBinary(data); err != nil {
slog.Error("Failed to send input message to room", "participant", participant.ID, "room", room.Name, "err", err)
}
}
}

View File

@@ -1,45 +1,61 @@
package relay
package internal
import (
"context"
"encoding/json"
"errors"
"fmt"
"github.com/gorilla/websocket"
"log"
"github.com/libp2p/go-reuseport"
"log/slog"
"net/http"
"relay/internal/common"
"relay/internal/connections"
"strconv"
)
var httpMux *http.ServeMux
func InitHTTPEndpoint() error {
func InitHTTPEndpoint(_ context.Context, ctxCancel context.CancelFunc) error {
// Create HTTP mux which serves our WS endpoint
httpMux = http.NewServeMux()
// Endpoints themselves
httpMux.Handle("/", http.NotFoundHandler())
// If control endpoint secret is set, enable the control endpoint
if len(common.GetFlags().ControlSecret) > 0 {
httpMux.HandleFunc("/api/control", corsAnyHandler(controlHandler))
}
// WS endpoint
httpMux.HandleFunc("/api/ws/{roomName}", corsAnyHandler(wsHandler))
// Get our serving port
port := GetFlags().EndpointPort
tlsCert := GetFlags().TLSCert
tlsKey := GetFlags().TLSKey
port := common.GetFlags().EndpointPort
tlsCert := common.GetFlags().TLSCert
tlsKey := common.GetFlags().TLSKey
// Create re-usable listener port
httpListener, err := reuseport.Listen("tcp", ":"+strconv.Itoa(port))
if err != nil {
return fmt.Errorf("failed to create TCP listener: %w", err)
}
// Log and start the endpoint server
if len(tlsCert) <= 0 && len(tlsKey) <= 0 {
log.Println("Starting HTTP endpoint server on :", strconv.Itoa(port))
slog.Info("Starting HTTP endpoint server", "port", port)
go func() {
log.Fatal((&http.Server{
Handler: httpMux,
Addr: ":" + strconv.Itoa(port),
}).ListenAndServe())
if err := http.Serve(httpListener, httpMux); err != nil {
slog.Error("Failed to start HTTP server", "err", err)
ctxCancel()
}
}()
} else if len(tlsCert) > 0 && len(tlsKey) > 0 {
log.Println("Starting HTTPS endpoint server on :", strconv.Itoa(port))
slog.Info("Starting HTTPS endpoint server", "port", port)
go func() {
log.Fatal((&http.Server{
Handler: httpMux,
Addr: ":" + strconv.Itoa(port),
}).ListenAndServeTLS(tlsCert, tlsKey))
if err := http.ServeTLS(httpListener, httpMux, tlsCert, tlsKey); err != nil {
slog.Error("Failed to start HTTPS server", "err", err)
ctxCancel()
}
}()
} else {
return errors.New("no TLS certificate or TLS key provided")
@@ -49,8 +65,8 @@ func InitHTTPEndpoint() error {
// logHTTPError logs (if verbose) and sends an error code to requester
func logHTTPError(w http.ResponseWriter, err string, code int) {
if GetFlags().Verbose {
log.Println(err)
if common.GetFlags().Verbose {
slog.Error("HTTP error", "code", code, "message", err)
}
http.Error(w, err, code)
}
@@ -78,8 +94,9 @@ func wsHandler(w http.ResponseWriter, r *http.Request) {
return
}
rel := GetRelay()
// Get or create room in any case
room := GetOrCreateRoom(roomName)
room := rel.GetOrCreateRoom(roomName)
// Upgrade to WebSocket
upgrader := websocket.Upgrader{
@@ -94,47 +111,92 @@ func wsHandler(w http.ResponseWriter, r *http.Request) {
}
// Create SafeWebSocket
ws := NewSafeWebSocket(wsConn)
ws := connections.NewSafeWebSocket(wsConn)
// Assign message handler for join request
ws.RegisterMessageCallback("join", func(data []byte) {
var joinMsg MessageJoin
var joinMsg connections.MessageJoin
if err = json.Unmarshal(data, &joinMsg); err != nil {
log.Printf("Failed to decode join message: %s\n", err)
slog.Error("Failed to unmarshal join message", "err", err)
return
}
if GetFlags().Verbose {
log.Printf("Join request for room: '%s' from: '%s'\n", room.Name, joinMsg.JoinerType.String())
}
slog.Debug("Join message", "room", room.Name, "joinerType", joinMsg.JoinerType)
// Handle join request, depending if it's from ingest/node or participant/client
switch joinMsg.JoinerType {
case JoinerNode:
case connections.JoinerNode:
// If room already online, send InUse answer
if room.Online {
if err = ws.SendAnswerMessageWS(AnswerInUse); err != nil {
log.Printf("Failed to send InUse answer for Room: '%s' - reason: %s\n", room.Name, err)
if err = ws.SendAnswerMessageWS(connections.AnswerInUse); err != nil {
slog.Error("Failed to send InUse answer to node", "room", room.Name, "err", err)
}
return
}
room.assignWebSocket(ws)
go ingestHandler(room)
case JoinerClient:
room.AssignWebSocket(ws)
go IngestHandler(room)
case connections.JoinerClient:
// Create participant and add to room regardless of online status
participant := NewParticipant(ws)
room.addParticipant(participant)
room.AddParticipant(participant)
// If room not online, send Offline answer
if !room.Online {
if err = ws.SendAnswerMessageWS(AnswerOffline); err != nil {
log.Printf("Failed to send Offline answer for Room: '%s' - reason: %s\n", room.Name, err)
if err = ws.SendAnswerMessageWS(connections.AnswerOffline); err != nil {
slog.Error("Failed to send offline answer to participant", "room", room.Name, "err", err)
}
}
go participantHandler(participant, room)
go ParticipantHandler(participant, room, rel)
default:
log.Printf("Unknown joiner type: %d\n", joinMsg.JoinerType)
slog.Error("Unknown joiner type", "joinerType", joinMsg.JoinerType)
}
// Unregister ourselves, if something happens on the other side they should just reconnect?
ws.UnregisterMessageCallback("join")
})
}
// controlMessage is the JSON struct for the control messages
type controlMessage struct {
Type string `json:"type"`
Value string `json:"value"`
}
// controlHandler is the handler for the /api/control endpoint, for controlling this relay
func controlHandler(w http.ResponseWriter, r *http.Request) {
// Check for control secret in Authorization header
authHeader := r.Header.Get("Authorization")
if len(authHeader) <= 0 || authHeader != common.GetFlags().ControlSecret {
logHTTPError(w, "missing or invalid Authorization header", http.StatusUnauthorized)
return
}
// Handle CORS preflight request
if r.Method == http.MethodOptions {
w.WriteHeader(http.StatusOK)
return
}
// Decode the control message
var msg controlMessage
if err := json.NewDecoder(r.Body).Decode(&msg); err != nil {
logHTTPError(w, "failed to decode control message", http.StatusBadRequest)
return
}
//relay := GetRelay()
switch msg.Type {
case "join_mesh":
// Join the mesh network, get relay address from msg.Value
if len(msg.Value) <= 0 {
logHTTPError(w, "missing relay address", http.StatusBadRequest)
return
}
ctx := r.Context()
if err := GetRelay().ConnectToRelay(ctx, msg.Value); err != nil {
http.Error(w, fmt.Sprintf("Failed to connect: %v", err), http.StatusInternalServerError)
return
}
w.Write([]byte("Successfully connected to relay"))
default:
logHTTPError(w, "unknown control message type", http.StatusBadRequest)
}
}

View File

@@ -1,117 +1,96 @@
package relay
package internal
import (
"encoding/json"
"errors"
"fmt"
"github.com/pion/rtp"
"github.com/pion/webrtc/v4"
"io"
"log"
"log/slog"
"relay/internal/common"
"relay/internal/connections"
"strings"
)
func ingestHandler(room *Room) {
func IngestHandler(room *Room) {
relay := GetRelay()
// Callback for closing PeerConnection
onPCClose := func() {
if GetFlags().Verbose {
log.Printf("Closed PeerConnection for room: '%s'\n", room.Name)
}
slog.Debug("ingest PeerConnection closed", "room", room.Name)
room.Online = false
DeleteRoomIfEmpty(room)
room.signalParticipantsOffline()
relay.DeleteRoomIfEmpty(room)
}
var err error
room.PeerConnection, err = CreatePeerConnection(onPCClose)
room.PeerConnection, err = common.CreatePeerConnection(onPCClose)
if err != nil {
log.Printf("Failed to create PeerConnection for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to create ingest PeerConnection", "room", room.Name, "err", err)
return
}
room.PeerConnection.OnTrack(func(remoteTrack *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
var localTrack *webrtc.TrackLocalStaticRTP
if remoteTrack.Kind() == webrtc.RTPCodecTypeVideo {
if GetFlags().Verbose {
log.Printf("Received video track for room: '%s'\n", room.Name)
}
localTrack, err = webrtc.NewTrackLocalStaticRTP(remoteTrack.Codec().RTPCodecCapability, "video", fmt.Sprint("nestri-", room.Name))
if err != nil {
log.Printf("Failed to create local video track for room: '%s' - reason: %s\n", room.Name, err)
return
}
room.VideoTrack = localTrack
} else if remoteTrack.Kind() == webrtc.RTPCodecTypeAudio {
if GetFlags().Verbose {
log.Printf("Received audio track for room: '%s'\n", room.Name)
}
localTrack, err = webrtc.NewTrackLocalStaticRTP(remoteTrack.Codec().RTPCodecCapability, "audio", fmt.Sprint("nestri-", room.Name))
if err != nil {
log.Printf("Failed to create local audio track for room: '%s' - reason: %s\n", room.Name, err)
return
}
room.AudioTrack = localTrack
localTrack, err := webrtc.NewTrackLocalStaticRTP(remoteTrack.Codec().RTPCodecCapability, remoteTrack.Kind().String(), fmt.Sprintf("nestri-%s-%s", room.Name, remoteTrack.Kind().String()))
if err != nil {
slog.Error("Failed to create local track for room", "room", room.Name, "kind", remoteTrack.Kind(), "err", err)
return
}
slog.Debug("Received track for room", "room", room.Name, "kind", remoteTrack.Kind())
// Set track and let Room handle state
room.SetTrack(remoteTrack.Kind(), localTrack)
// Prepare PlayoutDelayExtension so we don't need to recreate it for each packet
playoutExt := &rtp.PlayoutDelayExtension{
MinDelay: 0,
MaxDelay: 0,
}
playoutPayload, err := playoutExt.Marshal()
if err != nil {
slog.Error("Failed to marshal PlayoutDelayExtension for room", "room", room.Name, "err", err)
return
}
// If both audio and video tracks are set, set online state
if room.AudioTrack != nil && room.VideoTrack != nil {
room.Online = true
if GetFlags().Verbose {
log.Printf("Room online and receiving: '%s' - signaling participants\n", room.Name)
}
room.signalParticipantsWithTracks()
}
rtpBuffer := make([]byte, 1400)
for {
read, _, err := remoteTrack.Read(rtpBuffer)
rtpPacket, _, err := remoteTrack.ReadRTP()
if err != nil {
// EOF is expected when stopping room
if !errors.Is(err, io.EOF) {
log.Printf("RTP read error from room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to read RTP from remote track for room", "room", room.Name, "err", err)
}
break
}
_, err = localTrack.Write(rtpBuffer[:read])
// Use PlayoutDelayExtension for low latency, only for video tracks
if err := rtpPacket.SetExtension(common.ExtensionMap[common.ExtensionPlayoutDelay], playoutPayload); err != nil {
slog.Error("Failed to set PlayoutDelayExtension for room", "room", room.Name, "err", err)
continue
}
err = localTrack.WriteRTP(rtpPacket)
if err != nil && !errors.Is(err, io.ErrClosedPipe) {
log.Printf("Failed to write RTP to local track for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to write RTP to local track for room", "room", room.Name, "err", err)
break
}
}
if remoteTrack.Kind() == webrtc.RTPCodecTypeVideo {
room.VideoTrack = nil
} else if remoteTrack.Kind() == webrtc.RTPCodecTypeAudio {
room.AudioTrack = nil
}
slog.Debug("Track closed for room", "room", room.Name, "kind", remoteTrack.Kind())
if room.VideoTrack == nil && room.AudioTrack == nil {
room.Online = false
if GetFlags().Verbose {
log.Printf("Room offline and not receiving: '%s'\n", room.Name)
}
// Signal participants of room offline
room.signalParticipantsOffline()
DeleteRoomIfEmpty(room)
}
// Clear track when done
room.SetTrack(remoteTrack.Kind(), nil)
})
room.PeerConnection.OnDataChannel(func(dc *webrtc.DataChannel) {
room.DataChannel = NewNestriDataChannel(dc)
if GetFlags().Verbose {
log.Printf("New DataChannel for room: '%s' - '%s'\n", room.Name, room.DataChannel.Label())
}
room.DataChannel = connections.NewNestriDataChannel(dc)
slog.Debug("Ingest received DataChannel for room", "room", room.Name)
// Register channel opening handling
room.DataChannel.RegisterOnOpen(func() {
if GetFlags().Verbose {
log.Printf("DataChannel for room: '%s' - '%s' open\n", room.Name, room.DataChannel.Label())
}
slog.Debug("ingest DataChannel opened for room", "room", room.Name)
})
room.DataChannel.OnClose(func() {
if GetFlags().Verbose {
log.Printf("DataChannel for room: '%s' - '%s' closed\n", room.Name, room.DataChannel.Label())
}
slog.Debug("ingest DataChannel closed for room", "room", room.Name)
})
// We do not handle any messages from ingest via DataChannel yet
@@ -121,12 +100,10 @@ func ingestHandler(room *Room) {
if candidate == nil {
return
}
if GetFlags().Verbose {
log.Printf("ICE candidate for room: '%s'\n", room.Name)
}
slog.Debug("ingest received ICECandidate for room", "room", room.Name)
err = room.WebSocket.SendICECandidateMessageWS(candidate.ToJSON())
if err != nil {
log.Printf("Failed to send ICE candidate for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to send ICE candidate message to ingest for room", "room", room.Name, "err", err)
}
})
@@ -134,57 +111,52 @@ func ingestHandler(room *Room) {
// ICE callback
room.WebSocket.RegisterMessageCallback("ice", func(data []byte) {
var iceMsg MessageICECandidate
var iceMsg connections.MessageICECandidate
if err = json.Unmarshal(data, &iceMsg); err != nil {
log.Printf("Failed to decode ICE candidate message from ingest for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to decode ICE candidate message from ingest for room", "room", room.Name, "err", err)
return
}
candidate := webrtc.ICECandidateInit{
Candidate: iceMsg.Candidate.Candidate,
}
if room.PeerConnection != nil {
// If remote isn't set yet, store ICE candidates
if room.PeerConnection.RemoteDescription() != nil {
if err = room.PeerConnection.AddICECandidate(candidate); err != nil {
log.Printf("Failed to add ICE candidate for room: '%s' - reason: %s\n", room.Name, err)
if err = room.PeerConnection.AddICECandidate(iceMsg.Candidate); err != nil {
slog.Error("Failed to add ICE candidate for room", "room", room.Name, "err", err)
}
// Add any held ICE candidates
for _, heldCandidate := range iceHolder {
if err = room.PeerConnection.AddICECandidate(heldCandidate); err != nil {
log.Printf("Failed to add held ICE candidate for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to add held ICE candidate for room", "room", room.Name, "err", err)
}
}
iceHolder = nil
iceHolder = make([]webrtc.ICECandidateInit, 0)
} else {
iceHolder = append(iceHolder, candidate)
iceHolder = append(iceHolder, iceMsg.Candidate)
}
} else {
log.Printf("ICE candidate received before PeerConnection for room: '%s'\n", room.Name)
slog.Error("ICE candidate received but PeerConnection is nil for room", "room", room.Name)
}
})
// SDP offer callback
room.WebSocket.RegisterMessageCallback("sdp", func(data []byte) {
var sdpMsg MessageSDP
var sdpMsg connections.MessageSDP
if err = json.Unmarshal(data, &sdpMsg); err != nil {
log.Printf("Failed to decode SDP message from ingest for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to decode SDP message from ingest for room", "room", room.Name, "err", err)
return
}
answer := handleIngestSDP(room, sdpMsg)
if answer != nil {
if err = room.WebSocket.SendSDPMessageWS(*answer); err != nil {
log.Printf("Failed to send SDP answer to ingest for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to send SDP answer message to ingest for room", "room", room.Name, "err", err)
}
} else {
log.Printf("Failed to handle SDP message from ingest for room: '%s'\n", room.Name)
slog.Error("Failed to handle ingest SDP message for room", "room", room.Name)
}
})
// Log callback
room.WebSocket.RegisterMessageCallback("log", func(data []byte) {
var logMsg MessageLog
var logMsg connections.MessageLog
if err = json.Unmarshal(data, &logMsg); err != nil {
log.Printf("Failed to decode log message from ingest for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to decode log message from ingest for room", "room", room.Name, "err", err)
return
}
// TODO: Handle log message sending to metrics server
@@ -192,63 +164,52 @@ func ingestHandler(room *Room) {
// Metrics callback
room.WebSocket.RegisterMessageCallback("metrics", func(data []byte) {
var metricsMsg MessageMetrics
var metricsMsg connections.MessageMetrics
if err = json.Unmarshal(data, &metricsMsg); err != nil {
log.Printf("Failed to decode metrics message from ingest for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to decode metrics message from ingest for room", "room", room.Name, "err", err)
return
}
// TODO: Handle metrics message sending to metrics server
})
room.WebSocket.RegisterOnClose(func() {
// If PeerConnection is still open, close it
if room.PeerConnection != nil {
if err = room.PeerConnection.Close(); err != nil {
log.Printf("Failed to close PeerConnection for room: '%s' - reason: %s\n", room.Name, err)
}
room.PeerConnection = nil
}
slog.Debug("ingest WebSocket closed for room", "room", room.Name)
room.Online = false
DeleteRoomIfEmpty(room)
room.signalParticipantsOffline()
relay.DeleteRoomIfEmpty(room)
})
log.Printf("Room: '%s' is ready, sending an OK\n", room.Name)
if err = room.WebSocket.SendAnswerMessageWS(AnswerOK); err != nil {
log.Printf("Failed to send OK answer for room: '%s' - reason: %s\n", room.Name, err)
slog.Info("Room is ready, sending OK answer to ingest", "room", room.Name)
if err = room.WebSocket.SendAnswerMessageWS(connections.AnswerOK); err != nil {
slog.Error("Failed to send OK answer message to ingest for room", "room", room.Name, "err", err)
}
}
// SDP offer handler, returns SDP answer
func handleIngestSDP(room *Room, offerMsg MessageSDP) *webrtc.SessionDescription {
func handleIngestSDP(room *Room, offerMsg connections.MessageSDP) *webrtc.SessionDescription {
var err error
// Get SDP offer
sdpOffer := offerMsg.SDP.SDP
// Modify SDP offer to remove opus "sprop-maxcapturerate=24000" (fixes opus bad quality issue, present in GStreamer)
sdpOffer = strings.Replace(sdpOffer, ";sprop-maxcapturerate=24000", "", -1)
// Set new remote description
err = room.PeerConnection.SetRemoteDescription(webrtc.SessionDescription{
Type: webrtc.SDPTypeOffer,
SDP: sdpOffer,
})
if err != nil {
log.Printf("Failed to set remote description for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to set remote description for room", "room", room.Name, "err", err)
return nil
}
// Create SDP answer
answer, err := room.PeerConnection.CreateAnswer(nil)
if err != nil {
log.Printf("Failed to create SDP answer for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to create SDP answer for room", "room", room.Name, "err", err)
return nil
}
// Set local description
err = room.PeerConnection.SetLocalDescription(answer)
if err != nil {
log.Printf("Failed to set local description for room: '%s' - reason: %s\n", room.Name, err)
slog.Error("Failed to set local description for room", "room", room.Name, "err", err)
return nil
}

View File

@@ -1,30 +1,38 @@
package relay
package internal
import (
"fmt"
"github.com/google/uuid"
"github.com/oklog/ulid/v2"
"github.com/pion/webrtc/v4"
"log/slog"
"math/rand"
"relay/internal/common"
"relay/internal/connections"
)
type Participant struct {
ID uuid.UUID //< Internal IDs are useful to keeping unique internal track and not have conflicts later
ID ulid.ULID //< Internal IDs are useful to keeping unique internal track and not have conflicts later
Name string
WebSocket *SafeWebSocket
WebSocket *connections.SafeWebSocket
PeerConnection *webrtc.PeerConnection
DataChannel *NestriDataChannel
DataChannel *connections.NestriDataChannel
}
func NewParticipant(ws *SafeWebSocket) *Participant {
func NewParticipant(ws *connections.SafeWebSocket) *Participant {
id, err := common.NewULID()
if err != nil {
slog.Error("Failed to create ULID for Participant", "err", err)
return nil
}
return &Participant{
ID: uuid.New(),
ID: id,
Name: createRandomName(),
WebSocket: ws,
}
}
func (vw *Participant) addTrack(trackLocal *webrtc.TrackLocal) error {
rtpSender, err := vw.PeerConnection.AddTrack(*trackLocal)
func (p *Participant) addTrack(trackLocal *webrtc.TrackLocalStaticRTP) error {
rtpSender, err := p.PeerConnection.AddTrack(trackLocal)
if err != nil {
return err
}
@@ -41,22 +49,22 @@ func (vw *Participant) addTrack(trackLocal *webrtc.TrackLocal) error {
return nil
}
func (vw *Participant) signalOffer() error {
if vw.PeerConnection == nil {
return fmt.Errorf("peer connection is nil for participant: '%s' - cannot signal offer", vw.ID)
func (p *Participant) signalOffer() error {
if p.PeerConnection == nil {
return fmt.Errorf("peer connection is nil for participant: '%s' - cannot signal offer", p.ID)
}
offer, err := vw.PeerConnection.CreateOffer(nil)
offer, err := p.PeerConnection.CreateOffer(nil)
if err != nil {
return err
}
err = vw.PeerConnection.SetLocalDescription(offer)
err = p.PeerConnection.SetLocalDescription(offer)
if err != nil {
return err
}
return vw.WebSocket.SendSDPMessageWS(offer)
return p.WebSocket.SendSDPMessageWS(offer)
}
var namesFirst = []string{"Happy", "Sad", "Angry", "Calm", "Excited", "Bored", "Confused", "Confident", "Curious", "Depressed", "Disappointed", "Embarrassed", "Energetic", "Fearful", "Frustrated", "Glad", "Guilty", "Hopeful", "Impatient", "Jealous", "Lonely", "Motivated", "Nervous", "Optimistic", "Pessimistic", "Proud", "Relaxed", "Shy", "Stressed", "Surprised", "Tired", "Worried"}

View File

@@ -1,295 +0,0 @@
package relay
import (
"encoding/json"
"fmt"
"net/http"
"sync"
"time"
// "github.com/gorilla/mux"
"github.com/hashicorp/memberlist"
"github.com/pion/webrtc/v4"
)
// PeerInfo represents information about an SFU peer
type PeerInfo struct {
NodeID string `json:"nodeId"`
Zone string `json:"zone"`
PublicIP string `json:"publicIp"`
PrivateIP string `json:"privateIp,omitempty"`
Streams map[string]bool `json:"streams"` // streamID -> isOrigin
}
// StreamInfo tracks a stream's origin and local subscribers
type StreamInfo struct {
ID string
OriginPeerID string
IsLocal bool
Publisher *webrtc.PeerConnection
Subscribers map[string]*webrtc.PeerConnection
InterPeerConn map[string]*webrtc.PeerConnection // connections to other SFU peers
mu sync.RWMutex
}
// DistributedSFU manages streams and peer communication
type DistributedSFU struct {
nodeID string
zone string
publicIP string
privateIP string
streams map[string]*StreamInfo
peers map[string]*PeerInfo
memberlist *memberlist.Memberlist
mu sync.RWMutex
config webrtc.Configuration
}
// NewDistributedSFU creates a new distributed SFU instance
func NewDistributedSFU(nodeID, zone, publicIP, privateIP string, seeds []string) (*DistributedSFU, error) {
sfu := &DistributedSFU{
nodeID: nodeID,
zone: zone,
publicIP: publicIP,
privateIP: privateIP,
streams: make(map[string]*StreamInfo),
peers: make(map[string]*PeerInfo),
config: webrtc.Configuration{
ICEServers: []webrtc.ICEServer{
{URLs: []string{"stun:stun.l.google.com:19302"}},
},
},
}
// Configure memberlist for peer discovery
config := memberlist.DefaultLANConfig()
config.Name = nodeID
config.BindAddr = privateIP
config.AdvertiseAddr = publicIP
// Add delegate for handling peer updates
config.Delegate = &peerDelegate{sfu: sfu}
// Initialize memberlist
list, err := memberlist.Create(config)
if err != nil {
return nil, err
}
// Join the cluster if seeds are provided
if len(seeds) > 0 {
_, err = list.Join(seeds)
if err != nil {
return nil, err
}
}
sfu.memberlist = list
return sfu, nil
}
// peerDelegate implements memberlist.Delegate
type peerDelegate struct {
sfu *DistributedSFU
}
// NodeMeta returns metadata about the current node
func (d *peerDelegate) NodeMeta(limit int) []byte {
meta := PeerInfo{
NodeID: d.sfu.nodeID,
Zone: d.sfu.zone,
PublicIP: d.sfu.publicIP,
PrivateIP: d.sfu.privateIP,
Streams: make(map[string]bool),
}
d.sfu.mu.RLock()
for id, info := range d.sfu.streams {
meta.Streams[id] = info.IsLocal
}
d.sfu.mu.RUnlock()
data, _ := json.Marshal(meta)
return data
}
// NotifyMsg handles peer updates
func (d *peerDelegate) NotifyMsg(msg []byte) {
var peer PeerInfo
if err := json.Unmarshal(msg, &peer); err != nil {
return
}
d.sfu.mu.Lock()
d.sfu.peers[peer.NodeID] = &peer
// Check for new streams we don't have locally
for streamID, isOrigin := range peer.Streams {
if isOrigin {
if _, exists := d.sfu.streams[streamID]; !exists {
// Initialize inter-peer connection for this stream
d.sfu.initInterPeerStream(streamID, peer.NodeID)
}
}
}
d.sfu.mu.Unlock()
}
// initInterPeerStream sets up connection to another SFU for a stream
func (sfu *DistributedSFU) initInterPeerStream(streamID, peerID string) {
stream := &StreamInfo{
ID: streamID,
OriginPeerID: peerID,
IsLocal: false,
Subscribers: make(map[string]*webrtc.PeerConnection),
InterPeerConn: make(map[string]*webrtc.PeerConnection),
}
// Create peer connection to the origin SFU
pc, err := webrtc.NewPeerConnection(sfu.config)
if err != nil {
return
}
stream.InterPeerConn[peerID] = pc
sfu.streams[streamID] = stream
// Setup inter-peer WebRTC connection
go sfu.establishInterPeerConnection(streamID, peerID, pc)
}
// establishInterPeerConnection handles WebRTC signaling between SFU peers
func (sfu *DistributedSFU) establishInterPeerConnection(streamID, peerID string, pc *webrtc.PeerConnection) {
// This would typically involve making an HTTP request to the peer's control endpoint
// to exchange SDP offers/answers and ICE candidates
peerInfo := sfu.peers[peerID]
// Example endpoint URL construction
peerURL := fmt.Sprintf("http://%s:8080/peer/%s/stream/%s",
peerInfo.PublicIP, sfu.nodeID, streamID)
// Handle incoming tracks from peer
pc.OnTrack(func(remoteTrack *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
sfu.mu.RLock()
stream := sfu.streams[streamID]
sfu.mu.RUnlock()
// Forward the track to local subscribers
stream.mu.RLock()
for _, subscriber := range stream.Subscribers {
localTrack, err := webrtc.NewTrackLocalStaticRTP(
remoteTrack.Codec().RTPCodecCapability,
remoteTrack.ID(),
remoteTrack.StreamID(),
)
if err != nil {
continue
}
if _, err := subscriber.AddTrack(localTrack); err != nil {
continue
}
go func() {
for {
packet, _, err := remoteTrack.ReadRTP()
if err != nil {
return
}
if err := localTrack.WriteRTP(packet); err != nil {
return
}
}
}()
}
stream.mu.RUnlock()
})
// Implement SDP exchange with peer
// ... (signaling implementation)
}
// HandleWHIPPublish now includes peer notification
func (sfu *DistributedSFU) HandleWHIPPublish(w http.ResponseWriter, r *http.Request) {
streamID := mux.Vars(r)["streamID"]
// Create stream info
stream := &StreamInfo{
ID: streamID,
IsLocal: true,
Subscribers: make(map[string]*webrtc.PeerConnection),
InterPeerConn: make(map[string]*webrtc.PeerConnection),
}
// ... (rest of WHIP publish logic)
// Notify other peers about the new stream
sfu.broadcastStreamUpdate(streamID, true)
}
// HandleWHEPSubscribe now checks both local and remote streams
func (sfu *DistributedSFU) HandleWHEPSubscribe(w http.ResponseWriter, r *http.Request) {
streamID := mux.Vars(r)["streamID"]
sfu.mu.RLock()
stream, exists := sfu.streams[streamID]
sfu.mu.RUnlock()
if !exists {
// Check if any peer has this stream
if peer := sfu.findStreamPeer(streamID); peer != nil {
// Initialize inter-peer connection if needed
sfu.initInterPeerStream(streamID, peer.NodeID)
} else {
http.Error(w, "Stream not found", http.StatusNotFound)
return
}
}
// ... (rest of WHEP subscribe logic)
}
// findStreamPeer finds the peer that has the origin of a stream
func (sfu *DistributedSFU) findStreamPeer(streamID string) *PeerInfo {
sfu.mu.RLock()
defer sfu.mu.RUnlock()
for _, peer := range sfu.peers {
if isOrigin, exists := peer.Streams[streamID]; exists && isOrigin {
return peer
}
}
return nil
}
func main() {
// Initialize the distributed SFU
sfu, err := NewDistributedSFU(
"sfu-1",
"us-east",
"203.0.113.1",
"10.0.0.1",
[]string{"203.0.113.2:7946", "203.0.113.3:7946"},
)
if err != nil {
panic(err)
}
router := mux.NewRouter()
// Regular WHIP/WHEP endpoints
router.HandleFunc("/whip/{streamID}", sfu.HandleWHIPPublish).Methods("POST")
router.HandleFunc("/whep/{streamID}/{subscriberID}", sfu.HandleWHEPSubscribe).Methods("POST")
// Inter-peer communication endpoint
router.HandleFunc("/peer/{peerID}/stream/{streamID}", sfu.HandlePeerSignaling).Methods("POST")
server := &http.Server{
Addr: ":8080",
Handler: router,
ReadTimeout: 10 * time.Second,
WriteTimeout: 10 * time.Second,
}
server.ListenAndServe()
}

View File

@@ -1,6 +1,6 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.4
// protoc-gen-go v1.36.6
// protoc (unknown)
// source: latency_tracker.proto
@@ -128,27 +128,18 @@ func (x *ProtoLatencyTracker) GetTimestamps() []*ProtoTimestampEntry {
var File_latency_tracker_proto protoreflect.FileDescriptor
var file_latency_tracker_proto_rawDesc = string([]byte{
0x0a, 0x15, 0x6c, 0x61, 0x74, 0x65, 0x6e, 0x63, 0x79, 0x5f, 0x74, 0x72, 0x61, 0x63, 0x6b, 0x65,
0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x05, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1f,
0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f,
0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22,
0x5b, 0x0a, 0x13, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
0x70, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x73, 0x74, 0x61, 0x67, 0x65, 0x18,
0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x73, 0x74, 0x61, 0x67, 0x65, 0x12, 0x2e, 0x0a, 0x04,
0x74, 0x69, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x67, 0x6f, 0x6f,
0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d,
0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x04, 0x74, 0x69, 0x6d, 0x65, 0x22, 0x72, 0x0a, 0x13,
0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4c, 0x61, 0x74, 0x65, 0x6e, 0x63, 0x79, 0x54, 0x72, 0x61, 0x63,
0x6b, 0x65, 0x72, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6e, 0x63, 0x65, 0x5f,
0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6e,
0x63, 0x65, 0x49, 0x64, 0x12, 0x3a, 0x0a, 0x0a, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d,
0x70, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x45,
0x6e, 0x74, 0x72, 0x79, 0x52, 0x0a, 0x74, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x73,
0x42, 0x16, 0x5a, 0x14, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e,
0x61, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
})
const file_latency_tracker_proto_rawDesc = "" +
"\n" +
"\x15latency_tracker.proto\x12\x05proto\x1a\x1fgoogle/protobuf/timestamp.proto\"[\n" +
"\x13ProtoTimestampEntry\x12\x14\n" +
"\x05stage\x18\x01 \x01(\tR\x05stage\x12.\n" +
"\x04time\x18\x02 \x01(\v2\x1a.google.protobuf.TimestampR\x04time\"r\n" +
"\x13ProtoLatencyTracker\x12\x1f\n" +
"\vsequence_id\x18\x01 \x01(\tR\n" +
"sequenceId\x12:\n" +
"\n" +
"timestamps\x18\x02 \x03(\v2\x1a.proto.ProtoTimestampEntryR\n" +
"timestampsB\x16Z\x14relay/internal/protob\x06proto3"
var (
file_latency_tracker_proto_rawDescOnce sync.Once

File diff suppressed because it is too large Load Diff

View File

@@ -1,6 +1,6 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.4
// protoc-gen-go v1.36.6
// protoc (unknown)
// source: messages.proto
@@ -127,28 +127,15 @@ func (x *ProtoMessageInput) GetData() *ProtoInput {
var File_messages_proto protoreflect.FileDescriptor
var file_messages_proto_rawDesc = string([]byte{
0x0a, 0x0e, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
0x12, 0x05, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x0b, 0x74, 0x79, 0x70, 0x65, 0x73, 0x2e, 0x70,
0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x15, 0x6c, 0x61, 0x74, 0x65, 0x6e, 0x63, 0x79, 0x5f, 0x74, 0x72,
0x61, 0x63, 0x6b, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x6b, 0x0a, 0x10, 0x50,
0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x42, 0x61, 0x73, 0x65, 0x12,
0x21, 0x0a, 0x0c, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18,
0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x70, 0x61, 0x79, 0x6c, 0x6f, 0x61, 0x64, 0x54, 0x79,
0x70, 0x65, 0x12, 0x34, 0x0a, 0x07, 0x6c, 0x61, 0x74, 0x65, 0x6e, 0x63, 0x79, 0x18, 0x02, 0x20,
0x01, 0x28, 0x0b, 0x32, 0x1a, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x50, 0x72, 0x6f, 0x74,
0x6f, 0x4c, 0x61, 0x74, 0x65, 0x6e, 0x63, 0x79, 0x54, 0x72, 0x61, 0x63, 0x6b, 0x65, 0x72, 0x52,
0x07, 0x6c, 0x61, 0x74, 0x65, 0x6e, 0x63, 0x79, 0x22, 0x76, 0x0a, 0x11, 0x50, 0x72, 0x6f, 0x74,
0x6f, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x49, 0x6e, 0x70, 0x75, 0x74, 0x12, 0x3a, 0x0a,
0x0c, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x5f, 0x62, 0x61, 0x73, 0x65, 0x18, 0x01, 0x20,
0x01, 0x28, 0x0b, 0x32, 0x17, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x50, 0x72, 0x6f, 0x74,
0x6f, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x42, 0x61, 0x73, 0x65, 0x52, 0x0b, 0x6d, 0x65,
0x73, 0x73, 0x61, 0x67, 0x65, 0x42, 0x61, 0x73, 0x65, 0x12, 0x25, 0x0a, 0x04, 0x64, 0x61, 0x74,
0x61, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x11, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e,
0x50, 0x72, 0x6f, 0x74, 0x6f, 0x49, 0x6e, 0x70, 0x75, 0x74, 0x52, 0x04, 0x64, 0x61, 0x74, 0x61,
0x42, 0x16, 0x5a, 0x14, 0x72, 0x65, 0x6c, 0x61, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e,
0x61, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
})
const file_messages_proto_rawDesc = "" +
"\n" +
"\x0emessages.proto\x12\x05proto\x1a\vtypes.proto\x1a\x15latency_tracker.proto\"k\n" +
"\x10ProtoMessageBase\x12!\n" +
"\fpayload_type\x18\x01 \x01(\tR\vpayloadType\x124\n" +
"\alatency\x18\x02 \x01(\v2\x1a.proto.ProtoLatencyTrackerR\alatency\"v\n" +
"\x11ProtoMessageInput\x12:\n" +
"\fmessage_base\x18\x01 \x01(\v2\x17.proto.ProtoMessageBaseR\vmessageBase\x12%\n" +
"\x04data\x18\x02 \x01(\v2\x11.proto.ProtoInputR\x04dataB\x16Z\x14relay/internal/protob\x06proto3"
var (
file_messages_proto_rawDescOnce sync.Once

View File

@@ -0,0 +1,151 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.6
// protoc (unknown)
// source: state.proto
package proto
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
unsafe "unsafe"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
// EntityState represents the state of an entity in the mesh (e.g., a room).
type EntityState struct {
state protoimpl.MessageState `protogen:"open.v1"`
EntityType string `protobuf:"bytes,1,opt,name=entity_type,json=entityType,proto3" json:"entity_type,omitempty"` // Type of entity (e.g., "room")
EntityId string `protobuf:"bytes,2,opt,name=entity_id,json=entityId,proto3" json:"entity_id,omitempty"` // Unique identifier (e.g., room name)
Active bool `protobuf:"varint,3,opt,name=active,proto3" json:"active,omitempty"` // Whether the entity is active
OwnerRelayId string `protobuf:"bytes,4,opt,name=owner_relay_id,json=ownerRelayId,proto3" json:"owner_relay_id,omitempty"` // Relay ID that owns this entity
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *EntityState) Reset() {
*x = EntityState{}
mi := &file_state_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *EntityState) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*EntityState) ProtoMessage() {}
func (x *EntityState) ProtoReflect() protoreflect.Message {
mi := &file_state_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use EntityState.ProtoReflect.Descriptor instead.
func (*EntityState) Descriptor() ([]byte, []int) {
return file_state_proto_rawDescGZIP(), []int{0}
}
func (x *EntityState) GetEntityType() string {
if x != nil {
return x.EntityType
}
return ""
}
func (x *EntityState) GetEntityId() string {
if x != nil {
return x.EntityId
}
return ""
}
func (x *EntityState) GetActive() bool {
if x != nil {
return x.Active
}
return false
}
func (x *EntityState) GetOwnerRelayId() string {
if x != nil {
return x.OwnerRelayId
}
return ""
}
var File_state_proto protoreflect.FileDescriptor
const file_state_proto_rawDesc = "" +
"\n" +
"\vstate.proto\x12\x05proto\"\x89\x01\n" +
"\vEntityState\x12\x1f\n" +
"\ventity_type\x18\x01 \x01(\tR\n" +
"entityType\x12\x1b\n" +
"\tentity_id\x18\x02 \x01(\tR\bentityId\x12\x16\n" +
"\x06active\x18\x03 \x01(\bR\x06active\x12$\n" +
"\x0eowner_relay_id\x18\x04 \x01(\tR\fownerRelayIdB\x16Z\x14relay/internal/protob\x06proto3"
var (
file_state_proto_rawDescOnce sync.Once
file_state_proto_rawDescData []byte
)
func file_state_proto_rawDescGZIP() []byte {
file_state_proto_rawDescOnce.Do(func() {
file_state_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_state_proto_rawDesc), len(file_state_proto_rawDesc)))
})
return file_state_proto_rawDescData
}
var file_state_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
var file_state_proto_goTypes = []any{
(*EntityState)(nil), // 0: proto.EntityState
}
var file_state_proto_depIdxs = []int32{
0, // [0:0] is the sub-list for method output_type
0, // [0:0] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name
0, // [0:0] is the sub-list for extension extendee
0, // [0:0] is the sub-list for field type_name
}
func init() { file_state_proto_init() }
func file_state_proto_init() {
if File_state_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_state_proto_rawDesc), len(file_state_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_state_proto_goTypes,
DependencyIndexes: file_state_proto_depIdxs,
MessageInfos: file_state_proto_msgTypes,
}.Build()
File_state_proto = out.File
file_state_proto_goTypes = nil
file_state_proto_depIdxs = nil
}

View File

@@ -1,6 +1,6 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.4
// protoc-gen-go v1.36.6
// protoc (unknown)
// source: types.proto
@@ -581,65 +581,48 @@ func (*ProtoInput_KeyUp) isProtoInput_InputType() {}
var File_types_proto protoreflect.FileDescriptor
var file_types_proto_rawDesc = string([]byte{
0x0a, 0x0b, 0x74, 0x79, 0x70, 0x65, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x05, 0x70,
0x72, 0x6f, 0x74, 0x6f, 0x22, 0x40, 0x0a, 0x0e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x6f, 0x75,
0x73, 0x65, 0x4d, 0x6f, 0x76, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01,
0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x0c, 0x0a, 0x01, 0x78, 0x18,
0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x01, 0x78, 0x12, 0x0c, 0x0a, 0x01, 0x79, 0x18, 0x03, 0x20,
0x01, 0x28, 0x05, 0x52, 0x01, 0x79, 0x22, 0x43, 0x0a, 0x11, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4d,
0x6f, 0x75, 0x73, 0x65, 0x4d, 0x6f, 0x76, 0x65, 0x41, 0x62, 0x73, 0x12, 0x12, 0x0a, 0x04, 0x74,
0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12,
0x0c, 0x0a, 0x01, 0x78, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x01, 0x78, 0x12, 0x0c, 0x0a,
0x01, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x01, 0x79, 0x22, 0x41, 0x0a, 0x0f, 0x50,
0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x6f, 0x75, 0x73, 0x65, 0x57, 0x68, 0x65, 0x65, 0x6c, 0x12, 0x12,
0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79,
0x70, 0x65, 0x12, 0x0c, 0x0a, 0x01, 0x78, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x01, 0x78,
0x12, 0x0c, 0x0a, 0x01, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x05, 0x52, 0x01, 0x79, 0x22, 0x39,
0x0a, 0x11, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x6f, 0x75, 0x73, 0x65, 0x4b, 0x65, 0x79, 0x44,
0x6f, 0x77, 0x6e, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28,
0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x02,
0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0x37, 0x0a, 0x0f, 0x50, 0x72, 0x6f,
0x74, 0x6f, 0x4d, 0x6f, 0x75, 0x73, 0x65, 0x4b, 0x65, 0x79, 0x55, 0x70, 0x12, 0x12, 0x0a, 0x04,
0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65,
0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x6b,
0x65, 0x79, 0x22, 0x34, 0x0a, 0x0c, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4b, 0x65, 0x79, 0x44, 0x6f,
0x77, 0x6e, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x02, 0x20,
0x01, 0x28, 0x05, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0x32, 0x0a, 0x0a, 0x50, 0x72, 0x6f, 0x74,
0x6f, 0x4b, 0x65, 0x79, 0x55, 0x70, 0x12, 0x12, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01,
0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65,
0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x22, 0xab, 0x03, 0x0a,
0x0a, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x49, 0x6e, 0x70, 0x75, 0x74, 0x12, 0x36, 0x0a, 0x0a, 0x6d,
0x6f, 0x75, 0x73, 0x65, 0x5f, 0x6d, 0x6f, 0x76, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32,
0x15, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x6f, 0x75,
0x73, 0x65, 0x4d, 0x6f, 0x76, 0x65, 0x48, 0x00, 0x52, 0x09, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x4d,
0x6f, 0x76, 0x65, 0x12, 0x40, 0x0a, 0x0e, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x5f, 0x6d, 0x6f, 0x76,
0x65, 0x5f, 0x61, 0x62, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x18, 0x2e, 0x70, 0x72,
0x6f, 0x74, 0x6f, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x6f, 0x75, 0x73, 0x65, 0x4d, 0x6f,
0x76, 0x65, 0x41, 0x62, 0x73, 0x48, 0x00, 0x52, 0x0c, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x4d, 0x6f,
0x76, 0x65, 0x41, 0x62, 0x73, 0x12, 0x39, 0x0a, 0x0b, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x5f, 0x77,
0x68, 0x65, 0x65, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x70, 0x72, 0x6f,
0x74, 0x6f, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x6f, 0x75, 0x73, 0x65, 0x57, 0x68, 0x65,
0x65, 0x6c, 0x48, 0x00, 0x52, 0x0a, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x57, 0x68, 0x65, 0x65, 0x6c,
0x12, 0x40, 0x0a, 0x0e, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x64, 0x6f,
0x77, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x18, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x6f, 0x75, 0x73, 0x65, 0x4b, 0x65, 0x79, 0x44, 0x6f,
0x77, 0x6e, 0x48, 0x00, 0x52, 0x0c, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x4b, 0x65, 0x79, 0x44, 0x6f,
0x77, 0x6e, 0x12, 0x3a, 0x0a, 0x0c, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x5f,
0x75, 0x70, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4d, 0x6f, 0x75, 0x73, 0x65, 0x4b, 0x65, 0x79, 0x55, 0x70,
0x48, 0x00, 0x52, 0x0a, 0x6d, 0x6f, 0x75, 0x73, 0x65, 0x4b, 0x65, 0x79, 0x55, 0x70, 0x12, 0x30,
0x0a, 0x08, 0x6b, 0x65, 0x79, 0x5f, 0x64, 0x6f, 0x77, 0x6e, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b,
0x32, 0x13, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4b, 0x65,
0x79, 0x44, 0x6f, 0x77, 0x6e, 0x48, 0x00, 0x52, 0x07, 0x6b, 0x65, 0x79, 0x44, 0x6f, 0x77, 0x6e,
0x12, 0x2a, 0x0a, 0x06, 0x6b, 0x65, 0x79, 0x5f, 0x75, 0x70, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b,
0x32, 0x11, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x4b, 0x65,
0x79, 0x55, 0x70, 0x48, 0x00, 0x52, 0x05, 0x6b, 0x65, 0x79, 0x55, 0x70, 0x42, 0x0c, 0x0a, 0x0a,
0x69, 0x6e, 0x70, 0x75, 0x74, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x42, 0x16, 0x5a, 0x14, 0x72, 0x65,
0x6c, 0x61, 0x79, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x70, 0x72, 0x6f,
0x74, 0x6f, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
})
const file_types_proto_rawDesc = "" +
"\n" +
"\vtypes.proto\x12\x05proto\"@\n" +
"\x0eProtoMouseMove\x12\x12\n" +
"\x04type\x18\x01 \x01(\tR\x04type\x12\f\n" +
"\x01x\x18\x02 \x01(\x05R\x01x\x12\f\n" +
"\x01y\x18\x03 \x01(\x05R\x01y\"C\n" +
"\x11ProtoMouseMoveAbs\x12\x12\n" +
"\x04type\x18\x01 \x01(\tR\x04type\x12\f\n" +
"\x01x\x18\x02 \x01(\x05R\x01x\x12\f\n" +
"\x01y\x18\x03 \x01(\x05R\x01y\"A\n" +
"\x0fProtoMouseWheel\x12\x12\n" +
"\x04type\x18\x01 \x01(\tR\x04type\x12\f\n" +
"\x01x\x18\x02 \x01(\x05R\x01x\x12\f\n" +
"\x01y\x18\x03 \x01(\x05R\x01y\"9\n" +
"\x11ProtoMouseKeyDown\x12\x12\n" +
"\x04type\x18\x01 \x01(\tR\x04type\x12\x10\n" +
"\x03key\x18\x02 \x01(\x05R\x03key\"7\n" +
"\x0fProtoMouseKeyUp\x12\x12\n" +
"\x04type\x18\x01 \x01(\tR\x04type\x12\x10\n" +
"\x03key\x18\x02 \x01(\x05R\x03key\"4\n" +
"\fProtoKeyDown\x12\x12\n" +
"\x04type\x18\x01 \x01(\tR\x04type\x12\x10\n" +
"\x03key\x18\x02 \x01(\x05R\x03key\"2\n" +
"\n" +
"ProtoKeyUp\x12\x12\n" +
"\x04type\x18\x01 \x01(\tR\x04type\x12\x10\n" +
"\x03key\x18\x02 \x01(\x05R\x03key\"\xab\x03\n" +
"\n" +
"ProtoInput\x126\n" +
"\n" +
"mouse_move\x18\x01 \x01(\v2\x15.proto.ProtoMouseMoveH\x00R\tmouseMove\x12@\n" +
"\x0emouse_move_abs\x18\x02 \x01(\v2\x18.proto.ProtoMouseMoveAbsH\x00R\fmouseMoveAbs\x129\n" +
"\vmouse_wheel\x18\x03 \x01(\v2\x16.proto.ProtoMouseWheelH\x00R\n" +
"mouseWheel\x12@\n" +
"\x0emouse_key_down\x18\x04 \x01(\v2\x18.proto.ProtoMouseKeyDownH\x00R\fmouseKeyDown\x12:\n" +
"\fmouse_key_up\x18\x05 \x01(\v2\x16.proto.ProtoMouseKeyUpH\x00R\n" +
"mouseKeyUp\x120\n" +
"\bkey_down\x18\x06 \x01(\v2\x13.proto.ProtoKeyDownH\x00R\akeyDown\x12*\n" +
"\x06key_up\x18\a \x01(\v2\x11.proto.ProtoKeyUpH\x00R\x05keyUpB\f\n" +
"\n" +
"input_typeB\x16Z\x14relay/internal/protob\x06proto3"
var (
file_types_proto_rawDescOnce sync.Once

View File

@@ -0,0 +1,154 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.6
// protoc (unknown)
// source: webrtc.proto
package proto
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
unsafe "unsafe"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
type ICECandidateInit struct {
state protoimpl.MessageState `protogen:"open.v1"`
Candidate string `protobuf:"bytes,1,opt,name=candidate,proto3" json:"candidate,omitempty"`
SdpMid *string `protobuf:"bytes,2,opt,name=sdp_mid,json=sdpMid,proto3,oneof" json:"sdp_mid,omitempty"`
SdpMLineIndex *uint32 `protobuf:"varint,3,opt,name=sdp_m_line_index,json=sdpMLineIndex,proto3,oneof" json:"sdp_m_line_index,omitempty"`
UsernameFragment *string `protobuf:"bytes,4,opt,name=username_fragment,json=usernameFragment,proto3,oneof" json:"username_fragment,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ICECandidateInit) Reset() {
*x = ICECandidateInit{}
mi := &file_webrtc_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ICECandidateInit) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ICECandidateInit) ProtoMessage() {}
func (x *ICECandidateInit) ProtoReflect() protoreflect.Message {
mi := &file_webrtc_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use ICECandidateInit.ProtoReflect.Descriptor instead.
func (*ICECandidateInit) Descriptor() ([]byte, []int) {
return file_webrtc_proto_rawDescGZIP(), []int{0}
}
func (x *ICECandidateInit) GetCandidate() string {
if x != nil {
return x.Candidate
}
return ""
}
func (x *ICECandidateInit) GetSdpMid() string {
if x != nil && x.SdpMid != nil {
return *x.SdpMid
}
return ""
}
func (x *ICECandidateInit) GetSdpMLineIndex() uint32 {
if x != nil && x.SdpMLineIndex != nil {
return *x.SdpMLineIndex
}
return 0
}
func (x *ICECandidateInit) GetUsernameFragment() string {
if x != nil && x.UsernameFragment != nil {
return *x.UsernameFragment
}
return ""
}
var File_webrtc_proto protoreflect.FileDescriptor
const file_webrtc_proto_rawDesc = "" +
"\n" +
"\fwebrtc.proto\x12\x05proto\"\xe5\x01\n" +
"\x10ICECandidateInit\x12\x1c\n" +
"\tcandidate\x18\x01 \x01(\tR\tcandidate\x12\x1c\n" +
"\asdp_mid\x18\x02 \x01(\tH\x00R\x06sdpMid\x88\x01\x01\x12,\n" +
"\x10sdp_m_line_index\x18\x03 \x01(\rH\x01R\rsdpMLineIndex\x88\x01\x01\x120\n" +
"\x11username_fragment\x18\x04 \x01(\tH\x02R\x10usernameFragment\x88\x01\x01B\n" +
"\n" +
"\b_sdp_midB\x13\n" +
"\x11_sdp_m_line_indexB\x14\n" +
"\x12_username_fragmentB\x16Z\x14relay/internal/protob\x06proto3"
var (
file_webrtc_proto_rawDescOnce sync.Once
file_webrtc_proto_rawDescData []byte
)
func file_webrtc_proto_rawDescGZIP() []byte {
file_webrtc_proto_rawDescOnce.Do(func() {
file_webrtc_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_webrtc_proto_rawDesc), len(file_webrtc_proto_rawDesc)))
})
return file_webrtc_proto_rawDescData
}
var file_webrtc_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
var file_webrtc_proto_goTypes = []any{
(*ICECandidateInit)(nil), // 0: proto.ICECandidateInit
}
var file_webrtc_proto_depIdxs = []int32{
0, // [0:0] is the sub-list for method output_type
0, // [0:0] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name
0, // [0:0] is the sub-list for extension extendee
0, // [0:0] is the sub-list for field type_name
}
func init() { file_webrtc_proto_init() }
func file_webrtc_proto_init() {
if File_webrtc_proto != nil {
return
}
file_webrtc_proto_msgTypes[0].OneofWrappers = []any{}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_webrtc_proto_rawDesc), len(file_webrtc_proto_rawDesc)),
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 0,
},
GoTypes: file_webrtc_proto_goTypes,
DependencyIndexes: file_webrtc_proto_depIdxs,
MessageInfos: file_webrtc_proto_msgTypes,
}.Build()
File_webrtc_proto = out.File
file_webrtc_proto_goTypes = nil
file_webrtc_proto_depIdxs = nil
}

View File

@@ -0,0 +1,702 @@
package internal
import (
"context"
"crypto/sha256"
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"github.com/libp2p/go-libp2p"
"github.com/libp2p/go-libp2p-pubsub"
"github.com/libp2p/go-libp2p/core/host"
"github.com/libp2p/go-libp2p/core/network"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/libp2p/go-libp2p/core/pnet"
"github.com/libp2p/go-libp2p/p2p/security/noise"
"github.com/multiformats/go-multiaddr"
"github.com/oklog/ulid/v2"
"github.com/pion/webrtc/v4"
"io"
"log/slog"
"relay/internal/common"
"relay/internal/connections"
)
var globalRelay *Relay
// networkNotifier logs connection events
type networkNotifier struct{}
func (n *networkNotifier) Connected(net network.Network, conn network.Conn) {
slog.Info("Peer connected", "local", conn.LocalPeer(), "remote", conn.RemotePeer())
}
func (n *networkNotifier) Disconnected(net network.Network, conn network.Conn) {
slog.Info("Peer disconnected", "local", conn.LocalPeer(), "remote", conn.RemotePeer())
}
func (n *networkNotifier) Listen(net network.Network, addr multiaddr.Multiaddr) {}
func (n *networkNotifier) ListenClose(net network.Network, addr multiaddr.Multiaddr) {}
type ICEMessage struct {
PeerID string
TargetID string
RoomID ulid.ULID
Candidate []byte
}
type Relay struct {
ID peer.ID
Rooms *common.SafeMap[ulid.ULID, *Room]
Host host.Host // libp2p host for peer-to-peer networking
PubSub *pubsub.PubSub // PubSub for state synchronization
MeshState *common.SafeMap[ulid.ULID, RoomInfo] // room ID -> state
RelayPCs *common.SafeMap[ulid.ULID, *webrtc.PeerConnection] // room ID -> relay PeerConnection
pubTopicState *pubsub.Topic // topic for room states
pubTopicICECandidate *pubsub.Topic // topic for ICE candidates aimed to this relay
}
func NewRelay(ctx context.Context, port int) (*Relay, error) {
listenAddrs := []string{
fmt.Sprintf("/ip4/0.0.0.0/tcp/%d", port), // IPv4
fmt.Sprintf("/ip6/::/tcp/%d", port), // IPv6
}
// Use "testToken" as the pre-shared token for authentication
// TODO: Give via flags, before PR commit
token := "testToken"
// Generate 32-byte PSK from the token using SHA-256
shaToken := sha256.Sum256([]byte(token))
tokenPSK := pnet.PSK(shaToken[:])
// Initialize libp2p host
p2pHost, err := libp2p.New(
libp2p.ListenAddrStrings(listenAddrs...),
libp2p.Security(noise.ID, noise.New),
libp2p.EnableRelay(),
libp2p.EnableHolePunching(),
libp2p.PrivateNetwork(tokenPSK),
)
if err != nil {
return nil, fmt.Errorf("failed to create libp2p host for relay: %w", err)
}
// Set up pubsub
p2pPubsub, err := pubsub.NewGossipSub(ctx, p2pHost)
if err != nil {
return nil, fmt.Errorf("failed to create pubsub: %w", err)
}
// Add network notifier to log connections
p2pHost.Network().Notify(&networkNotifier{})
r := &Relay{
ID: p2pHost.ID(),
Host: p2pHost,
PubSub: p2pPubsub,
Rooms: common.NewSafeMap[ulid.ULID, *Room](),
MeshState: common.NewSafeMap[ulid.ULID, RoomInfo](),
RelayPCs: common.NewSafeMap[ulid.ULID, *webrtc.PeerConnection](),
}
// Set up state synchronization and stream handling
r.setupStateSync(ctx)
r.setupStreamHandler()
slog.Info("Relay initialized", "id", r.ID, "addrs", p2pHost.Addrs())
peerInfo := peer.AddrInfo{
ID: p2pHost.ID(),
Addrs: p2pHost.Addrs(),
}
addrs, err := peer.AddrInfoToP2pAddrs(&peerInfo)
if err != nil {
return nil, fmt.Errorf("failed to convert peer info to addresses: %w", err)
}
slog.Debug("Connect with one of the following URLs below:")
for _, addr := range addrs {
slog.Debug(fmt.Sprintf("- %s", addr.String()))
}
return r, nil
}
func InitRelay(ctx context.Context, ctxCancel context.CancelFunc, port int) error {
var err error
globalRelay, err = NewRelay(ctx, port)
if err != nil {
return fmt.Errorf("failed to create relay: %w", err)
}
if err := common.InitWebRTCAPI(); err != nil {
return err
}
if err := InitHTTPEndpoint(ctx, ctxCancel); err != nil {
return err
}
slog.Info("Relay initialized", "id", globalRelay.ID)
return nil
}
func GetRelay() *Relay {
return globalRelay
}
func (r *Relay) GetRoomByID(id ulid.ULID) *Room {
if room, ok := r.Rooms.Get(id); ok {
return room
}
return nil
}
func (r *Relay) GetOrCreateRoom(name string) *Room {
if room := r.GetRoomByName(name); room != nil {
return room
}
id, err := common.NewULID()
if err != nil {
slog.Error("Failed to generate new ULID for room", "err", err)
return nil
}
room := NewRoom(name, id, r.ID)
room.Relay = r
r.Rooms.Set(room.ID, room)
slog.Debug("Created new room", "name", name, "id", room.ID)
return room
}
func (r *Relay) DeleteRoomIfEmpty(room *Room) {
participantCount := room.Participants.Len()
if participantCount > 0 {
slog.Debug("Room not empty, not deleting", "name", room.Name, "id", room.ID, "participants", participantCount)
return
}
// Create a "tombstone" state for the room, this allows propagation of the room deletion
tombstoneState := RoomInfo{
ID: room.ID,
Name: room.Name,
Online: false,
OwnerID: room.OwnerID,
}
// Publish updated state to mesh
if err := r.publishRoomState(context.Background(), tombstoneState); err != nil {
slog.Error("Failed to publish room states on change", "room", room.Name, "err", err)
}
slog.Info("Deleting room since empty and offline", "name", room.Name, "id", room.ID)
r.Rooms.Delete(room.ID)
}
func (r *Relay) setupStateSync(ctx context.Context) {
var err error
r.pubTopicState, err = r.PubSub.Join("room-states")
if err != nil {
slog.Error("Failed to join pubsub topic", "err", err)
return
}
sub, err := r.pubTopicState.Subscribe()
if err != nil {
slog.Error("Failed to subscribe to topic", "err", err)
return
}
r.pubTopicICECandidate, err = r.PubSub.Join("ice-candidates")
if err != nil {
slog.Error("Failed to join ICE candidates topic", "err", err)
return
}
iceCandidateSub, err := r.pubTopicICECandidate.Subscribe()
if err != nil {
slog.Error("Failed to subscribe to ICE candidates topic", "err", err)
return
}
// Handle state updates only from authenticated peers
go func() {
for {
msg, err := sub.Next(ctx)
if err != nil {
slog.Error("Error receiving pubsub message", "err", err)
return
}
if msg.GetFrom() == r.Host.ID() {
continue // Ignore own messages
}
var states []RoomInfo
if err := json.Unmarshal(msg.Data, &states); err != nil {
slog.Error("Failed to unmarshal room states", "err", err)
continue
}
r.updateMeshState(states)
}
}()
// Handle incoming ICE candidates for given room
go func() {
// Map of ICE candidate slices per room ID
iceHolder := make(map[ulid.ULID][]webrtc.ICECandidateInit)
for {
msg, err := iceCandidateSub.Next(ctx)
if err != nil {
slog.Error("Error receiving ICE candidate message", "err", err)
return
}
if msg.GetFrom() == r.Host.ID() {
continue // Ignore own messages
}
var iceMsg ICEMessage
if err := json.Unmarshal(msg.Data, &iceMsg); err != nil {
slog.Error("Failed to unmarshal ICE candidate message", "err", err)
continue
}
if iceMsg.TargetID != r.ID.String() {
continue // Ignore messages not meant for this relay
}
if iceHolder[iceMsg.RoomID] == nil {
iceHolder[iceMsg.RoomID] = make([]webrtc.ICECandidateInit, 0)
}
if pc, ok := r.RelayPCs.Get(iceMsg.RoomID); ok {
// Unmarshal ice candidate
var candidate webrtc.ICECandidateInit
if err := json.Unmarshal(iceMsg.Candidate, &candidate); err != nil {
slog.Error("Failed to unmarshal ICE candidate", "err", err)
continue
}
if pc.RemoteDescription() != nil {
if err := pc.AddICECandidate(candidate); err != nil {
slog.Error("Failed to add ICE candidate", "err", err)
}
// Add any held candidates
for _, heldCandidate := range iceHolder[iceMsg.RoomID] {
if err := pc.AddICECandidate(heldCandidate); err != nil {
slog.Error("Failed to add held ICE candidate", "err", err)
}
}
iceHolder[iceMsg.RoomID] = make([]webrtc.ICECandidateInit, 0)
} else {
iceHolder[iceMsg.RoomID] = append(iceHolder[iceMsg.RoomID], candidate)
}
} else {
slog.Error("PeerConnection for room not found when adding ICE candidate", "roomID", iceMsg.RoomID)
}
}
}()
}
func (r *Relay) publishRoomState(ctx context.Context, state RoomInfo) error {
data, err := json.Marshal([]RoomInfo{state})
if err != nil {
return err
}
return r.pubTopicState.Publish(ctx, data)
}
func (r *Relay) publishRoomStates(ctx context.Context) error {
var states []RoomInfo
for _, room := range r.Rooms.Copy() {
states = append(states, RoomInfo{
ID: room.ID,
Name: room.Name,
Online: room.Online,
OwnerID: r.ID,
})
}
data, err := json.Marshal(states)
if err != nil {
return err
}
return r.pubTopicState.Publish(ctx, data)
}
func (r *Relay) updateMeshState(states []RoomInfo) {
for _, state := range states {
if state.OwnerID == r.ID {
continue // Skip own state
}
existing, exists := r.MeshState.Get(state.ID)
r.MeshState.Set(state.ID, state)
slog.Debug("Updated mesh state", "room", state.Name, "online", state.Online, "owner", state.OwnerID)
// React to state changes
if !exists || existing.Online != state.Online {
room := r.GetRoomByName(state.Name)
if state.Online {
if room == nil || !room.Online {
slog.Info("Room became active remotely, requesting stream", "room", state.Name, "owner", state.OwnerID)
go func() {
if _, err := r.requestStream(context.Background(), state.Name, state.ID, state.OwnerID); err != nil {
slog.Error("Failed to request stream", "room", state.Name, "err", err)
} else {
slog.Info("Successfully requested stream", "room", state.Name, "owner", state.OwnerID)
}
}()
}
} else if room != nil && room.Online {
slog.Info("Room became inactive remotely, stopping local stream", "room", state.Name)
if pc, ok := r.RelayPCs.Get(state.ID); ok {
_ = pc.Close()
r.RelayPCs.Delete(state.ID)
}
room.Online = false
room.signalParticipantsOffline()
} else if room == nil && !exists {
slog.Info("Received tombstone state for room", "name", state.Name, "id", state.ID)
if pc, ok := r.RelayPCs.Get(state.ID); ok {
_ = pc.Close()
r.RelayPCs.Delete(state.ID)
}
}
}
}
}
func (r *Relay) IsRoomActive(roomID ulid.ULID) (bool, peer.ID) {
if state, exists := r.MeshState.Get(roomID); exists && state.Online {
return true, state.OwnerID
}
return false, ""
}
func (r *Relay) GetRoomByName(name string) *Room {
for _, room := range r.Rooms.Copy() {
if room.Name == name {
return room
}
}
return nil
}
func writeMessage(stream network.Stream, data []byte) error {
length := uint32(len(data))
if err := binary.Write(stream, binary.BigEndian, length); err != nil {
return err
}
_, err := stream.Write(data)
return err
}
func readMessage(stream network.Stream) ([]byte, error) {
var length uint32
if err := binary.Read(stream, binary.BigEndian, &length); err != nil {
return nil, err
}
data := make([]byte, length)
_, err := io.ReadFull(stream, data)
return data, err
}
func (r *Relay) setupStreamHandler() {
r.Host.SetStreamHandler("/nestri-relay/stream/1.0.0", func(stream network.Stream) {
defer func(stream network.Stream) {
err := stream.Close()
if err != nil {
slog.Error("Failed to close stream", "err", err)
}
}(stream)
remotePeer := stream.Conn().RemotePeer()
roomNameData, err := readMessage(stream)
if err != nil && err != io.EOF {
slog.Error("Failed to read room name", "peer", remotePeer, "err", err)
return
}
roomName := string(roomNameData)
slog.Info("Stream request from peer", "peer", remotePeer, "room", roomName)
room := r.GetRoomByName(roomName)
if room == nil || !room.Online {
slog.Error("Cannot provide stream for inactive room", "room", roomName)
return
}
pc, err := common.CreatePeerConnection(func() {
r.RelayPCs.Delete(room.ID)
})
if err != nil {
slog.Error("Failed to create relay PeerConnection", "err", err)
return
}
r.RelayPCs.Set(room.ID, pc)
if room.AudioTrack != nil {
_, err := pc.AddTrack(room.AudioTrack)
if err != nil {
slog.Error("Failed to add audio track", "err", err)
return
}
}
if room.VideoTrack != nil {
_, err := pc.AddTrack(room.VideoTrack)
if err != nil {
slog.Error("Failed to add video track", "err", err)
return
}
}
settingOrdered := true
settingMaxRetransmits := uint16(0)
dc, err := pc.CreateDataChannel("relay-data", &webrtc.DataChannelInit{
Ordered: &settingOrdered,
MaxRetransmits: &settingMaxRetransmits,
})
if err != nil {
slog.Error("Failed to create relay DataChannel", "err", err)
return
}
relayDC := connections.NewNestriDataChannel(dc)
relayDC.RegisterOnOpen(func() {
slog.Debug("Relay DataChannel opened", "room", roomName)
})
relayDC.RegisterOnClose(func() {
slog.Debug("Relay DataChannel closed", "room", roomName)
})
relayDC.RegisterMessageCallback("input", func(data []byte) {
if room.DataChannel != nil {
// Forward message to the room's data channel
if err := room.DataChannel.SendBinary(data); err != nil {
slog.Error("Failed to send DataChannel message", "room", roomName, "err", err)
}
}
})
offer, err := pc.CreateOffer(nil)
if err != nil {
slog.Error("Failed to create offer", "err", err)
return
}
if err := pc.SetLocalDescription(offer); err != nil {
slog.Error("Failed to set local description", "err", err)
return
}
offerData, err := json.Marshal(offer)
if err != nil {
slog.Error("Failed to marshal offer", "err", err)
return
}
if err := writeMessage(stream, offerData); err != nil {
slog.Error("Failed to send offer", "peer", remotePeer, "err", err)
return
}
// Handle our generated ICE candidates
pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
if candidate == nil {
return
}
candidateData, err := json.Marshal(candidate.ToJSON())
if err != nil {
slog.Error("Failed to marshal ICE candidate", "err", err)
return
}
iceMsg := ICEMessage{
PeerID: r.Host.ID().String(),
TargetID: remotePeer.String(),
RoomID: room.ID,
Candidate: candidateData,
}
data, err := json.Marshal(iceMsg)
if err != nil {
slog.Error("Failed to marshal ICE message", "err", err)
return
}
if err := r.pubTopicICECandidate.Publish(context.Background(), data); err != nil {
slog.Error("Failed to publish ICE candidate message", "err", err)
}
})
answerData, err := readMessage(stream)
if err != nil && err != io.EOF {
slog.Error("Failed to read answer", "peer", remotePeer, "err", err)
return
}
var answer webrtc.SessionDescription
if err := json.Unmarshal(answerData, &answer); err != nil {
slog.Error("Failed to unmarshal answer", "err", err)
return
}
if err := pc.SetRemoteDescription(answer); err != nil {
slog.Error("Failed to set remote description", "err", err)
return
}
})
}
func (r *Relay) requestStream(ctx context.Context, roomName string, roomID ulid.ULID, providerPeer peer.ID) (*webrtc.PeerConnection, error) {
stream, err := r.Host.NewStream(ctx, providerPeer, "/nestri-relay/stream/1.0.0")
if err != nil {
return nil, fmt.Errorf("failed to create stream: %w", err)
}
defer func(stream network.Stream) {
err := stream.Close()
if err != nil {
slog.Error("Failed to close stream", "err", err)
}
}(stream)
if err := writeMessage(stream, []byte(roomName)); err != nil {
return nil, fmt.Errorf("failed to send room name: %w", err)
}
room := r.GetRoomByName(roomName)
if room == nil {
room = NewRoom(roomName, roomID, providerPeer)
r.Rooms.Set(roomID, room)
} else if room.ID != roomID {
// Mismatch, prefer the one from the provider
// TODO: When mesh is created, if there are mismatches, we should have relays negotiate common room IDs
room.ID = roomID
room.OwnerID = providerPeer
r.Rooms.Set(roomID, room)
}
pc, err := common.CreatePeerConnection(func() {
r.RelayPCs.Delete(roomID)
})
if err != nil {
return nil, fmt.Errorf("failed to create PeerConnection: %w", err)
}
r.RelayPCs.Set(roomID, pc)
offerData, err := readMessage(stream)
if err != nil && err != io.EOF {
return nil, fmt.Errorf("failed to read offer: %w", err)
}
var offer webrtc.SessionDescription
if err := json.Unmarshal(offerData, &offer); err != nil {
return nil, fmt.Errorf("failed to unmarshal offer: %w", err)
}
if err := pc.SetRemoteDescription(offer); err != nil {
return nil, fmt.Errorf("failed to set remote description: %w", err)
}
pc.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
localTrack, _ := webrtc.NewTrackLocalStaticRTP(track.Codec().RTPCodecCapability, track.ID(), "relay-"+roomName+"-"+track.Kind().String())
slog.Debug("Received track for mesh relay room", "room", roomName, "kind", track.Kind())
room.SetTrack(track.Kind(), localTrack)
go func() {
for {
rtpPacket, _, err := track.ReadRTP()
if err != nil {
if !errors.Is(err, io.EOF) {
slog.Error("Failed to read RTP packet from remote track for room", "room", roomName, "err", err)
}
break
}
err = localTrack.WriteRTP(rtpPacket)
if err != nil && !errors.Is(err, io.ErrClosedPipe) {
slog.Error("Failed to write RTP to local track for room", "room", room.Name, "err", err)
break
}
}
}()
})
// ICE candidate handling
pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
if candidate == nil {
return
}
candidateData, err := json.Marshal(candidate.ToJSON())
if err != nil {
slog.Error("Failed to marshal ICE candidate", "err", err)
return
}
iceMsg := ICEMessage{
PeerID: r.Host.ID().String(),
TargetID: providerPeer.String(),
RoomID: roomID,
Candidate: candidateData,
}
data, err := json.Marshal(iceMsg)
if err != nil {
slog.Error("Failed to marshal ICE message", "err", err)
return
}
if err := r.pubTopicICECandidate.Publish(ctx, data); err != nil {
slog.Error("Failed to publish ICE candidate message", "err", err)
}
})
pc.OnDataChannel(func(dc *webrtc.DataChannel) {
relayDC := connections.NewNestriDataChannel(dc)
slog.Debug("Received DataChannel from peer", "room", roomName)
relayDC.RegisterOnOpen(func() {
slog.Debug("Relay DataChannel opened", "room", roomName)
})
relayDC.OnClose(func() {
slog.Debug("Relay DataChannel closed", "room", roomName)
})
// Override room DataChannel with the mesh-relay one to forward messages
room.DataChannel = relayDC
})
answer, err := pc.CreateAnswer(nil)
if err != nil {
return nil, fmt.Errorf("failed to create answer: %w", err)
}
if err := pc.SetLocalDescription(answer); err != nil {
return nil, fmt.Errorf("failed to set local description: %w", err)
}
answerData, err := json.Marshal(answer)
if err != nil {
return nil, fmt.Errorf("failed to marshal answer: %w", err)
}
if err := writeMessage(stream, answerData); err != nil {
return nil, fmt.Errorf("failed to send answer: %w", err)
}
return pc, nil
}
// ConnectToRelay manually connects to another relay by its multiaddress
func (r *Relay) ConnectToRelay(ctx context.Context, addr string) error {
// Parse the multiaddress
ma, err := multiaddr.NewMultiaddr(addr)
if err != nil {
slog.Error("Invalid multiaddress", "addr", addr, "err", err)
return fmt.Errorf("invalid multiaddress: %w", err)
}
// Extract peer ID from multiaddress
peerInfo, err := peer.AddrInfoFromP2pAddr(ma)
if err != nil {
slog.Error("Failed to extract peer info", "addr", addr, "err", err)
return fmt.Errorf("failed to extract peer info: %w", err)
}
// Connect to the peer
if err := r.Host.Connect(ctx, *peerInfo); err != nil {
slog.Error("Failed to connect to peer", "peer", peerInfo.ID, "addr", addr, "err", err)
return fmt.Errorf("failed to connect: %w", err)
}
// Publish challenge on join
//go r.sendAuthChallenge(ctx)
slog.Info("Successfully connected to peer", "peer", peerInfo.ID, "addr", addr)
return nil
}

View File

@@ -1,179 +1,164 @@
package relay
package internal
import (
"github.com/google/uuid"
"context"
"fmt"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/oklog/ulid/v2"
"github.com/pion/webrtc/v4"
"log"
"sync"
"log/slog"
"relay/internal/common"
"relay/internal/connections"
)
var Rooms = make(map[uuid.UUID]*Room) //< Room ID -> Room
var RoomsMutex = sync.RWMutex{}
func GetRoomByID(id uuid.UUID) *Room {
RoomsMutex.RLock()
defer RoomsMutex.RUnlock()
if room, ok := Rooms[id]; ok {
return room
}
return nil
}
func GetRoomByName(name string) *Room {
RoomsMutex.RLock()
defer RoomsMutex.RUnlock()
for _, room := range Rooms {
if room.Name == name {
return room
}
}
return nil
}
func GetOrCreateRoom(name string) *Room {
if room := GetRoomByName(name); room != nil {
return room
}
RoomsMutex.Lock()
room := NewRoom(name)
Rooms[room.ID] = room
if GetFlags().Verbose {
log.Printf("New room: '%s'\n", room.Name)
}
RoomsMutex.Unlock()
return room
}
func DeleteRoomIfEmpty(room *Room) {
room.ParticipantsMutex.RLock()
defer room.ParticipantsMutex.RUnlock()
if !room.Online && len(room.Participants) <= 0 {
RoomsMutex.Lock()
delete(Rooms, room.ID)
RoomsMutex.Unlock()
}
type RoomInfo struct {
ID ulid.ULID `json:"id"`
Name string `json:"name"`
Online bool `json:"online"`
OwnerID peer.ID `json:"owner_id"`
}
type Room struct {
ID uuid.UUID //< Internal IDs are useful to keeping unique internal track
Name string
Online bool //< Whether the room is currently online, i.e. receiving data from a nestri-server
WebSocket *SafeWebSocket
PeerConnection *webrtc.PeerConnection
AudioTrack webrtc.TrackLocal
VideoTrack webrtc.TrackLocal
DataChannel *NestriDataChannel
Participants map[uuid.UUID]*Participant
ParticipantsMutex sync.RWMutex
RoomInfo
WebSocket *connections.SafeWebSocket
PeerConnection *webrtc.PeerConnection
AudioTrack *webrtc.TrackLocalStaticRTP
VideoTrack *webrtc.TrackLocalStaticRTP
DataChannel *connections.NestriDataChannel
Participants *common.SafeMap[ulid.ULID, *Participant]
Relay *Relay
}
func NewRoom(name string) *Room {
func NewRoom(name string, roomID ulid.ULID, ownerID peer.ID) *Room {
return &Room{
ID: uuid.New(),
Name: name,
Online: false,
Participants: make(map[uuid.UUID]*Participant),
RoomInfo: RoomInfo{
ID: roomID,
Name: name,
Online: false,
OwnerID: ownerID,
},
Participants: common.NewSafeMap[ulid.ULID, *Participant](),
}
}
// Assigns a WebSocket connection to a Room
func (r *Room) assignWebSocket(ws *SafeWebSocket) {
// If WS already assigned, warn
// AssignWebSocket assigns a WebSocket connection to a Room
func (r *Room) AssignWebSocket(ws *connections.SafeWebSocket) {
if r.WebSocket != nil {
log.Printf("Warning: Room '%s' already has a WebSocket assigned\n", r.Name)
slog.Warn("WebSocket already assigned to room", "room", r.Name)
}
r.WebSocket = ws
}
// Adds a Participant to a Room
func (r *Room) addParticipant(participant *Participant) {
r.ParticipantsMutex.Lock()
r.Participants[participant.ID] = participant
r.ParticipantsMutex.Unlock()
// AddParticipant adds a Participant to a Room
func (r *Room) AddParticipant(participant *Participant) {
slog.Debug("Adding participant to room", "participant", participant.ID, "room", r.Name)
r.Participants.Set(participant.ID, participant)
}
// Removes a Participant from a Room by participant's ID.
// If Room is offline and this is the last participant, the room is deleted
func (r *Room) removeParticipantByID(pID uuid.UUID) {
r.ParticipantsMutex.Lock()
delete(r.Participants, pID)
r.ParticipantsMutex.Unlock()
DeleteRoomIfEmpty(r)
// Removes a Participant from a Room by participant's ID
func (r *Room) removeParticipantByID(pID ulid.ULID) {
if _, ok := r.Participants.Get(pID); ok {
r.Participants.Delete(pID)
}
}
// Removes a Participant from a Room by participant's name.
// If Room is offline and this is the last participant, the room is deleted
// Removes a Participant from a Room by participant's name
func (r *Room) removeParticipantByName(pName string) {
r.ParticipantsMutex.Lock()
for id, p := range r.Participants {
if p.Name == pName {
delete(r.Participants, id)
for id, participant := range r.Participants.Copy() {
if participant.Name == pName {
if err := r.signalParticipantOffline(participant); err != nil {
slog.Error("Failed to signal participant offline", "participant", participant.ID, "room", r.Name, "err", err)
}
r.Participants.Delete(id)
break
}
}
r.ParticipantsMutex.Unlock()
DeleteRoomIfEmpty(r)
}
// Signals all participants with offer and add tracks to their PeerConnections
// Removes all participants from a Room
func (r *Room) removeAllParticipants() {
for id, participant := range r.Participants.Copy() {
if err := r.signalParticipantOffline(participant); err != nil {
slog.Error("Failed to signal participant offline", "participant", participant.ID, "room", r.Name, "err", err)
}
r.Participants.Delete(id)
slog.Debug("Removed participant from room", "participant", id, "room", r.Name)
}
}
func (r *Room) SetTrack(trackType webrtc.RTPCodecType, track *webrtc.TrackLocalStaticRTP) {
switch trackType {
case webrtc.RTPCodecTypeAudio:
r.AudioTrack = track
slog.Debug("Audio track set", "room", r.Name, "track", track != nil)
case webrtc.RTPCodecTypeVideo:
r.VideoTrack = track
slog.Debug("Video track set", "room", r.Name, "track", track != nil)
default:
slog.Warn("Unknown track type", "room", r.Name, "trackType", trackType)
}
newOnline := r.AudioTrack != nil && r.VideoTrack != nil
if r.Online != newOnline {
r.Online = newOnline
if r.Online {
slog.Debug("Room online, participants will be signaled", "room", r.Name)
r.signalParticipantsWithTracks()
} else {
slog.Debug("Room offline, signaling participants", "room", r.Name)
r.signalParticipantsOffline()
}
// Publish updated state to mesh
go func() {
if err := r.Relay.publishRoomStates(context.Background()); err != nil {
slog.Error("Failed to publish room states on change", "room", r.Name, "err", err)
}
}()
}
}
func (r *Room) signalParticipantsWithTracks() {
r.ParticipantsMutex.RLock()
for _, participant := range r.Participants {
// Add tracks to participant's PeerConnection
if r.AudioTrack != nil {
if err := participant.addTrack(&r.AudioTrack); err != nil {
log.Printf("Failed to add audio track to participant: '%s' - reason: %s\n", participant.ID, err)
}
}
if r.VideoTrack != nil {
if err := participant.addTrack(&r.VideoTrack); err != nil {
log.Printf("Failed to add video track to participant: '%s' - reason: %s\n", participant.ID, err)
}
}
// Signal participant with offer
if err := participant.signalOffer(); err != nil {
log.Printf("Error signaling participant: %v\n", err)
for _, participant := range r.Participants.Copy() {
if err := r.signalParticipantWithTracks(participant); err != nil {
slog.Error("Failed to signal participant with tracks", "participant", participant.ID, "room", r.Name, "err", err)
}
}
r.ParticipantsMutex.RUnlock()
}
// Signals all participants that the Room is offline
func (r *Room) signalParticipantWithTracks(participant *Participant) error {
if r.AudioTrack != nil {
if err := participant.addTrack(r.AudioTrack); err != nil {
return fmt.Errorf("failed to add audio track: %w", err)
}
}
if r.VideoTrack != nil {
if err := participant.addTrack(r.VideoTrack); err != nil {
return fmt.Errorf("failed to add video track: %w", err)
}
}
if err := participant.signalOffer(); err != nil {
return fmt.Errorf("failed to signal offer: %w", err)
}
return nil
}
func (r *Room) signalParticipantsOffline() {
r.ParticipantsMutex.RLock()
for _, participant := range r.Participants {
if err := participant.WebSocket.SendAnswerMessageWS(AnswerOffline); err != nil {
log.Printf("Failed to send Offline answer for participant: '%s' - reason: %s\n", participant.ID, err)
for _, participant := range r.Participants.Copy() {
if err := r.signalParticipantOffline(participant); err != nil {
slog.Error("Failed to signal participant offline", "participant", participant.ID, "room", r.Name, "err", err)
}
}
r.ParticipantsMutex.RUnlock()
}
// Broadcasts a message to Room's Participant's - excluding one given ID of
func (r *Room) broadcastMessage(msg webrtc.DataChannelMessage, excludeID uuid.UUID) {
r.ParticipantsMutex.RLock()
for d, participant := range r.Participants {
if participant.DataChannel != nil {
if d != excludeID { // Don't send back to the sender
if err := participant.DataChannel.SendText(string(msg.Data)); err != nil {
log.Printf("Error broadcasting to %s: %v\n", participant.Name, err)
}
}
}
// signalParticipantOffline signals a single participant offline
func (r *Room) signalParticipantOffline(participant *Participant) error {
// Skip if websocket is nil or closed
if participant.WebSocket == nil || participant.WebSocket.IsClosed() {
return nil
}
if r.DataChannel != nil {
if err := r.DataChannel.SendText(string(msg.Data)); err != nil {
log.Printf("Error broadcasting to Room: %v\n", err)
}
}
r.ParticipantsMutex.RUnlock()
}
// Sends message to Room (nestri-server)
func (r *Room) sendToRoom(msg webrtc.DataChannelMessage) {
if r.DataChannel != nil {
if err := r.DataChannel.SendText(string(msg.Data)); err != nil {
log.Printf("Error broadcasting to Room: %v\n", err)
}
if err := participant.WebSocket.SendAnswerMessageWS(connections.AnswerOffline); err != nil {
return err
}
return nil
}