refactoring, enabling sending out room updates

pkg/rtc/participant.go

@@ -16,7 +16,8 @@ import (
 
 type Participant struct {
 	id          string
-	conn        *webrtc.PeerConnection
+	peerConn    *webrtc.PeerConnection
+	sigConn     SignalConnection
 	ctx         context.Context
 	cancel      context.CancelFunc
 	mediaEngine *MediaEngine
@@ -35,10 +36,11 @@ type Participant struct {
 	OnOffer func(webrtc.SessionDescription)
 	// OnIceCandidate - ice candidate discovered for local peer
 	OnICECandidate func(c *webrtc.ICECandidateInit)
+	OnStateChange  func(p *Participant, oldState livekit.ParticipantInfo_State)
 	OnClose        func(*Participant)
 }
 
-func NewParticipant(conf *WebRTCConfig, name string) (*Participant, error) {
+func NewParticipant(conf *WebRTCConfig, sc SignalConnection, name string) (*Participant, error) {
 	me := webrtc.MediaEngine{}
 	me.RegisterDefaultCodecs()
 	api := webrtc.NewAPI(webrtc.WithMediaEngine(me), webrtc.WithSettingEngine(conf.SettingEngine))
@@ -52,7 +54,8 @@ func NewParticipant(conf *WebRTCConfig, name string) (*Participant, error) {
 	participant := &Participant{
 		id:             utils.NewGuid(utils.ParticipantPrefix),
 		name:           name,
-		conn:           pc,
+		peerConn:       pc,
+		sigConn:        sc,
 		ctx:            ctx,
 		cancel:         cancel,
 		state:          livekit.ParticipantInfo_JOINING,
@@ -63,14 +66,30 @@ func NewParticipant(conf *WebRTCConfig, name string) (*Participant, error) {
 	}
 	participant.mediaEngine.RegisterDefaultCodecs()
 
+	log := logger.GetLogger()
+
 	pc.OnTrack(participant.onTrack)
 
 	pc.OnICECandidate(func(c *webrtc.ICECandidate) {
 		if c == nil {
 			return
 		}
+		ci := c.ToJSON()
+
+		// write candidate
+		err := sc.WriteResponse(&livekit.SignalResponse{
+			Message: &livekit.SignalResponse_Trickle{
+				Trickle: &livekit.Trickle{
+					Candidate: ci.Candidate,
+					// TODO: there are other candidateInit fields that we might want
+				},
+			},
+		})
+		if err != nil {
+			log.Errorw("could not send trickle", "err", err)
+		}
+
 		if participant.OnICECandidate != nil {
-			ci := c.ToJSON()
 			participant.OnICECandidate(&ci)
 		}
 	})
@@ -78,7 +97,7 @@ func NewParticipant(conf *WebRTCConfig, name string) (*Participant, error) {
 	pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
 		logger.GetLogger().Debugw("ICE connection state changed", "state", state.String())
 		if state == webrtc.ICEConnectionStateConnected {
-			participant.state = livekit.ParticipantInfo_ACTIVE
+			participant.updateState(livekit.ParticipantInfo_ACTIVE)
 		}
 	})
 
@@ -108,58 +127,73 @@ func (p *Participant) ToProto() *livekit.ParticipantInfo {
 
 // Answer an offer from remote participant
 func (p *Participant) Answer(sdp webrtc.SessionDescription) (answer webrtc.SessionDescription, err error) {
-	// media engine is already set to default codecs upon startup
-	//if err = p.mediaEngine.PopulateFromSDP(sdp); err != nil {
-	//	err = errors.Wrapf(err, "could not parse SDP")
-	//	return
-	//}
-
 	if p.state == livekit.ParticipantInfo_JOINING {
-		p.state = livekit.ParticipantInfo_JOINED
+		p.updateState(livekit.ParticipantInfo_JOINED)
 	}
 
 	if err = p.SetRemoteDescription(sdp); err != nil {
 		return
 	}
 
-	answer, err = p.conn.CreateAnswer(nil)
+	answer, err = p.peerConn.CreateAnswer(nil)
 	if err != nil {
 		err = errors.Wrap(err, "could not create answer")
 		return
 	}
 
-	if err = p.conn.SetLocalDescription(answer); err != nil {
+	if err = p.peerConn.SetLocalDescription(answer); err != nil {
 		err = errors.Wrap(err, "could not set local description")
 		return
 	}
 
+	// send client the answer
+	err = p.sigConn.WriteResponse(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_Answer{
+			Answer: ToProtoSessionDescription(answer),
+		},
+	})
+	if err != nil {
+		return
+	}
+
 	// only set after answered
-	p.conn.OnNegotiationNeeded(func() {
+	p.peerConn.OnNegotiationNeeded(func() {
 		logger.GetLogger().Debugw("negotiation needed", "participantId", p.ID())
-		offer, err := p.conn.CreateOffer(nil)
+		offer, err := p.peerConn.CreateOffer(nil)
 		if err != nil {
 			logger.GetLogger().Errorw("could not create offer", "err", err)
 			return
 		}
 
-		err = p.conn.SetLocalDescription(offer)
+		err = p.peerConn.SetLocalDescription(offer)
 		if err != nil {
 			logger.GetLogger().Errorw("could not set local description", "err", err)
 			return
 		}
 
 		logger.GetLogger().Debugw("created new offer", "offer", offer, "onOffer", p.OnOffer)
+
+		logger.GetLogger().Debugw("sending available offer to participant")
+		err = p.sigConn.WriteResponse(&livekit.SignalResponse{
+			Message: &livekit.SignalResponse_Negotiate{
+				Negotiate: ToProtoSessionDescription(offer),
+			},
+		})
+		if err != nil {
+			logger.GetLogger().Errorw("could not send offer to peer",
+				"err", err)
+		}
+
 		if p.OnOffer != nil {
 			p.OnOffer(offer)
 		}
 	})
-
 	return
 }
 
 // SetRemoteDescription when receiving an answer from remote
 func (p *Participant) SetRemoteDescription(sdp webrtc.SessionDescription) error {
-	if err := p.conn.SetRemoteDescription(sdp); err != nil {
+	if err := p.peerConn.SetRemoteDescription(sdp); err != nil {
 		return errors.Wrap(err, "could not set remote description")
 	}
 	return nil
@@ -167,7 +201,7 @@ func (p *Participant) SetRemoteDescription(sdp webrtc.SessionDescription) error
 
 // AddICECandidate adds candidates for remote peer
 func (p *Participant) AddICECandidate(candidate webrtc.ICECandidateInit) error {
-	if err := p.conn.AddICECandidate(candidate); err != nil {
+	if err := p.peerConn.AddICECandidate(candidate); err != nil {
 		return err
 	}
 	return nil
@@ -183,11 +217,12 @@ func (p *Participant) Close() error {
 	if p.ctx.Err() != nil {
 		return p.ctx.Err()
 	}
+	p.updateState(livekit.ParticipantInfo_DISCONNECTED)
 	if p.OnClose != nil {
 		p.OnClose(p)
 	}
 	p.cancel()
-	return p.conn.Close()
+	return p.peerConn.Close()
 }
 
 // Subscribes otherPeer to all of the tracks
@@ -216,6 +251,42 @@ func (p *Participant) RemoveSubscriber(peerId string) {
 	}
 }
 
+// signal connection methods
+func (p *Participant) SendJoinResponse(otherParticipants []*Participant) error {
+	// send Join response
+	return p.sigConn.WriteResponse(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_Join{
+			Join: &livekit.JoinResponse{
+				Participant:       p.ToProto(),
+				OtherParticipants: ToProtoParticipants(otherParticipants),
+			},
+		},
+	})
+}
+
+func (p *Participant) SendParticipantUpdate(participants []*livekit.ParticipantInfo) error {
+	return p.sigConn.WriteResponse(&livekit.SignalResponse{
+		Message: &livekit.SignalResponse_Update{
+			Update: &livekit.ParticipantUpdate{
+				Participants: participants,
+			},
+		},
+	})
+}
+
+func (p *Participant) updateState(state livekit.ParticipantInfo_State) {
+	if state == p.state {
+		return
+	}
+	oldState := p.state
+	p.state = state
+	if p.OnStateChange != nil {
+		go func() {
+			p.OnStateChange(p, oldState)
+		}()
+	}
+}
+
 // when a new mediaTrack is created, creates a Track and adds it to room
 func (p *Participant) onTrack(track *webrtc.Track, rtpReceiver *webrtc.RTPReceiver) {
 	logger.GetLogger().Debugw("mediaTrack added", "participantId", p.ID(), "mediaTrack", track.Label())
@@ -223,7 +294,7 @@ func (p *Participant) onTrack(track *webrtc.Track, rtpReceiver *webrtc.RTPReceiv
 	// create Receiver
 	// p.mediaEngine.TCCExt
 	receiver := NewReceiver(p.ctx, p.id, rtpReceiver, p.receiverConfig, 0)
-	pt := NewTrack(p.ctx, p.id, p.conn, track, receiver)
+	pt := NewTrack(p.ctx, p.id, p.peerConn, track, receiver)
 
 	p.lock.Lock()
 	p.tracks = append(p.tracks, pt)
@@ -254,7 +325,7 @@ func (p *Participant) rtcpSendWorker() {
 			}
 			p.lock.RUnlock()
 			if len(pkts) > 0 {
-				if err := p.conn.WriteRTCP(pkts); err != nil {
+				if err := p.peerConn.WriteRTCP(pkts); err != nil {
 					logger.GetLogger().Errorw("error writing RTCP to peer",
 						"peer", p.id,
 						"err", err,

pkg/rtc/room.go

@@ -62,6 +62,10 @@ func (r *Room) Join(participant *Participant) error {
 
 	participant.OnPeerTrack = r.onTrackAdded
 
+	participant.OnStateChange = func(p *Participant, oldState livekit.ParticipantInfo_State) {
+		r.broadcastParticipantState(p)
+	}
+
 	logger.GetLogger().Infow("new participant joined",
 		"id", participant.ID(),
 		"name", participant.Name(),
@@ -79,7 +83,15 @@ func (r *Room) Join(participant *Participant) error {
 
 	r.participants[participant.ID()] = participant
 
-	return nil
+	// gather other participants and send join response
+	otherParticipants := make([]*Participant, 0, len(r.participants))
+	for _, p := range r.participants {
+		if p.id != participant.id {
+			otherParticipants = append(otherParticipants, p)
+		}
+	}
+
+	return participant.SendJoinResponse(otherParticipants)
 }
 
 func (r *Room) RemoveParticipant(id string) {
@@ -113,3 +125,23 @@ func (r *Room) onTrackAdded(peer *Participant, track *Track) {
 		}
 	}
 }
+
+func (r *Room) broadcastParticipantState(p *Participant) {
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+
+	updates := ToProtoParticipants([]*Participant{p})
+	for _, op := range r.participants {
+		// skip itself
+		if p.id == op.id {
+			continue
+		}
+
+		err := p.SendParticipantUpdate(updates)
+		if err != nil {
+			logger.GetLogger().Errorw("could not send update to participant",
+				"participant", p.id,
+				"err", err)
+		}
+	}
+}

pkg/rtc/track.go

@@ -62,12 +62,12 @@ func (t *Track) AddSubscriber(participant *Participant) error {
 	packedId := PackTrackId(t.participantId, t.mediaTrack.ID())
 
 	// use existing SSRC with simple forwarders. adaptive forwarders require unique SSRC per layer
-	outTrack, err := participant.conn.NewTrack(codecs[0].PayloadType, t.mediaTrack.SSRC(), packedId, t.mediaTrack.Label())
+	outTrack, err := participant.peerConn.NewTrack(codecs[0].PayloadType, t.mediaTrack.SSRC(), packedId, t.mediaTrack.Label())
 	if err != nil {
 		return err
 	}
 
-	rtpSender, err := participant.conn.AddTrack(outTrack)
+	rtpSender, err := participant.peerConn.AddTrack(outTrack)
 	if err != nil {
 		return err
 	}
@@ -78,7 +78,7 @@ func (t *Track) AddSubscriber(participant *Participant) error {
 		delete(t.forwarders, participant.ID())
 		t.lock.Unlock()
 
-		if err := participant.conn.RemoveTrack(rtpSender); err != nil {
+		if err := participant.peerConn.RemoveTrack(rtpSender); err != nil {
 			logger.GetLogger().Warnw("could not remove mediaTrack from forwarder",
 				"participant", participant.ID(),
 				"err", err)

pkg/rtc/utils.go

@@ -2,6 +2,10 @@ package rtc
 
 import (
 	"strings"
+
+	"github.com/pion/webrtc/v3"
+
+	"github.com/livekit/livekit-server/proto/livekit"
 )
 
 const (
@@ -19,3 +23,48 @@ func UnpackTrackId(packed string) (peerId string, trackId string) {
 func PackTrackId(participantId, trackId string) string {
 	return participantId + trackIdSeparator + trackId
 }
+
+func ToProtoParticipants(participants []*Participant) []*livekit.ParticipantInfo {
+	infos := make([]*livekit.ParticipantInfo, 0, len(participants))
+	for _, op := range participants {
+		infos = append(infos, op.ToProto())
+	}
+	return infos
+}
+
+func ToProtoSessionDescription(sd webrtc.SessionDescription) *livekit.SessionDescription {
+	return &livekit.SessionDescription{
+		Type: sd.Type.String(),
+		Sdp:  sd.SDP,
+	}
+}
+
+func FromProtoSessionDescription(sd *livekit.SessionDescription) webrtc.SessionDescription {
+	var sdType webrtc.SDPType
+	switch sd.Type {
+	case webrtc.SDPTypeOffer.String():
+		sdType = webrtc.SDPTypeOffer
+	case webrtc.SDPTypeAnswer.String():
+		sdType = webrtc.SDPTypeAnswer
+	case webrtc.SDPTypePranswer.String():
+		sdType = webrtc.SDPTypePranswer
+	case webrtc.SDPTypeRollback.String():
+		sdType = webrtc.SDPTypeRollback
+	}
+	return webrtc.SessionDescription{
+		Type: sdType,
+		SDP:  sd.Sdp,
+	}
+}
+
+func ToProtoTrickle(candidateInit *webrtc.ICECandidateInit) *livekit.Trickle {
+	return &livekit.Trickle{
+		Candidate: candidateInit.Candidate,
+	}
+}
+
+func FromProtoTrickle(trickle *livekit.Trickle) *webrtc.ICECandidateInit {
+	return &webrtc.ICECandidateInit{
+		Candidate: trickle.Candidate,
+	}
+}

pkg/rtc/wsprotocol.go

@@ -1,4 +1,4 @@
-package service
+package rtc
 
 import (
 	"github.com/gorilla/websocket"
@@ -19,9 +19,10 @@ type WSSignalConnection struct {
 	useJSON bool
 }
 
-func NewWSSignalConnection(conn *websocket.Conn, name string) *WSSignalConnection {
+func NewWSSignalConnection(conn *websocket.Conn) *WSSignalConnection {
 	return &WSSignalConnection{
-		conn: conn,
+		conn:    conn,
+		useJSON: true,
 	}
 }
 

pkg/service/rtc.go

@@ -59,11 +59,6 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	participant, err := rtc.NewParticipant(s.manager.Config(), pName)
-	if err != nil {
-		writeJSONError(w, http.StatusInternalServerError, "could not create participant", err.Error())
-	}
-
 	// upgrade only once the basics are good to go
 	conn, err := s.upgrader.Upgrade(w, r, nil)
 	if err != nil {
@@ -75,22 +70,25 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		log.Infow("websocket closed by remote")
 		return nil
 	})
-	defer func() {
-		log.Infow("connection returned")
-	}()
-	signalConn := NewWSSignalConnection(conn, pName)
+	signalConn := rtc.NewWSSignalConnection(conn)
 
-	// send Join response
-	signalConn.WriteResponse(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_Join{
-			Join: &livekit.JoinResponse{
-				Participant: participant.ToProto(),
-			},
-		},
-	})
+	participant, err := rtc.NewParticipant(s.manager.Config(), signalConn, pName)
+	if err != nil {
+		writeJSONError(w, http.StatusInternalServerError, "could not create participant", err.Error())
+		return
+	}
+
+	if err := room.Join(participant); err != nil {
+		writeJSONError(w, http.StatusInternalServerError, "could not join room", err.Error())
+		return
+	}
+
+	defer func() {
+		participant.Close()
+		log.Infow("WS connection closed")
+	}()
 
 	// read connection and wait for commands
-	// TODO: pass in context from WS, so termination of WS would disconnect RTC
 	//ctx := context.Background()
 	for {
 		req, err := signalConn.ReadRequest()
@@ -104,7 +102,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 		switch msg := req.Message.(type) {
 		case *livekit.SignalRequest_Offer:
-			err = s.handleJoin(signalConn, room, participant, msg.Offer.Sdp)
+			err = s.handleOffer(participant, msg.Offer)
 			if err != nil {
 				log.Errorw("could not handle join", "err", err, "participant", participant.ID())
 				return
@@ -145,74 +143,22 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func (s *RTCService) handleJoin(sc SignalConnection, room *rtc.Room, participant *rtc.Participant, sdp string) error {
+func (s *RTCService) handleOffer(participant *rtc.Participant, offer *livekit.SessionDescription) error {
 	log := logger.GetLogger()
 
-	log.Infow("handling join")
-	err := room.Join(participant)
-	if err != nil {
-		return errors.Wrap(err, "could not join room")
-	}
-
-	// TODO: it might be better to return error instead of nil
-	participant.OnICECandidate = func(c *webrtc.ICECandidateInit) {
-		log.Debugw("sending ICE candidate", "participantId", participant.ID())
-		err = sc.WriteResponse(&livekit.SignalResponse{
-			Message: &livekit.SignalResponse_Trickle{
-				Trickle: &livekit.Trickle{
-					Candidate: c.Candidate,
-					// TODO: there are other candidateInit fields that we might want
-				},
-			},
-		})
-		if err != nil {
-			log.Errorw("could not send trickle", "err", err)
-		}
-	}
-
-	// send peer new offer
-	participant.OnOffer = func(o webrtc.SessionDescription) {
-		log.Debugw("sending available offer to participant")
-		err := sc.WriteResponse(&livekit.SignalResponse{
-			Message: &livekit.SignalResponse_Negotiate{
-				Negotiate: ToProtoSessionDescription(o),
-			},
-		})
-		if err != nil {
-			logger.GetLogger().Errorw("could not send offer to peer",
-				"err", err)
-		}
-	}
-
-	offer := webrtc.SessionDescription{
-		Type: webrtc.SDPTypeOffer,
-		SDP:  sdp,
-	}
-	answer, err := participant.Answer(offer)
+	_, err := participant.Answer(rtc.FromProtoSessionDescription(offer))
 	if err != nil {
 		return errors.Wrap(err, "could not answer offer")
 	}
 
-	logger.GetLogger().Debugw("answered, writing response")
-
-	// finally send answer
-	err = sc.WriteResponse(&livekit.SignalResponse{
-		Message: &livekit.SignalResponse_Answer{
-			Answer: ToProtoSessionDescription(answer),
-		},
-	})
-
-	if err != nil {
-		return errors.Wrap(err, "could not create answer")
-	}
-
+	log.Debugw("answered client offer")
 	return nil
 }
 
-func (s *RTCService) handleNegotiate(sc SignalConnection, peer *rtc.Participant, neg *livekit.SessionDescription) error {
+func (s *RTCService) handleNegotiate(sc rtc.SignalConnection, peer *rtc.Participant, neg *livekit.SessionDescription) error {
 	logger.GetLogger().Debugw("handling incoming negotiate")
 	if neg.Type == webrtc.SDPTypeOffer.String() {
-		offer := FromProtoSessionDescription(neg)
+		offer := rtc.FromProtoSessionDescription(neg)
 		answer, err := peer.Answer(offer)
 		if err != nil {
 			return err
@@ -220,7 +166,7 @@ func (s *RTCService) handleNegotiate(sc SignalConnection, peer *rtc.Participant,
 
 		err = sc.WriteResponse(&livekit.SignalResponse{
 			Message: &livekit.SignalResponse_Negotiate{
-				Negotiate: ToProtoSessionDescription(answer),
+				Negotiate: rtc.ToProtoSessionDescription(answer),
 			},
 		})
 
@@ -228,7 +174,7 @@ func (s *RTCService) handleNegotiate(sc SignalConnection, peer *rtc.Participant,
 			return err
 		}
 	} else if neg.Type == webrtc.SDPTypeAnswer.String() {
-		answer := FromProtoSessionDescription(neg)
+		answer := rtc.FromProtoSessionDescription(neg)
 		err := peer.SetRemoteDescription(answer)
 		if err != nil {
 			return err
@@ -238,7 +184,7 @@ func (s *RTCService) handleNegotiate(sc SignalConnection, peer *rtc.Participant,
 }
 
 func (s *RTCService) handleTrickle(peer *rtc.Participant, trickle *livekit.Trickle) error {
-	candidateInit := FromProtoTrickle(trickle)
+	candidateInit := rtc.FromProtoTrickle(trickle)
 	logger.GetLogger().Debugw("adding peer candidate", "participantId", peer.ID())
 	if err := peer.AddICECandidate(*candidateInit); err != nil {
 		return err

pkg/service/utils.go

@@ -1,44 +0,0 @@
-package service
-
-import (
-	"github.com/pion/webrtc/v3"
-
-	"github.com/livekit/livekit-server/proto/livekit"
-)
-
-func ToProtoSessionDescription(sd webrtc.SessionDescription) *livekit.SessionDescription {
-	return &livekit.SessionDescription{
-		Type: sd.Type.String(),
-		Sdp:  sd.SDP,
-	}
-}
-
-func FromProtoSessionDescription(sd *livekit.SessionDescription) webrtc.SessionDescription {
-	var sdType webrtc.SDPType
-	switch sd.Type {
-	case webrtc.SDPTypeOffer.String():
-		sdType = webrtc.SDPTypeOffer
-	case webrtc.SDPTypeAnswer.String():
-		sdType = webrtc.SDPTypeAnswer
-	case webrtc.SDPTypePranswer.String():
-		sdType = webrtc.SDPTypePranswer
-	case webrtc.SDPTypeRollback.String():
-		sdType = webrtc.SDPTypeRollback
-	}
-	return webrtc.SessionDescription{
-		Type: sdType,
-		SDP:  sd.Sdp,
-	}
-}
-
-func ToProtoTrickle(candidateInit *webrtc.ICECandidateInit) *livekit.Trickle {
-	return &livekit.Trickle{
-		Candidate: candidateInit.Candidate,
-	}
-}
-
-func FromProtoTrickle(trickle *livekit.Trickle) *webrtc.ICECandidateInit {
-	return &webrtc.ICECandidateInit{
-		Candidate: trickle.Candidate,
-	}
-}