dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-10-15 13:20:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

Supports client protocol 3 - subscriber as primary (#110)

Browse Source 
* use subscriber connection as primary whenever possible

* backwards compat for data messages

* bump protocol version
This commit is contained in:
David Zhao
2021-09-09 00:53:15 -07:00
committed by GitHub
parent 2a3fb5f0e7
commit 614ede2edc
8 changed files with 64 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -2
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/rtc/participant.go
+44 -10
View File
@@ -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 {
pkg/rtc/participant_internal_test.go
+1 -1
View File
@@ -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:
pkg/rtc/room.go
+10 -1
View File
@@ -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) {
pkg/rtc/types/protocol_version.go
+5
View File
@@ -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
}
pkg/service/roommanager.go
-5
View File
@@ -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())
pkg/service/rtcservice.go
+1 -1
View File
@@ -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 {