From 614ede2edcf5f35ac3502766c63d62ce31a9ed06 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Thu, 9 Sep 2021 00:53:15 -0700 Subject: [PATCH] Supports client protocol 3 - subscriber as primary (#110) * use subscriber connection as primary whenever possible * backwards compat for data messages * bump protocol version --- go.mod | 3 +- go.sum | 4 +-- pkg/rtc/participant.go | 54 ++++++++++++++++++++++------ pkg/rtc/participant_internal_test.go | 2 +- pkg/rtc/room.go | 11 +++++- pkg/rtc/types/protocol_version.go | 5 +++ pkg/service/roommanager.go | 5 --- pkg/service/rtcservice.go | 2 +- 8 files changed, 64 insertions(+), 22 deletions(-) diff --git a/go.mod b/go.mod index d1fa5e1e5..4785feaaa 100644 --- a/go.mod +++ b/go.mod @@ -7,14 +7,13 @@ require ( github.com/c9s/goprocinfo v0.0.0-20210130143923-c95fcf8c64a8 github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect github.com/gammazero/workerpool v1.1.2 - github.com/go-logr/logr v1.1.0 github.com/go-logr/zapr v1.1.0 github.com/go-redis/redis/v8 v8.11.3 github.com/google/subcommands v1.2.0 // indirect github.com/google/wire v0.5.0 github.com/gorilla/websocket v1.4.2 github.com/jxskiss/base62 v0.0.0-20191017122030-4f11678b909b - github.com/livekit/protocol v0.8.6 + github.com/livekit/protocol v0.8.7 github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 diff --git a/go.sum b/go.sum index 612bfd487..2d8b0c471 100644 --- a/go.sum +++ b/go.sum @@ -246,8 +246,8 @@ github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= github.com/livekit/ion-sfu v1.20.9 h1:Ih+tE5CXqdL1o6iXfZn+uBoddBXWceaOKVIun6aVuZM= github.com/livekit/ion-sfu v1.20.9/go.mod h1:g8hwobZI5fvX1RXvayf4ZXkgP7spV5YGE4yTSsumpB4= -github.com/livekit/protocol v0.8.6 h1:pNdxWrXFXNvxF5wzsmKiiIYyrwt13dUmZiUQ4awjOEM= -github.com/livekit/protocol v0.8.6/go.mod h1:0LxJXnPRSQ05umVz1NnvRjNWNZct4fsT1OKpIUkukO0= +github.com/livekit/protocol v0.8.7 h1:6zM1ynTseWgWVXzNktEnGGBTE/Lsp5ytojX5o1wV4E8= +github.com/livekit/protocol v0.8.7/go.mod h1:0LxJXnPRSQ05umVz1NnvRjNWNZct4fsT1OKpIUkukO0= github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ= github.com/lucsky/cuid v1.0.2/go.mod h1:QaaJqckboimOmhRSJXSx/+IT+VTfxfPGSo/6mfgUfmE= github.com/lyft/protoc-gen-validate v0.0.13/go.mod h1:XbGvPuh87YZc5TdIa2/I4pLk0QoUACkjt2znoq26NVQ= diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 449da22c4..fcecc3c59 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -55,8 +55,10 @@ type ParticipantImpl struct { pliThrottle *pliThrottle // reliable and unreliable data channels - reliableDC *webrtc.DataChannel - lossyDC *webrtc.DataChannel + reliableDC *webrtc.DataChannel + reliableDCSub *webrtc.DataChannel + lossyDC *webrtc.DataChannel + lossyDCSub *webrtc.DataChannel // when first connected connectedAt time.Time @@ -134,7 +136,28 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { p.sendIceCandidate(c, livekit.SignalTarget_SUBSCRIBER) }) - p.publisher.pc.OnICEConnectionStateChange(p.handlePublisherICEStateChange) + primaryPC := p.publisher.pc + + if p.ProtocolVersion().SubscriberAsPrimary() { + primaryPC = p.subscriber.pc + ordered := true + // also create data channels for subs + p.reliableDCSub, err = primaryPC.CreateDataChannel(reliableDataChannel, &webrtc.DataChannelInit{ + Ordered: &ordered, + }) + if err != nil { + return nil, err + } + retransmits := uint16(0) + p.lossyDCSub, err = primaryPC.CreateDataChannel(lossyDataChannel, &webrtc.DataChannelInit{ + Ordered: &ordered, + MaxRetransmits: &retransmits, + }) + if err != nil { + return nil, err + } + } + primaryPC.OnICEConnectionStateChange(p.handlePrimaryICEStateChange) p.publisher.pc.OnTrack(p.onMediaTrack) p.publisher.pc.OnDataChannel(p.onDataChannel) @@ -476,6 +499,8 @@ func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticip OtherParticipants: ToProtoParticipants(otherParticipants), ServerVersion: version.Version, IceServers: iceServers, + // indicates both server and client support subscriber as primary + SubscriberPrimary: p.ProtocolVersion().SubscriberAsPrimary(), }, }, }) @@ -523,17 +548,26 @@ func (p *ParticipantImpl) SendDataPacket(dp *livekit.DataPacket) error { if err != nil { return err } + + var dc *webrtc.DataChannel if dp.Kind == livekit.DataPacket_RELIABLE { - if p.reliableDC == nil { - return ErrDataChannelUnavailable + if p.ProtocolVersion().SubscriberAsPrimary() { + dc = p.reliableDCSub + } else { + dc = p.reliableDC } - return p.reliableDC.Send(data) } else { - if p.lossyDC == nil { - return ErrDataChannelUnavailable + if p.ProtocolVersion().SubscriberAsPrimary() { + dc = p.lossyDCSub + } else { + dc = p.lossyDC } - return p.lossyDC.Send(data) } + + if dc == nil { + return ErrDataChannelUnavailable + } + return dc.Send(data) } func (p *ParticipantImpl) SetTrackMuted(trackId string, muted bool, fromAdmin bool) { @@ -889,7 +923,7 @@ func (p *ParticipantImpl) handleTrackPublished(track types.PublishedTrack) { } } -func (p *ParticipantImpl) handlePublisherICEStateChange(state webrtc.ICEConnectionState) { +func (p *ParticipantImpl) handlePrimaryICEStateChange(state webrtc.ICEConnectionState) { // logger.Debugw("ICE connection state changed", "state", state.String(), // "participant", p.identity, "pID", p.ID()) if state == webrtc.ICEConnectionStateConnected { diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 82ff33a4a..1c8cb46ec 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -54,7 +54,7 @@ func TestICEStateChange(t *testing.T) { p.onClose = func(participant types.Participant) { close(closeChan) } - p.handlePublisherICEStateChange(webrtc.ICEConnectionStateFailed) + p.handlePrimaryICEStateChange(webrtc.ICEConnectionStateFailed) select { case <-closeChan: diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 74044cf5d..81566047b 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -216,7 +216,16 @@ func (r *Room) Join(participant types.Participant, opts *ParticipantOptions) err } }) - return participant.SendJoinResponse(r.Room, otherParticipants, r.iceServers) + if err := participant.SendJoinResponse(r.Room, otherParticipants, r.iceServers); err != nil { + return err + } + + if participant.ProtocolVersion().SubscriberAsPrimary() { + // initiates sub connection as primary + participant.Negotiate() + } + + return nil } func (r *Room) RemoveParticipant(identity string) { diff --git a/pkg/rtc/types/protocol_version.go b/pkg/rtc/types/protocol_version.go index 9113ddb69..df74a1d33 100644 --- a/pkg/rtc/types/protocol_version.go +++ b/pkg/rtc/types/protocol_version.go @@ -15,3 +15,8 @@ func (v ProtocolVersion) SupportsProtobuf() bool { func (v ProtocolVersion) HandlesDataPackets() bool { return v > 1 } + +// SubscriberAsPrimary indicates clients initiate subscriber connection as primary +func (v ProtocolVersion) SubscriberAsPrimary() bool { + return v > 2 +} diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 2d23eedd8..d88c5f4a2 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -434,11 +434,6 @@ func (r *LocalRoomManager) rtcSessionWorker(room *rtc.Room, participant types.Pa "track", msg.AddTrack.Cid) participant.AddTrack(msg.AddTrack) case *livekit.SignalRequest_Answer: - if participant.State() == livekit.ParticipantInfo_JOINING { - logger.Errorw("cannot negotiate before peer offer", nil, "participant", participant.Identity(), "pID", participant.ID()) - // conn.WriteJSON(jsonError(http.StatusNotAcceptable, "cannot negotiate before peer offer")) - return - } sd := rtc.FromProtoSessionDescription(msg.Answer) if err := participant.HandleAnswer(sd); err != nil { logger.Errorw("could not handle answer", err, "participant", participant.Identity(), "pID", participant.ID()) diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 250b090f9..6fc5ac29b 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -150,7 +150,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { defer func() { // when the source is terminated, this means Participant.Close had been called and RTC connection is done // we would terminate the signal connection as well - conn.Close() + _ = conn.Close() }() defer rtc.Recover() for {