peer interfaces and callbacks

pkg/rtc/errors.go

@@ -0,0 +1,8 @@
+package rtc
+
+import "errors"
+
+var (
+	ErrPeerExists       = errors.New("peer already exists")
+	ErrPeerNotConnected = errors.New("peer has not been connected")
+)

pkg/rtc/manager.go

@@ -40,7 +40,7 @@ func (m *RoomManager) GetRoom(roomId string) *Room {
 }
 
 func (m *RoomManager) CreateRoom(req *livekit.CreateRoomRequest) (r *Room, err error) {
-	r, err = NewRoomForRequest(req)
+	r, err = NewRoomForRequest(req, &m.config)
 	if err != nil {
 		return
 	}

pkg/rtc/peer.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
+	"github.com/pkg/errors"
 )
 
 type WebRTCPeer struct {
@@ -19,10 +20,16 @@ type WebRTCPeer struct {
 	lock           sync.RWMutex
 	receiverConfig ReceiverConfig
 	tracks         []*PeerTrack // tracks that the peer is publishing
+	once           sync.Once
 
 	// callbacks & handlers
-	onClose     func(*WebRTCPeer)
-	onPeerTrack func(*WebRTCPeer, *PeerTrack)
+	// OnPeerTrack - remote peer added a track
+	OnPeerTrack func(*WebRTCPeer, *PeerTrack)
+	// OnOffer - offer is ready for remote peer
+	OnOffer func(webrtc.SessionDescription)
+	// OnIceCandidate - ice candidate discovered for local peer
+	OnICECandidate func(c *webrtc.ICECandidate)
+	OnClose        func(*WebRTCPeer)
 }
 
 func NewWebRTCPeer(id string, me *MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) {
@@ -47,6 +54,32 @@ func NewWebRTCPeer(id string, me *MediaEngine, conf WebRTCConfig) (*WebRTCPeer,
 
 	pc.OnTrack(peer.onTrack)
 
+	pc.OnNegotiationNeeded(func() {
+		offer, err := pc.CreateOffer(nil)
+		if err != nil {
+			// TODO: log
+			return
+		}
+
+		err = pc.SetLocalDescription(offer)
+		if err != nil {
+			// TODO: log
+			return
+		}
+		if peer.OnOffer != nil {
+			peer.OnOffer(offer)
+		}
+	})
+
+	pc.OnICECandidate(func(c *webrtc.ICECandidate) {
+		if c == nil {
+			return
+		}
+		if peer.OnICECandidate != nil {
+			peer.OnICECandidate(c)
+		}
+	})
+
 	// TODO: handle data channel
 
 	return peer, nil
@@ -56,26 +89,59 @@ func (p *WebRTCPeer) ID() string {
 	return p.id
 }
 
+// Answer an offer from remote peer
+func (p *WebRTCPeer) Answer(sdp webrtc.SessionDescription) (answer webrtc.SessionDescription, err error) {
+	if err = p.SetRemoteDescription(sdp); err != nil {
+		return
+	}
+
+	answer, err = p.conn.CreateAnswer(nil)
+	if err != nil {
+		err = errors.Wrap(err, "could not create answer")
+		return
+	}
+
+	if err = p.conn.SetLocalDescription(answer); err != nil {
+		err = errors.Wrap(err, "could not set local description")
+		return
+	}
+
+	return
+}
+
+// SetRemoteDescription when receiving an answer from remote
+func (p *WebRTCPeer) SetRemoteDescription(sdp webrtc.SessionDescription) error {
+	if err := p.conn.SetRemoteDescription(sdp); err != nil {
+		return errors.Wrap(err, "could not set remote description")
+	}
+	return nil
+}
+
+// AddICECandidate adds candidates for remote peer
+func (p *WebRTCPeer) AddICECandidate(candidate webrtc.ICECandidateInit) error {
+	if err := p.conn.AddICECandidate(candidate); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (p *WebRTCPeer) Start() {
+	p.once.Do(func() {
+		go p.rtcpSendWorker()
+	})
+}
+
 func (p *WebRTCPeer) Close() error {
 	if p.ctx.Err() != nil {
 		return p.ctx.Err()
 	}
-	// TODO: notify handler of closure
-	if p.onClose != nil {
-		p.onClose(p)
+	if p.OnClose != nil {
+		p.OnClose(p)
 	}
 	p.cancel()
 	return p.conn.Close()
 }
 
-func (p *WebRTCPeer) OnClose(f func(*WebRTCPeer)) {
-	p.onClose = f
-}
-
-func (p *WebRTCPeer) OnPeerTrack(f func(*WebRTCPeer, *PeerTrack)) {
-	p.onPeerTrack = f
-}
-
 // when a new track is created, creates a PeerTrack and adds it to room
 func (p *WebRTCPeer) onTrack(track *webrtc.Track, rtpReceiver *webrtc.RTPReceiver) {
 
@@ -87,9 +153,9 @@ func (p *WebRTCPeer) onTrack(track *webrtc.Track, rtpReceiver *webrtc.RTPReceive
 	defer p.lock.Unlock()
 	p.tracks = append(p.tracks, pt)
 
-	if p.onPeerTrack != nil {
+	if p.OnPeerTrack != nil {
 		// caller should hook up what happens when the peer track is available
-		p.onPeerTrack(p, pt)
+		p.OnPeerTrack(p, pt)
 	}
 }
 

pkg/rtc/room.go

@@ -1,27 +1,32 @@
 package rtc
 
 import (
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/pion/webrtc/v3"
+	"github.com/pkg/errors"
 
 	"github.com/livekit/livekit-server/proto/livekit"
 )
 
 type Room struct {
 	livekit.Room
-
+	config WebRTCConfig
+	lock   sync.RWMutex
 	// map of peerId -> WebRTCPeer
 	peers map[string]*WebRTCPeer
 	// Client ID => list of tracks they are publishing
 	//tracks map[string][]PeerTrack
 }
 
-func NewRoomForRequest(req *livekit.CreateRoomRequest) (*Room, error) {
+func NewRoomForRequest(req *livekit.CreateRoomRequest, config *WebRTCConfig) (*Room, error) {
 	id, err := uuid.NewRandom()
 	if err != nil {
 		return nil, err
 	}
+
 	return &Room{
 		Room: livekit.Room{
 			RoomId:          req.RoomId,
@@ -30,7 +35,9 @@ func NewRoomForRequest(req *livekit.CreateRoomRequest) (*Room, error) {
 			CreationTime:    time.Now().Unix(),
 			Token:           id.String(),
 		},
-		peers: make(map[string]*WebRTCPeer),
+		config: *config,
+		lock:   sync.RWMutex{},
+		peers:  make(map[string]*WebRTCPeer),
 	}, nil
 }
 
@@ -43,3 +50,34 @@ func (r *Room) ToRoomInfo(node *livekit.Node) *livekit.RoomInfo {
 		Token:        r.Token,
 	}
 }
+
+func (r *Room) Join(peerId string, token string, sdp string) (peer *WebRTCPeer, err error) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	_, ok := r.peers[peerId]
+
+	if ok {
+		// already exists, return error
+		return nil, ErrPeerExists
+	}
+
+	offer := webrtc.SessionDescription{
+		Type: webrtc.SDPTypeOffer,
+		SDP:  sdp,
+	}
+
+	me := MediaEngine{}
+	if err := me.PopulateFromSDP(offer); err != nil {
+		return nil, errors.Wrapf(err, "could not parse SDP")
+	}
+
+	peer, err = NewWebRTCPeer(peerId, &me, r.config)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not create peer")
+	}
+
+	r.peers[peerId] = peer
+
+	return
+}