diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index ec91ad118..d8d9d3d08 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1319,8 +1319,15 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { return } + if p.ProtocolVersion().SupportsSinglePeerConnection() { + if err := p.TransportManager.AddRemoteTrackAndNegotiate(ti); err != nil { + return + } + } + p.sendTrackPublished(req.Cid, ti) + // RAJA-TODO: is this needed for single peer connection case p.handlePendingRemoteTracks() } @@ -1850,6 +1857,10 @@ func (h SubscriberTransportHandler) OnOffer(sd webrtc.SessionDescription, offerI return h.p.onSubscriberOffer(sd, offerId) } +func (h SubscriberTransportHandler) OnTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) { + h.p.onMediaTrack(track, rtpReceiver) +} + func (h SubscriberTransportHandler) OnStreamStateChange(update *streamallocator.StreamStateUpdate) error { return h.p.onStreamStateChange(update) } @@ -1912,6 +1923,7 @@ func (p *ParticipantImpl) setupTransportManager() error { // primary connection does not change, canSubscribe can change if permission was updated // after the participant has joined SubscriberAsPrimary: subscriberAsPrimary, + SinglePeerConnection: p.ProtocolVersion().SupportsSinglePeerConnection(), Config: p.params.Config, Twcc: p.twcc, ProtocolVersion: p.params.ProtocolVersion, @@ -2118,6 +2130,43 @@ func (p *ParticipantImpl) setIsPublisher(isPublisher bool) { // when the server has an offer for participant func (p *ParticipantImpl) onSubscriberOffer(offer webrtc.SessionDescription, offerId uint32) error { + if p.ProtocolVersion().SupportsSinglePeerConnection() { + parsedOffer, err := offer.Unmarshal() + if err != nil { + p.pubLogger.Warnw( + "could not parse offer", err, + "transport", livekit.SignalTarget_PUBLISHER, + "offer", offer, + "offerId", offerId, + ) + return err + } + + unmatchAudios, err := p.TransportManager.GetUnmatchMediaForOffer(parsedOffer, "audio") + if err != nil { + p.subLogger.Warnw("could not get unmatch audios", err) + return err + } + unmatchVideos, _ := p.TransportManager.GetUnmatchMediaForOffer(parsedOffer, "video") + if err != nil { + p.subLogger.Warnw("could not get unmatch videos", err) + return err + } + parsedOffer = p.setCodecPreferencesForPublisher(parsedOffer, unmatchAudios, unmatchVideos) + + // put together munged offer after setting codec preferences + bytes, err := parsedOffer.Marshal() + if err != nil { + p.pubLogger.Errorw("failed to marshal offer", err) + return err + } + + offer = webrtc.SessionDescription{ + Type: offer.Type, + SDP: string(bytes), + } + } + p.subLogger.Debugw( "sending offer", "transport", livekit.SignalTarget_SUBSCRIBER, diff --git a/pkg/rtc/participant_sdp.go b/pkg/rtc/participant_sdp.go index cc95667df..d7b003866 100644 --- a/pkg/rtc/participant_sdp.go +++ b/pkg/rtc/participant_sdp.go @@ -25,6 +25,7 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu/mime" "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" lksdp "github.com/livekit/protocol/sdp" "github.com/livekit/protocol/utils" ) @@ -244,6 +245,7 @@ func (p *ParticipantImpl) setCodecPreferencesForPublisherMedia( unprocessed := make([]*sdp.MediaDescription, 0, len(unmatches)) // unmatched media is pending for publish, set codec preference for _, unmatch := range unmatches { + /* RAJA-TODO streamID, ok := lksdp.ExtractStreamID(unmatch) if !ok { unprocessed = append(unprocessed, unmatch) @@ -258,7 +260,10 @@ func (p *ParticipantImpl) setCodecPreferencesForPublisherMedia( } else { _, info, _, _, _ = p.getPendingTrack(streamID, trackType, false) } + */ + p.pendingTracksLock.RLock() + _, info, _, _, _ := p.getPendingTrack("junk", trackType, false) // RAJA-TODO if info == nil { p.pendingTracksLock.RUnlock() unprocessed = append(unprocessed, unmatch) @@ -266,15 +271,18 @@ func (p *ParticipantImpl) setCodecPreferencesForPublisherMedia( } var mimeType string + /* RAJA-TODO for _, c := range info.Codecs { if c.Cid == streamID || c.SdpCid == streamID { mimeType = c.MimeType break } } + */ if mimeType == "" && len(info.Codecs) > 0 { mimeType = info.Codecs[0].MimeType } + p.params.Logger.Infow("RAJA mimeType", "mimeType", mimeType, "info", logger.Proto(info)) // REMOVE p.pendingTracksLock.RUnlock() if mimeType == "" { @@ -334,6 +342,9 @@ func (p *ParticipantImpl) setCodecPreferencesForPublisherMedia( unmatch.MediaName.Formats = append(unmatch.MediaName.Formats, leftCodecs...) } + + // RAJA-TODO: can set mid in TrackInfo before sending offer and check on receiving answer to set SdpCid + // RAJA-TODO: Have to check which simulcast codec is not published } return parsedOffer, unprocessed diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index e4af28af2..9fb51a12c 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -101,6 +101,7 @@ var ( ErrNoBundleMid = errors.New("could not get bundle mid") ErrMidMismatch = errors.New("media mid does not match bundle mid") ErrICECredentialMismatch = errors.New("ice credential mismatch") + ErrInvalidCodecIndex = errors.New("invalid codec index") ) // ------------------------------------------------------------------------- @@ -244,6 +245,10 @@ type PCTransport struct { remoteOfferId atomic.Uint32 localAnswerId atomic.Uint32 + // RAJA-TODO: check if this can be gotten from pc + lastLocalDescription atomic.Value + lastRemoteDescription atomic.Value + eventsQueue *utils.TypedOpsQueue[event] // the following should be accessed only in event processing go routine @@ -970,10 +975,98 @@ func (t *PCTransport) AddTransceiverFromTrack(trackLocal webrtc.TrackLocal, para return } +func (t *PCTransport) AddRemoteTrackAndNegotiate(ti *livekit.TrackInfo) error { + if ti == nil { + return nil + } + + /* RAJA-REMOVE + if codecIdx >= len(ti.Codecs) { + t.params.Logger.Warnw( + "invalid codec index", ErrInvalidCodecIndex, + "trackID", ti.Sid, + "track", logger.Proto(ti), + ) + return ErrInvalidCodecIndex + } + + sci := ti.Codecs[codecIdx] + var mimeType mime.MimeType + if sci.MimeType == "" { + if ti.Type == livekit.TrackType_AUDIO { + mimeType = mime.MimeTypeOpus + } else { + mimeType = mime.MimeTypeVP8 + } + } else { + mimeType = mime.MimeTypeFromString(sci.MimeType) + } + track, err := webrtc.NewTrackLocalStaticRTP( + webrtc.RTPCodecCapability{ + MimeType: mimeType.String(), + }, + sci.Cid, + sci.Cid, + ) + if err != nil { + t.params.Logger.Warnw( + "could not create local track", err, + "trackID", ti.Sid, + "track", logger.Proto(ti), + "sci", logger.Proto(sci), + ) + return err + } + + _, err = t.pc.AddTransceiverFromTrack( + track, + webrtc.RTPTransceiverInit{ + Direction: webrtc.RTPTransceiverDirectionSendrecv, + }, + ) + if err != nil { + t.params.Logger.Warnw( + "could not add transceiver from track", err, + "trackID", ti.Sid, + "track", logger.Proto(ti), + "sci", logger.Proto(sci), + ) + return err + } + */ + + rtpCodecType := webrtc.RTPCodecTypeVideo + if ti.Type == livekit.TrackType_AUDIO { + rtpCodecType = webrtc.RTPCodecTypeAudio + } + _, err := t.pc.AddTransceiverFromKind( + rtpCodecType, + webrtc.RTPTransceiverInit{ + Direction: webrtc.RTPTransceiverDirectionRecvonly, + }, + ) + if err != nil { + t.params.Logger.Warnw( + "could not add transceiver from kind", err, + "trackID", ti.Sid, + "track", logger.Proto(ti), + ) + return err + } + + t.Negotiate(true) + + return nil +} + func (t *PCTransport) RemoveTrack(sender *webrtc.RTPSender) error { return t.pc.RemoveTrack(sender) } +func (t *PCTransport) CurrentRemoteDescription() *webrtc.SessionDescription { + return t.pc.CurrentRemoteDescription() +} + func (t *PCTransport) GetMid(rtpReceiver *webrtc.RTPReceiver) string { for _, tr := range t.pc.GetTransceivers() { if tr.Receiver() == rtpReceiver { @@ -2258,6 +2351,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { ) } + t.lastLocalDescription.Store(offer) if err := t.params.Handler.OnOffer(offer, t.localOfferId.Inc()); err != nil { prometheus.ServiceOperationCounter.WithLabelValues("offer", "error", "write_message").Add(1) return errors.Wrap(err, "could not send offer") @@ -2278,6 +2372,7 @@ type remoteDescriptionData struct { func (t *PCTransport) handleRemoteDescriptionReceived(e event) error { rdd := e.data.(remoteDescriptionData) + t.lastRemoteDescription.Store(*rdd.sessionDescription) if rdd.sessionDescription.Type == webrtc.SDPTypeOffer { return t.handleRemoteOfferReceived(rdd.sessionDescription, rdd.remoteId) } else { @@ -2384,6 +2479,7 @@ func (t *PCTransport) createAndSendAnswer() error { ) } + t.lastLocalDescription.Store(answer) answerId := t.remoteOfferId.Load() if err := t.params.Handler.OnAnswer(answer, answerId); err != nil { prometheus.ServiceOperationCounter.WithLabelValues("answer", "error", "write_message").Add(1) @@ -2549,6 +2645,7 @@ func (t *PCTransport) doICERestart() error { ) } + t.lastLocalDescription.Store(*offer) err := t.params.Handler.OnOffer(*offer, t.localOfferId.Inc()) if err != nil { prometheus.ServiceOperationCounter.WithLabelValues("offer", "error", "write_message").Add(1) diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index a064fdf4c..df97dd4b6 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -84,6 +84,7 @@ func (h TransportManagerPublisherTransportHandler) OnAnswer(sd webrtc.SessionDes type TransportManagerParams struct { SubscriberAsPrimary bool + SinglePeerConnection bool Config *WebRTCConfig Twcc *twcc.Responder ProtocolVersion types.ProtocolVersion @@ -124,9 +125,10 @@ type TransportManager struct { pendingOfferPublisher *webrtc.SessionDescription pendingOfferIdPublisher uint32 pendingDataChannelsPublisher []*livekit.DataChannelInfo - lastPublisherAnswer atomic.Value - lastPublisherOffer atomic.Value - iceConfig *livekit.ICEConfig + // RAJA-TODO: check if this can be gotten from peer connection directly + lastPublisherAnswer atomic.Value + lastPublisherOffer atomic.Value + iceConfig *livekit.ICEConfig mediaLossProxy *MediaLossProxy udpLossUnstableCount uint32 @@ -148,49 +150,59 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro } t.mediaLossProxy.OnMediaLossUpdate(t.onMediaLossUpdate) - lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER) - publisher, err := NewPCTransport(TransportParams{ - ProtocolVersion: params.ProtocolVersion, - Config: params.Config, - Twcc: params.Twcc, - DirectionConfig: params.Config.Publisher, - CongestionControlConfig: params.CongestionControlConfig, - EnabledCodecs: params.EnabledPublishCodecs, - Logger: lgr, - SimTracks: params.SimTracks, - ClientInfo: params.ClientInfo, - Transport: livekit.SignalTarget_PUBLISHER, - Handler: TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t, lgr}}, - UseOneShotSignallingMode: params.UseOneShotSignallingMode, - DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, - DatachannelSlowThreshold: params.DatachannelSlowThreshold, - FireOnTrackBySdp: params.FireOnTrackBySdp, - }) - if err != nil { - return nil, err + if t.params.UseOneShotSignallingMode || !t.params.SinglePeerConnection { + lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER) + publisher, err := NewPCTransport(TransportParams{ + ProtocolVersion: params.ProtocolVersion, + Config: params.Config, + Twcc: params.Twcc, + DirectionConfig: params.Config.Publisher, + CongestionControlConfig: params.CongestionControlConfig, + EnabledCodecs: params.EnabledPublishCodecs, + Logger: lgr, + SimTracks: params.SimTracks, + ClientInfo: params.ClientInfo, + Transport: livekit.SignalTarget_PUBLISHER, + Handler: TransportManagerPublisherTransportHandler{TransportManagerTransportHandler{params.PublisherHandler, t, lgr}}, + UseOneShotSignallingMode: params.UseOneShotSignallingMode, + DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, + DatachannelSlowThreshold: params.DatachannelSlowThreshold, + FireOnTrackBySdp: params.FireOnTrackBySdp, + }) + if err != nil { + return nil, err + } + t.publisher = publisher } - t.publisher = publisher - lgr = LoggerWithPCTarget(params.Logger, livekit.SignalTarget_SUBSCRIBER) - subscriber, err := NewPCTransport(TransportParams{ - ProtocolVersion: params.ProtocolVersion, - Config: params.Config, - DirectionConfig: params.Config.Subscriber, - CongestionControlConfig: params.CongestionControlConfig, - EnabledCodecs: params.EnabledSubscribeCodecs, - Logger: lgr, - ClientInfo: params.ClientInfo, - IsOfferer: true, - IsSendSide: true, - AllowPlayoutDelay: params.AllowPlayoutDelay, - DatachannelSlowThreshold: params.DatachannelSlowThreshold, - Transport: livekit.SignalTarget_SUBSCRIBER, - Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, - }) - if err != nil { - return nil, err + if !t.params.UseOneShotSignallingMode { + lgr := LoggerWithPCTarget(params.Logger, livekit.SignalTarget_SUBSCRIBER) + subscriber, err := NewPCTransport(TransportParams{ + ProtocolVersion: params.ProtocolVersion, + Config: params.Config, + DirectionConfig: params.Config.Subscriber, + CongestionControlConfig: params.CongestionControlConfig, + EnabledCodecs: params.EnabledSubscribeCodecs, + Logger: lgr, + ClientInfo: params.ClientInfo, + IsOfferer: true, + IsSendSide: true, + AllowPlayoutDelay: params.AllowPlayoutDelay, + DatachannelSlowThreshold: params.DatachannelSlowThreshold, + Transport: livekit.SignalTarget_SUBSCRIBER, + Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, + FireOnTrackBySdp: params.FireOnTrackBySdp, + }) + if err != nil { + return nil, err + } + t.subscriber = subscriber + } else { + t.subscriber = t.publisher + } + if t.params.SinglePeerConnection { + t.publisher = t.subscriber } - t.subscriber = subscriber if !t.params.Migration { if err := t.createDataChannelsForSubscriber(nil); err != nil { return nil, err @@ -264,6 +276,10 @@ func (t *TransportManager) AddTransceiverFromTrackLocal( } } +func (t *TransportManager) AddRemoteTrackAndNegotiate(ti *livekit.TrackInfo) error { + return t.subscriber.AddRemoteTrackAndNegotiate(ti) +} + func (t *TransportManager) RemoveTrackLocal(sender *webrtc.RTPSender) error { if t.params.UseOneShotSignallingMode { return t.publisher.RemoveTrack(sender) @@ -395,10 +411,12 @@ func (t *TransportManager) createDataChannelsForSubscriber(pendingDataChannels [ func (t *TransportManager) GetUnmatchMediaForOffer(parsedOffer *sdp.SessionDescription, mediaType string) (unmatched []*sdp.MediaDescription, err error) { // prefer codec from offer for clients that don't support setCodecPreferences var lastMatchedMid string - lastAnswer := t.lastPublisherAnswer.Load() + // RAJA-TODO lastAnswer := t.lastPublisherAnswer.Load() + lastAnswer := t.publisher.CurrentRemoteDescription() if lastAnswer != nil { - answer := lastAnswer.(webrtc.SessionDescription) - parsedAnswer, err1 := answer.Unmarshal() + // RAJA-TODO answer := lastAnswer.(webrtc.SessionDescription) + // RAJA-TODO parsedAnswer, err1 := answer.Unmarshal() + parsedAnswer, err1 := lastAnswer.Unmarshal() if err1 != nil { // should not happen t.params.Logger.Errorw("failed to parse last answer", err1) diff --git a/pkg/rtc/types/protocol_version.go b/pkg/rtc/types/protocol_version.go index f1c54a7cb..36fe45527 100644 --- a/pkg/rtc/types/protocol_version.go +++ b/pkg/rtc/types/protocol_version.go @@ -16,7 +16,7 @@ package types type ProtocolVersion int -const CurrentProtocol = 16 +const CurrentProtocol = 17 func (v ProtocolVersion) SupportsPackedStreamId() bool { return v > 0 @@ -99,3 +99,7 @@ func (v ProtocolVersion) SupportsNonErrorSignalResponse() bool { func (v ProtocolVersion) SupportsMoving() bool { return v > 15 } + +func (v ProtocolVersion) SupportsSinglePeerConnection() bool { + return v > 16 +} diff --git a/pkg/sfu/mime/mimetype.go b/pkg/sfu/mime/mimetype.go index cada587ae..9d2ac387e 100644 --- a/pkg/sfu/mime/mimetype.go +++ b/pkg/sfu/mime/mimetype.go @@ -267,6 +267,41 @@ func (m MimeType) ReporterType() roomobs.MimeType { return roomobs.MimeTypeUndefined } +func MimeTypeFromString(mimeTypeString string) MimeType { + switch mimeTypeString { + case "MimeTypeUnknown": + return MimeTypeUnknown + case webrtc.MimeTypeH264: + return MimeTypeH264 + case webrtc.MimeTypeH265: + return MimeTypeH265 + case webrtc.MimeTypeOpus: + return MimeTypeOpus + case "audio/red": + return MimeTypeRED + case webrtc.MimeTypeVP8: + return MimeTypeVP8 + case webrtc.MimeTypeVP9: + return MimeTypeVP9 + case webrtc.MimeTypeAV1: + return MimeTypeAV1 + case webrtc.MimeTypeG722: + return MimeTypeG722 + case webrtc.MimeTypePCMU: + return MimeTypePCMU + case webrtc.MimeTypePCMA: + return MimeTypePCMA + case webrtc.MimeTypeRTX: + return MimeTypeRTX + case webrtc.MimeTypeFlexFEC: + return MimeTypeFlexFEC + case "video/ulpfec": + return MimeTypeULPFEC + } + + return MimeTypeUnknown +} + func NormalizeMimeType(mime string) MimeType { switch { case strings.EqualFold(mime, webrtc.MimeTypeH264): diff --git a/test/client/client.go b/test/client/client.go index c364f3d0c..a1cec9402 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -356,9 +356,10 @@ func NewRTCClient(conn *websocket.Conn, opts *Options) (*RTCClient, error) { }) subscriberHandler.OnAnswerCalls(func(answer webrtc.SessionDescription, answerId uint32) error { // send remote an answer - logger.Infow("sending subscriber answer", + logger.Infow( + "sending subscriber answer", "participant", c.localParticipant.Identity, - // "sdp", answer, + "sdp", answer, ) return c.SendRequest(&livekit.SignalRequest{ Message: &livekit.SignalRequest_Answer{ @@ -433,10 +434,13 @@ func (c *RTCClient) handleSignalResponse(res *livekit.SignalResponse) error { // "answer", msg.Answer.Sdp) c.handleAnswer(signalling.FromProtoSessionDescription(msg.Answer)) case *livekit.SignalResponse_Offer: - logger.Infow("received server offer", - "participant", c.localParticipant.Identity, - ) desc, offerId := signalling.FromProtoSessionDescription(msg.Offer) + logger.Infow( + "received server offer", + "participant", c.localParticipant.Identity, + "sdp", desc, + "offerId", offerId, + ) c.handleOffer(desc, offerId) case *livekit.SignalResponse_Trickle: candidateInit, err := signalling.FromProtoTrickle(msg.Trickle) @@ -658,7 +662,8 @@ func (c *RTCClient) hasPrimaryEverConnected() bool { } type AddTrackParams struct { - NoWriter bool + NoWriter bool + UseSubscriberPeerConnection bool } type AddTrackOption func(params *AddTrackParams) @@ -669,6 +674,12 @@ func AddTrackNoWriter() AddTrackOption { } } +func AddTrackUseSubscriberPeerConnectionr() AddTrackOption { + return func(params *AddTrackParams) { + params.UseSubscriberPeerConnection = true + } +} + func (c *RTCClient) AddTrack(track *webrtc.TrackLocalStaticSample, path string, opts ...AddTrackOption) (writer *TrackWriter, err error) { var params AddTrackParams for _, opt := range opts { @@ -679,7 +690,30 @@ func (c *RTCClient) AddTrack(track *webrtc.TrackLocalStaticSample, path string, trackType = livekit.TrackType_VIDEO } - if err = c.SendAddTrack(track.ID(), track.StreamID(), trackType); err != nil { + var sender *webrtc.RTPSender + if !params.UseSubscriberPeerConnection { + sender, _, err = c.publisher.AddTrack(track, types.AddTrackParams{}) + } else { + sender, _, err = c.subscriber.AddTrack(track, types.AddTrackParams{}) + } + if err != nil { + logger.Errorw( + "add track failed", err, + "participant", c.localParticipant.Identity, + "pID", c.localParticipant.Sid, + "trackID", track.ID(), + ) + return + } + logger.Infow( + "RAJA add track sender", + "participant", c.localParticipant.Identity, + "pID", c.localParticipant.Sid, + "trackID", track.ID(), + "sender", sender.GetParameters(), + ) // REMOVE + + if err = c.SendAddTrack(track.ID(), track.Codec().MimeType, track.StreamID(), trackType); err != nil { return } @@ -707,14 +741,35 @@ func (c *RTCClient) AddTrack(track *webrtc.TrackLocalStaticSample, path string, c.lock.Lock() defer c.lock.Unlock() - sender, _, err := c.publisher.AddTrack(track, types.AddTrackParams{}) + /* RAJA-TODO + var sender *webrtc.RTPSender + if !params.UseSubscriberPeerConnection { + sender, _, err = c.publisher.AddTrack(track, types.AddTrackParams{}) + } else { + sender, _, err = c.subscriber.AddTrack(track, types.AddTrackParams{}) + } if err != nil { - logger.Errorw("add track failed", err, "trackID", ti.Sid, "participant", c.localParticipant.Identity, "pID", c.localParticipant.Sid) + logger.Errorw( + "add track failed", err, + "participant", c.localParticipant.Identity, + "pID", c.localParticipant.Sid, + "trackID", ti.Sid, + ) return } + logger.Infow( + "RAJA add track sender", + "participant", c.localParticipant.Identity, + "pID", c.localParticipant.Sid, + "trackID", ti.Sid, + "sender", sender.GetParameters(), + ) // REMOVE + */ c.localTracks[ti.Sid] = track c.trackSenders[ti.Sid] = sender - c.publisher.Negotiate(false) + if !params.UseSubscriberPeerConnection { + c.publisher.Negotiate(false) + } if !params.NoWriter { writer = NewTrackWriter(c.ctx, track, path) @@ -767,13 +822,19 @@ func (c *RTCClient) AddFileTrack(path string, id string, label string) (writer * } // send AddTrack command to server to initiate server-side negotiation -func (c *RTCClient) SendAddTrack(cid string, name string, trackType livekit.TrackType) error { +func (c *RTCClient) SendAddTrack(cid string, mimeType string, name string, trackType livekit.TrackType) error { return c.SendRequest(&livekit.SignalRequest{ Message: &livekit.SignalRequest_AddTrack{ AddTrack: &livekit.AddTrackRequest{ Cid: cid, Name: name, Type: trackType, + SimulcastCodecs: []*livekit.SimulcastCodec{ + { + Cid: cid, + Codec: mimeType, + }, + }, }, }, }) @@ -889,23 +950,26 @@ func (c *RTCClient) onOffer(offer webrtc.SessionDescription, offerId uint32) err func (c *RTCClient) processTrack(track *webrtc.TrackRemote) { lastUpdate := time.Time{} - pId, trackId := rtc.UnpackStreamID(track.StreamID()) - if trackId == "" { - trackId = livekit.TrackID(track.ID()) + pID, trackID := rtc.UnpackStreamID(track.StreamID()) + if trackID == "" { + trackID = livekit.TrackID(track.ID()) } c.lock.Lock() - c.subscribedTracks[pId] = append(c.subscribedTracks[pId], track) + c.subscribedTracks[pID] = append(c.subscribedTracks[pID], track) c.lock.Unlock() - logger.Infow("client added track", "participant", c.localParticipant.Identity, - "pID", pId, - "trackID", trackId, + logger.Infow( + "client added track", + "participant", c.localParticipant.Identity, + "pID", pID, + "trackID", trackID, "codec", track.Codec(), + "ssrc", track.SSRC(), ) defer func() { c.lock.Lock() - c.subscribedTracks[pId] = funk.Without(c.subscribedTracks[pId], track).([]*webrtc.TrackRemote) + c.subscribedTracks[pID] = funk.Without(c.subscribedTracks[pID], track).([]*webrtc.TrackRemote) c.lock.Unlock() }() @@ -923,14 +987,17 @@ func (c *RTCClient) processTrack(track *webrtc.TrackRemote) { continue } c.lock.Lock() - c.lastPackets[pId] = pkt - c.bytesReceived[pId] += uint64(pkt.MarshalSize()) + c.lastPackets[pID] = pkt + c.bytesReceived[pID] += uint64(pkt.MarshalSize()) c.lock.Unlock() numBytes += pkt.MarshalSize() if time.Since(lastUpdate) > 30*time.Second { - logger.Infow("consumed from participant", - "trackID", trackId, "pID", pId, - "size", numBytes) + logger.Infow( + "consumed from participant", + "pID", pID, + "trackID", trackID, + "size", numBytes, + ) lastUpdate = time.Now() } } diff --git a/test/client/trackwriter.go b/test/client/trackwriter.go index 803bf3348..07adbdd86 100644 --- a/test/client/trackwriter.go +++ b/test/client/trackwriter.go @@ -146,6 +146,7 @@ func (w *TrackWriter) writeOgg() { return } + logger.Infow("RAJA wrote sample") // REMOVE time.Sleep(sampleDuration) } } diff --git a/test/integration_helpers.go b/test/integration_helpers.go index abb6bc149..03266959d 100644 --- a/test/integration_helpers.go +++ b/test/integration_helpers.go @@ -149,7 +149,7 @@ func waitUntilConnected(t *testing.T, clients ...*testclient.RTCClient) { func createSingleNodeServer(configUpdater func(*config.Config)) *service.LivekitServer { var err error - conf, err := config.NewConfig("", true, nil, nil) + conf, err := config.NewConfig(`development: true`, true, nil, nil) if err != nil { panic(fmt.Sprintf("could not create config: %v", err)) } @@ -157,6 +157,7 @@ func createSingleNodeServer(configUpdater func(*config.Config)) *service.Livekit if configUpdater != nil { configUpdater(conf) } + conf.Logging.Level = "debug" currentNode, err := routing.NewLocalNode(conf) if err != nil { diff --git a/test/singlenode_test.go b/test/singlenode_test.go index fe76a8f94..7422af120 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -894,8 +894,8 @@ func TestFireTrackBySdp(t *testing.T) { { name: "js client could pub a/v tracks", codecs: []webrtc.RTPCodecCapability{ - {MimeType: "video/H264"}, - {MimeType: "audio/opus"}, + {MimeType: mime.MimeTypeH264.String()}, + {MimeType: mime.MimeTypeOpus.String()}, }, pubSDK: livekit.ClientInfo_JS, }, @@ -954,3 +954,52 @@ func TestFireTrackBySdp(t *testing.T) { }) } } + +func TestSinglePeerConnection(t *testing.T) { + _, finish := setupSingleNodeTest("TestSinglePeerConnection") + defer finish() + + c1 := createRTCClient("c1", defaultServerPort, &testclient.Options{ + ClientInfo: &livekit.ClientInfo{ + Sdk: livekit.ClientInfo_GO, + }, + }) + c2 := createRTCClient("c2", defaultServerPort, &testclient.Options{ + AutoSubscribe: true, + ClientInfo: &livekit.ClientInfo{ + Sdk: livekit.ClientInfo_GO, + }, + }) + waitUntilConnected(t, c1, c2) + defer func() { + c1.Stop() + c2.Stop() + }() + + codecs := []webrtc.RTPCodecCapability{ + {MimeType: mime.MimeTypeH264.String()}, + {MimeType: mime.MimeTypeOpus.String()}, + } + for _, codec := range codecs { + _, err := c1.AddStaticTrackWithCodec(codec, codec.MimeType, codec.MimeType, testclient.AddTrackUseSubscriberPeerConnectionr()) + require.NoError(t, err) + } + + require.Eventually(t, func() bool { + return len(c2.SubscribedTracks()[c1.ID()]) == len(codecs) + }, 5*time.Second, 10*time.Millisecond) + + var found int + for _, pubTrack := range c1.GetPublishedTrackIDs() { + t.Log("pub track", pubTrack) + tracks := c2.SubscribedTracks()[c1.ID()] + for _, track := range tracks { + t.Log("sub track", track.ID(), track.Codec()) + if track.ID() == pubTrack { + found++ + break + } + } + } + require.Equal(t, len(codecs), found) +}