diff --git a/Makefile b/Makefile index d456399b9..d18471376 100644 --- a/Makefile +++ b/Makefile @@ -25,14 +25,17 @@ server: generate go build -i -o ../../bin/livekit-server ;\ } -generate: wire +generate: wire mockgen @{ \ echo "wiring" ;\ cd cmd/server ;\ $(WIRE) ;\ + cd ../.. ;\ + echo "updating mocks" ;\ + $(MOCKGEN) -source pkg/rtc/interfaces.go -destination pkg/rtc/mock_test.go -package rtc ;\ } -test: wire +test: generate go test ./... GO_TARGET=proto/livekit @@ -82,3 +85,14 @@ WIRE=$(GOBIN)/wire else WIRE=$(shell which wire) endif + +mockgen: +ifeq (, $(shell which mockgen)) + @{ \ + echo "installing mockgen" ;\ + go get github.com/golang/mock/mockgen ;\ + } +MOCKGEN=$(GOBIN)/mockgen +else +MOCKGEN=$(shell which mockgen) +endif diff --git a/go.mod b/go.mod index d69dcfdc1..4b5075488 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/livekit/livekit-server go 1.15 require ( + github.com/golang/mock v1.4.4 github.com/golang/protobuf v1.4.3 github.com/google/uuid v1.1.2 github.com/google/wire v0.4.0 diff --git a/go.sum b/go.sum index cefa0373d..795c0165d 100644 --- a/go.sum +++ b/go.sum @@ -111,7 +111,10 @@ github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4er github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A= github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A= +github.com/golang/mock v1.3.1 h1:qGJ6qTW+x6xX/my+8YUVl4WNpX9B7+/l2tRsHGZ7f2s= github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y= +github.com/golang/mock v1.4.4 h1:l75CXGRSwbaYNpl/Z2X1XIIAMSCquvXgpVZDhwEIJsc= +github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= diff --git a/pkg/rtc/interfaces.go b/pkg/rtc/interfaces.go new file mode 100644 index 000000000..64ddc384c --- /dev/null +++ b/pkg/rtc/interfaces.go @@ -0,0 +1,43 @@ +package rtc + +import ( + "time" + + "github.com/pion/rtcp" + "github.com/pion/webrtc/v3" + + "github.com/livekit/livekit-server/proto/livekit" +) + +type WebsocketClient interface { + ReadMessage() (messageType int, p []byte, err error) + WriteMessage(messageType int, data []byte) error + WriteControl(messageType int, data []byte, deadline time.Time) error +} + +type SignalConnection interface { + ReadRequest() (*livekit.SignalRequest, error) + WriteResponse(*livekit.SignalResponse) error +} + +// use interface to make it easier to mock +type PeerConnection interface { + OnICECandidate(f func(*webrtc.ICECandidate)) + OnICEConnectionStateChange(func(webrtc.ICEConnectionState)) + OnTrack(f func(*webrtc.TrackRemote, *webrtc.RTPReceiver)) + OnDataChannel(func(d *webrtc.DataChannel)) + OnNegotiationNeeded(f func()) + Close() error + SetRemoteDescription(desc webrtc.SessionDescription) error + SetLocalDescription(desc webrtc.SessionDescription) error + CreateOffer(options *webrtc.OfferOptions) (webrtc.SessionDescription, error) + CreateAnswer(options *webrtc.AnswerOptions) (webrtc.SessionDescription, error) + AddICECandidate(candidate webrtc.ICECandidateInit) error + WriteRTCP(pkts []rtcp.Packet) error + // used by datatrack + CreateDataChannel(label string, options *webrtc.DataChannelInit) (*webrtc.DataChannel, error) + // used by mediatrack + AddTransceiverFromTrack(track webrtc.TrackLocal, init ...webrtc.RtpTransceiverInit) (*webrtc.RTPTransceiver, error) + ConnectionState() webrtc.PeerConnectionState + RemoveTrack(sender *webrtc.RTPSender) error +} diff --git a/pkg/rtc/mock_test.go b/pkg/rtc/mock_test.go new file mode 100644 index 000000000..6bacffa57 --- /dev/null +++ b/pkg/rtc/mock_test.go @@ -0,0 +1,379 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: pkg/rtc/interfaces.go + +// Package rtc is a generated GoMock package. +package rtc + +import ( + gomock "github.com/golang/mock/gomock" + livekit "github.com/livekit/livekit-server/proto/livekit" + rtcp "github.com/pion/rtcp" + webrtc "github.com/pion/webrtc/v3" + reflect "reflect" + time "time" +) + +// MockWebsocketClient is a mock of WebsocketClient interface +type MockWebsocketClient struct { + ctrl *gomock.Controller + recorder *MockWebsocketClientMockRecorder +} + +// MockWebsocketClientMockRecorder is the mock recorder for MockWebsocketClient +type MockWebsocketClientMockRecorder struct { + mock *MockWebsocketClient +} + +// NewMockWebsocketClient creates a new mock instance +func NewMockWebsocketClient(ctrl *gomock.Controller) *MockWebsocketClient { + mock := &MockWebsocketClient{ctrl: ctrl} + mock.recorder = &MockWebsocketClientMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use +func (m *MockWebsocketClient) EXPECT() *MockWebsocketClientMockRecorder { + return m.recorder +} + +// ReadMessage mocks base method +func (m *MockWebsocketClient) ReadMessage() (int, []byte, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ReadMessage") + ret0, _ := ret[0].(int) + ret1, _ := ret[1].([]byte) + ret2, _ := ret[2].(error) + return ret0, ret1, ret2 +} + +// ReadMessage indicates an expected call of ReadMessage +func (mr *MockWebsocketClientMockRecorder) ReadMessage() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReadMessage", reflect.TypeOf((*MockWebsocketClient)(nil).ReadMessage)) +} + +// WriteMessage mocks base method +func (m *MockWebsocketClient) WriteMessage(messageType int, data []byte) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "WriteMessage", messageType, data) + ret0, _ := ret[0].(error) + return ret0 +} + +// WriteMessage indicates an expected call of WriteMessage +func (mr *MockWebsocketClientMockRecorder) WriteMessage(messageType, data interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WriteMessage", reflect.TypeOf((*MockWebsocketClient)(nil).WriteMessage), messageType, data) +} + +// WriteControl mocks base method +func (m *MockWebsocketClient) WriteControl(messageType int, data []byte, deadline time.Time) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "WriteControl", messageType, data, deadline) + ret0, _ := ret[0].(error) + return ret0 +} + +// WriteControl indicates an expected call of WriteControl +func (mr *MockWebsocketClientMockRecorder) WriteControl(messageType, data, deadline interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WriteControl", reflect.TypeOf((*MockWebsocketClient)(nil).WriteControl), messageType, data, deadline) +} + +// MockSignalConnection is a mock of SignalConnection interface +type MockSignalConnection struct { + ctrl *gomock.Controller + recorder *MockSignalConnectionMockRecorder +} + +// MockSignalConnectionMockRecorder is the mock recorder for MockSignalConnection +type MockSignalConnectionMockRecorder struct { + mock *MockSignalConnection +} + +// NewMockSignalConnection creates a new mock instance +func NewMockSignalConnection(ctrl *gomock.Controller) *MockSignalConnection { + mock := &MockSignalConnection{ctrl: ctrl} + mock.recorder = &MockSignalConnectionMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use +func (m *MockSignalConnection) EXPECT() *MockSignalConnectionMockRecorder { + return m.recorder +} + +// ReadRequest mocks base method +func (m *MockSignalConnection) ReadRequest() (*livekit.SignalRequest, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ReadRequest") + ret0, _ := ret[0].(*livekit.SignalRequest) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ReadRequest indicates an expected call of ReadRequest +func (mr *MockSignalConnectionMockRecorder) ReadRequest() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReadRequest", reflect.TypeOf((*MockSignalConnection)(nil).ReadRequest)) +} + +// WriteResponse mocks base method +func (m *MockSignalConnection) WriteResponse(arg0 *livekit.SignalResponse) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "WriteResponse", arg0) + ret0, _ := ret[0].(error) + return ret0 +} + +// WriteResponse indicates an expected call of WriteResponse +func (mr *MockSignalConnectionMockRecorder) WriteResponse(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WriteResponse", reflect.TypeOf((*MockSignalConnection)(nil).WriteResponse), arg0) +} + +// MockPeerConnection is a mock of PeerConnection interface +type MockPeerConnection struct { + ctrl *gomock.Controller + recorder *MockPeerConnectionMockRecorder +} + +// MockPeerConnectionMockRecorder is the mock recorder for MockPeerConnection +type MockPeerConnectionMockRecorder struct { + mock *MockPeerConnection +} + +// NewMockPeerConnection creates a new mock instance +func NewMockPeerConnection(ctrl *gomock.Controller) *MockPeerConnection { + mock := &MockPeerConnection{ctrl: ctrl} + mock.recorder = &MockPeerConnectionMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use +func (m *MockPeerConnection) EXPECT() *MockPeerConnectionMockRecorder { + return m.recorder +} + +// OnICECandidate mocks base method +func (m *MockPeerConnection) OnICECandidate(f func(*webrtc.ICECandidate)) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "OnICECandidate", f) +} + +// OnICECandidate indicates an expected call of OnICECandidate +func (mr *MockPeerConnectionMockRecorder) OnICECandidate(f interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnICECandidate", reflect.TypeOf((*MockPeerConnection)(nil).OnICECandidate), f) +} + +// OnICEConnectionStateChange mocks base method +func (m *MockPeerConnection) OnICEConnectionStateChange(arg0 func(webrtc.ICEConnectionState)) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "OnICEConnectionStateChange", arg0) +} + +// OnICEConnectionStateChange indicates an expected call of OnICEConnectionStateChange +func (mr *MockPeerConnectionMockRecorder) OnICEConnectionStateChange(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnICEConnectionStateChange", reflect.TypeOf((*MockPeerConnection)(nil).OnICEConnectionStateChange), arg0) +} + +// OnTrack mocks base method +func (m *MockPeerConnection) OnTrack(f func(*webrtc.TrackRemote, *webrtc.RTPReceiver)) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "OnTrack", f) +} + +// OnTrack indicates an expected call of OnTrack +func (mr *MockPeerConnectionMockRecorder) OnTrack(f interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnTrack", reflect.TypeOf((*MockPeerConnection)(nil).OnTrack), f) +} + +// OnDataChannel mocks base method +func (m *MockPeerConnection) OnDataChannel(arg0 func(*webrtc.DataChannel)) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "OnDataChannel", arg0) +} + +// OnDataChannel indicates an expected call of OnDataChannel +func (mr *MockPeerConnectionMockRecorder) OnDataChannel(arg0 interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnDataChannel", reflect.TypeOf((*MockPeerConnection)(nil).OnDataChannel), arg0) +} + +// OnNegotiationNeeded mocks base method +func (m *MockPeerConnection) OnNegotiationNeeded(f func()) { + m.ctrl.T.Helper() + m.ctrl.Call(m, "OnNegotiationNeeded", f) +} + +// OnNegotiationNeeded indicates an expected call of OnNegotiationNeeded +func (mr *MockPeerConnectionMockRecorder) OnNegotiationNeeded(f interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OnNegotiationNeeded", reflect.TypeOf((*MockPeerConnection)(nil).OnNegotiationNeeded), f) +} + +// Close mocks base method +func (m *MockPeerConnection) Close() error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Close") + ret0, _ := ret[0].(error) + return ret0 +} + +// Close indicates an expected call of Close +func (mr *MockPeerConnectionMockRecorder) Close() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockPeerConnection)(nil).Close)) +} + +// SetRemoteDescription mocks base method +func (m *MockPeerConnection) SetRemoteDescription(desc webrtc.SessionDescription) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SetRemoteDescription", desc) + ret0, _ := ret[0].(error) + return ret0 +} + +// SetRemoteDescription indicates an expected call of SetRemoteDescription +func (mr *MockPeerConnectionMockRecorder) SetRemoteDescription(desc interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetRemoteDescription", reflect.TypeOf((*MockPeerConnection)(nil).SetRemoteDescription), desc) +} + +// SetLocalDescription mocks base method +func (m *MockPeerConnection) SetLocalDescription(desc webrtc.SessionDescription) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SetLocalDescription", desc) + ret0, _ := ret[0].(error) + return ret0 +} + +// SetLocalDescription indicates an expected call of SetLocalDescription +func (mr *MockPeerConnectionMockRecorder) SetLocalDescription(desc interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetLocalDescription", reflect.TypeOf((*MockPeerConnection)(nil).SetLocalDescription), desc) +} + +// CreateOffer mocks base method +func (m *MockPeerConnection) CreateOffer(options *webrtc.OfferOptions) (webrtc.SessionDescription, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "CreateOffer", options) + ret0, _ := ret[0].(webrtc.SessionDescription) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// CreateOffer indicates an expected call of CreateOffer +func (mr *MockPeerConnectionMockRecorder) CreateOffer(options interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateOffer", reflect.TypeOf((*MockPeerConnection)(nil).CreateOffer), options) +} + +// CreateAnswer mocks base method +func (m *MockPeerConnection) CreateAnswer(options *webrtc.AnswerOptions) (webrtc.SessionDescription, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "CreateAnswer", options) + ret0, _ := ret[0].(webrtc.SessionDescription) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// CreateAnswer indicates an expected call of CreateAnswer +func (mr *MockPeerConnectionMockRecorder) CreateAnswer(options interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateAnswer", reflect.TypeOf((*MockPeerConnection)(nil).CreateAnswer), options) +} + +// AddICECandidate mocks base method +func (m *MockPeerConnection) AddICECandidate(candidate webrtc.ICECandidateInit) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "AddICECandidate", candidate) + ret0, _ := ret[0].(error) + return ret0 +} + +// AddICECandidate indicates an expected call of AddICECandidate +func (mr *MockPeerConnectionMockRecorder) AddICECandidate(candidate interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddICECandidate", reflect.TypeOf((*MockPeerConnection)(nil).AddICECandidate), candidate) +} + +// WriteRTCP mocks base method +func (m *MockPeerConnection) WriteRTCP(pkts []rtcp.Packet) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "WriteRTCP", pkts) + ret0, _ := ret[0].(error) + return ret0 +} + +// WriteRTCP indicates an expected call of WriteRTCP +func (mr *MockPeerConnectionMockRecorder) WriteRTCP(pkts interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "WriteRTCP", reflect.TypeOf((*MockPeerConnection)(nil).WriteRTCP), pkts) +} + +// CreateDataChannel mocks base method +func (m *MockPeerConnection) CreateDataChannel(label string, options *webrtc.DataChannelInit) (*webrtc.DataChannel, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "CreateDataChannel", label, options) + ret0, _ := ret[0].(*webrtc.DataChannel) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// CreateDataChannel indicates an expected call of CreateDataChannel +func (mr *MockPeerConnectionMockRecorder) CreateDataChannel(label, options interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateDataChannel", reflect.TypeOf((*MockPeerConnection)(nil).CreateDataChannel), label, options) +} + +// AddTransceiverFromTrack mocks base method +func (m *MockPeerConnection) AddTransceiverFromTrack(track webrtc.TrackLocal, init ...webrtc.RtpTransceiverInit) (*webrtc.RTPTransceiver, error) { + m.ctrl.T.Helper() + varargs := []interface{}{track} + for _, a := range init { + varargs = append(varargs, a) + } + ret := m.ctrl.Call(m, "AddTransceiverFromTrack", varargs...) + ret0, _ := ret[0].(*webrtc.RTPTransceiver) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// AddTransceiverFromTrack indicates an expected call of AddTransceiverFromTrack +func (mr *MockPeerConnectionMockRecorder) AddTransceiverFromTrack(track interface{}, init ...interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + varargs := append([]interface{}{track}, init...) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddTransceiverFromTrack", reflect.TypeOf((*MockPeerConnection)(nil).AddTransceiverFromTrack), varargs...) +} + +// ConnectionState mocks base method +func (m *MockPeerConnection) ConnectionState() webrtc.PeerConnectionState { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ConnectionState") + ret0, _ := ret[0].(webrtc.PeerConnectionState) + return ret0 +} + +// ConnectionState indicates an expected call of ConnectionState +func (mr *MockPeerConnectionMockRecorder) ConnectionState() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ConnectionState", reflect.TypeOf((*MockPeerConnection)(nil).ConnectionState)) +} + +// RemoveTrack mocks base method +func (m *MockPeerConnection) RemoveTrack(sender *webrtc.RTPSender) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "RemoveTrack", sender) + ret0, _ := ret[0].(error) + return ret0 +} + +// RemoveTrack indicates an expected call of RemoveTrack +func (mr *MockPeerConnectionMockRecorder) RemoveTrack(sender interface{}) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RemoveTrack", reflect.TypeOf((*MockPeerConnection)(nil).RemoveTrack), sender) +} diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 66f74f5d7..757c4613c 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -25,7 +25,7 @@ const ( type Participant struct { id string - peerConn *webrtc.PeerConnection + peerConn PeerConnection sigConn SignalConnection ctx context.Context cancel context.CancelFunc @@ -36,10 +36,9 @@ type Participant struct { rtcpCh chan []rtcp.Packet downTracks map[string][]*sfu.DownTrack - lock sync.RWMutex - receiverConfig ReceiverConfig - tracks map[string]PublishedTrack // tracks that the peer is publishing - once sync.Once + lock sync.RWMutex + tracks map[string]PublishedTrack // tracks that the peer is publishing + once sync.Once // callbacks & handlers // OnTrackPublished - remote peer added a remoteTrack @@ -52,14 +51,16 @@ type Participant struct { OnClose func(*Participant) } -func NewParticipant(conf *WebRTCConfig, sc SignalConnection, name string) (*Participant, error) { +func NewPeerConnection(conf *WebRTCConfig) (*webrtc.PeerConnection, error) { me := &webrtc.MediaEngine{} me.RegisterDefaultCodecs() api := webrtc.NewAPI(webrtc.WithMediaEngine(me), webrtc.WithSettingEngine(conf.SettingEngine)) - pc, err := api.NewPeerConnection(conf.Configuration) - if err != nil { - return nil, err - } + return api.NewPeerConnection(conf.Configuration) +} + +func NewParticipant(pc PeerConnection, sc SignalConnection, name string) (*Participant, error) { + me := &webrtc.MediaEngine{} + me.RegisterDefaultCodecs() bi := buffer.NewBufferInterceptor() ir := &interceptor.Registry{} @@ -67,20 +68,19 @@ func NewParticipant(conf *WebRTCConfig, sc SignalConnection, name string) (*Part ctx, cancel := context.WithCancel(context.Background()) participant := &Participant{ - id: utils.NewGuid(utils.ParticipantPrefix), - name: name, - peerConn: pc, - sigConn: sc, - ctx: ctx, - cancel: cancel, - bi: bi, - rtcpCh: make(chan []rtcp.Packet, 10), - downTracks: make(map[string][]*sfu.DownTrack), - state: livekit.ParticipantInfo_JOINING, - lock: sync.RWMutex{}, - receiverConfig: conf.receiver, - tracks: make(map[string]PublishedTrack, 0), - mediaEngine: me, + id: utils.NewGuid(utils.ParticipantPrefix), + name: name, + peerConn: pc, + sigConn: sc, + ctx: ctx, + cancel: cancel, + bi: bi, + rtcpCh: make(chan []rtcp.Packet, 10), + downTracks: make(map[string][]*sfu.DownTrack), + state: livekit.ParticipantInfo_JOINING, + lock: sync.RWMutex{}, + tracks: make(map[string]PublishedTrack, 0), + mediaEngine: me, } log := logger.GetLogger() diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 0914c8e96..be1920c12 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -70,20 +70,20 @@ func (r *Room) Join(participant *Participant) error { if oldState == livekit.ParticipantInfo_JOINING && p.state == livekit.ParticipantInfo_JOINED { // subscribe participant to existing tracks - for _, p := range r.participants { - if p.id == participant.id { + for _, op := range r.participants { + if p.id == op.id { // don't send to itself continue } - if err := p.AddSubscriber(participant); err != nil { + if err := op.AddSubscriber(p); err != nil { // TODO: log error? or disconnect? logger.GetLogger().Errorw("could not subscribe to participant", - "dstParticipant", participant.ID(), - "srcParticipant", p.ID()) + "dstParticipant", p.ID(), + "srcParticipant", op.ID()) } } // start the workers once connectivity is established - participant.Start() + p.Start() } } diff --git a/pkg/rtc/wsprotocol.go b/pkg/rtc/wsprotocol.go index de4460868..21cb2d26c 100644 --- a/pkg/rtc/wsprotocol.go +++ b/pkg/rtc/wsprotocol.go @@ -17,17 +17,6 @@ const ( pingTimeout = 2 * time.Second ) -type WebsocketClient interface { - ReadMessage() (messageType int, p []byte, err error) - WriteMessage(messageType int, data []byte) error - WriteControl(messageType int, data []byte, deadline time.Time) error -} - -type SignalConnection interface { - ReadRequest() (*livekit.SignalRequest, error) - WriteResponse(*livekit.SignalResponse) error -} - type WSSignalConnection struct { conn WebsocketClient mu sync.Mutex diff --git a/pkg/service/rtc.go b/pkg/service/rtc.go index 476832b1d..e5dbaf2b3 100644 --- a/pkg/service/rtc.go +++ b/pkg/service/rtc.go @@ -67,7 +67,12 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { }) signalConn := rtc.NewWSSignalConnection(conn) - participant, err := rtc.NewParticipant(s.manager.Config(), signalConn, pName) + pc, err := rtc.NewPeerConnection(s.manager.Config()) + if err != nil { + writeJSONError(w, http.StatusInternalServerError, "could not create peerConnection", err.Error()) + return + } + participant, err := rtc.NewParticipant(pc, signalConn, pName) if err != nil { writeJSONError(w, http.StatusInternalServerError, "could not create participant", err.Error()) return diff --git a/proto/model.proto b/proto/model.proto index b58b1bb9e..3a4a7061f 100644 --- a/proto/model.proto +++ b/proto/model.proto @@ -59,11 +59,3 @@ message TrackInfo { Type type = 2; string name = 3; } - - -message DataMessage { - oneof value { - string text = 1; - bytes binary = 2; - } -}