WIP

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,

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

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)

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)

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
+}

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):

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()
 		}
 	}

test/client/trackwriter.go

@@ -146,6 +146,7 @@ func (w *TrackWriter) writeOgg() {
 			return
 		}
 
+		logger.Infow("RAJA wrote sample") // REMOVE
 		time.Sleep(sampleDuration)
 	}
 }

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 {

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)
+}