diff --git a/go.mod b/go.mod index 671b64f3f..3f6dba843 100644 --- a/go.mod +++ b/go.mod @@ -23,7 +23,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded - github.com/livekit/protocol v1.39.4-0.20250718071209-86b0892a5d8c + github.com/livekit/protocol v1.39.4-0.20250720165053-c700929d2b5f github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c github.com/mackerelio/go-osstat v0.2.5 github.com/magefile/mage v1.15.0 @@ -55,7 +55,7 @@ require ( go.uber.org/atomic v1.11.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.27.0 - golang.org/x/exp v0.0.0-20250717185816-542afb5b7346 + golang.org/x/exp v0.0.0-20250718183923-645b1fa84792 golang.org/x/mod v0.26.0 golang.org/x/sync v0.16.0 google.golang.org/protobuf v1.36.6 diff --git a/go.sum b/go.sum index 781a2dbd1..2754287c4 100644 --- a/go.sum +++ b/go.sum @@ -167,8 +167,8 @@ github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731 h1:9x+U2HGLrSw5AT github.com/livekit/mageutil v0.0.0-20250511045019-0f1ff63f7731/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded h1:ylZPdnlX1RW9Z15SD4mp87vT2D2shsk0hpLJwSPcq3g= github.com/livekit/mediatransportutil v0.0.0-20250519131108-fb90f5acfded/go.mod h1:mSNtYzSf6iY9xM3UX42VEI+STHvMgHmrYzEHPcdhB8A= -github.com/livekit/protocol v1.39.4-0.20250718071209-86b0892a5d8c h1:3mCdD0e2zy71Q5hSbqQI/AtbcrQR0qf6yc+PFhs7VzI= -github.com/livekit/protocol v1.39.4-0.20250718071209-86b0892a5d8c/go.mod h1:6l+zgRJZ9sY96LM7DA3EMcKQC5zsVyZVP73c+9wgvCA= +github.com/livekit/protocol v1.39.4-0.20250720165053-c700929d2b5f h1:rdTSf5lLtnXbJENVntGKsasl7YNPEz+piA1T225af94= +github.com/livekit/protocol v1.39.4-0.20250720165053-c700929d2b5f/go.mod h1:6l+zgRJZ9sY96LM7DA3EMcKQC5zsVyZVP73c+9wgvCA= github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c h1:WwEr0YBejYbKzk8LSaO9h8h0G9MnE7shyDu8yXQWmEc= github.com/livekit/psrpc v0.6.1-0.20250511053145-465289d72c3c/go.mod h1:kmD+AZPkWu0MaXIMv57jhNlbiSZZ/Jx4bzlxBDVmJes= github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o= @@ -361,8 +361,8 @@ golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1m golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= golang.org/x/crypto v0.40.0 h1:r4x+VvoG5Fm+eJcxMaY8CQM7Lb0l1lsmjGBQ6s8BfKM= golang.org/x/crypto v0.40.0/go.mod h1:Qr1vMER5WyS2dfPHAlsOj01wgLbsyWtFn/aY+5+ZdxY= -golang.org/x/exp v0.0.0-20250717185816-542afb5b7346 h1:vuCObX8mQzik1tfEcYxWZBuVsmQtD1IjxCyPKM18Bh4= -golang.org/x/exp v0.0.0-20250717185816-542afb5b7346/go.mod h1:A+z0yzpGtvnG90cToK5n2tu8UJVP2XUATh+r+sfOOOc= +golang.org/x/exp v0.0.0-20250718183923-645b1fa84792 h1:R9PFI6EUdfVKgwKjZef7QIwGcBKu86OEFpJ9nUEP2l4= +golang.org/x/exp v0.0.0-20250718183923-645b1fa84792/go.mod h1:A+z0yzpGtvnG90cToK5n2tu8UJVP2XUATh+r+sfOOOc= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= diff --git a/pkg/rtc/errors.go b/pkg/rtc/errors.go index d66fc0076..bbe390683 100644 --- a/pkg/rtc/errors.go +++ b/pkg/rtc/errors.go @@ -32,9 +32,6 @@ var ( ErrEmptyParticipantID = errors.New("participant ID cannot be empty") ErrMissingGrants = errors.New("VideoGrant is missing") ErrInternalError = errors.New("internal error") - ErrNameExceedsLimits = errors.New("name length exceeds limits") - ErrMetadataExceedsLimits = errors.New("metadata size exceeds limits") - ErrAttributesExceedsLimits = errors.New("attributes size exceeds limits") // Track subscription related ErrNoTrackPermission = errors.New("participant is not allowed to subscribe to this track") diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 42bdd25ed..644eac62d 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -223,9 +223,6 @@ type ParticipantImpl struct { state atomic.Value // livekit.ParticipantInfo_State disconnected chan struct{} - resSinkMu sync.Mutex - resSink routing.MessageSink - grants atomic.Pointer[auth.ClaimGrants] isPublisher atomic.Bool @@ -315,8 +312,9 @@ type ParticipantImpl struct { metricsCollector *metric.MetricsCollector metricsReporter *metric.MetricsReporter - signalling signalling.ParticipantSignalling - signaller signalling.ParticipantSignaller + signalhandler signalling.ParticipantSignalHandler + signalling signalling.ParticipantSignalling + signaller signalling.ParticipantSignaller // loggers for publisher and subscriber pubLogger logger.Logger @@ -354,6 +352,10 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) { joiningMessageLastWrittenSeqs: make(map[livekit.ParticipantID]uint32), }, } + p.signalhandler = signalling.NewSignalHandler(signalling.SignalHandlerParams{ + Logger: params.Logger, + Participant: p, + }) p.signalling = signalling.NewSignalling(signalling.SignallingParams{ Logger: params.Logger, }) @@ -554,15 +556,15 @@ func (p *ParticipantImpl) CheckMetadataLimits( attributes map[string]string, ) error { if !p.params.LimitConfig.CheckParticipantNameLength(name) { - return ErrNameExceedsLimits + return signalling.ErrNameExceedsLimits } if !p.params.LimitConfig.CheckMetadataSize(metadata) { - return ErrMetadataExceedsLimits + return signalling.ErrMetadataExceedsLimits } if !p.params.LimitConfig.CheckAttributesSize(attributes) { - return ErrAttributesExceedsLimits + return signalling.ErrAttributesExceedsLimits } return nil @@ -3742,3 +3744,7 @@ func (p *ParticipantImpl) HandleLeaveRequest(reason types.ParticipantCloseReason onLeave(p, reason) } } + +func (p *ParticipantImpl) HandleSignalRequest(msg proto.Message) error { + return p.signalhandler.HandleRequest(msg) +} diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 3d0cf1dc3..3338d4c02 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -39,6 +39,7 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/routing/routingfakes" + "github.com/livekit/livekit-server/pkg/rtc/signalling" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" "github.com/livekit/livekit-server/pkg/testutils" @@ -376,7 +377,7 @@ func TestDisableCodecs(t *testing.T) { sink.WriteMessageCalls(func(msg proto.Message) error { if res, ok := msg.(*livekit.SignalResponse); ok { if res.GetAnswer() != nil { - answer, answerId = FromProtoSessionDescription(res.GetAnswer()) + answer, answerId = signalling.FromProtoSessionDescription(res.GetAnswer()) answerReceived.Store(true) answerIdReceived.Store(answerId) } @@ -525,7 +526,7 @@ func TestPreferVideoCodecForPublisher(t *testing.T) { sink.WriteMessageCalls(func(msg proto.Message) error { if res, ok := msg.(*livekit.SignalResponse); ok { if res.GetAnswer() != nil { - answer, answerId = FromProtoSessionDescription(res.GetAnswer()) + answer, answerId = signalling.FromProtoSessionDescription(res.GetAnswer()) pc.SetRemoteDescription(answer) answerReceived.Store(true) answerIdReceived.Store(answerId) @@ -613,7 +614,7 @@ func TestPreferAudioCodecForRed(t *testing.T) { sink.WriteMessageCalls(func(msg proto.Message) error { if res, ok := msg.(*livekit.SignalResponse); ok { if res.GetAnswer() != nil { - answer, answerId = FromProtoSessionDescription(res.GetAnswer()) + answer, answerId = signalling.FromProtoSessionDescription(res.GetAnswer()) pc.SetRemoteDescription(answer) answerReceived.Store(true) answerIdReceived.Store(answerId) diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 473134ae3..fd8aa3473 100644 --- a/pkg/rtc/participant_signal.go +++ b/pkg/rtc/participant_signal.go @@ -23,6 +23,7 @@ import ( "github.com/livekit/protocol/logger" "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/livekit-server/pkg/rtc/signalling" "github.com/livekit/livekit-server/pkg/rtc/types" "google.golang.org/protobuf/proto" @@ -221,7 +222,7 @@ func (p *ParticipantImpl) sendICECandidate(ic *webrtc.ICECandidate, target livek return nil } - trickle := ToProtoTrickle(prevIC.ToJSON(), target, ic == nil) + trickle := signalling.ToProtoTrickle(prevIC.ToJSON(), target, ic == nil) p.params.Logger.Debugw("sending ICE candidate", "transport", target, "trickle", logger.Proto(trickle)) return p.signaller.WriteMessage(p.signalling.SignalICECandidate(trickle)) @@ -293,11 +294,11 @@ func (p *ParticipantImpl) sendLeaveRequest( } func (p *ParticipantImpl) sendSdpAnswer(answer webrtc.SessionDescription, answerId uint32) error { - return p.signaller.WriteMessage(p.signalling.SignalSdpAnswer(ToProtoSessionDescription(answer, answerId))) + return p.signaller.WriteMessage(p.signalling.SignalSdpAnswer(signalling.ToProtoSessionDescription(answer, answerId))) } func (p *ParticipantImpl) sendSdpOffer(offer webrtc.SessionDescription, offerId uint32) error { - return p.signaller.WriteMessage(p.signalling.SignalSdpOffer(ToProtoSessionDescription(offer, offerId))) + return p.signaller.WriteMessage(p.signalling.SignalSdpOffer(signalling.ToProtoSessionDescription(offer, offerId))) } func (p *ParticipantImpl) sendStreamStateUpdate(streamStateUpdate *livekit.StreamStateUpdate) error { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index a2db7bd96..0f8d25e46 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -39,6 +39,7 @@ import ( "github.com/livekit/livekit-server/pkg/agent" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/livekit-server/pkg/rtc/signalling" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/sfu/buffer" @@ -796,7 +797,7 @@ func (r *Room) Joinv2( prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "get_subscriber_offer").Add(1) return nil, err } - connectResponse.SubscriberSdp = ToProtoSessionDescription(offer, 0) // SIGNALLING-V2-TODO - need to proper offerId? + connectResponse.SubscriberSdp = signalling.ToProtoSessionDescription(offer, 0) // SIGNALLING-V2-TODO - need to proper offerId? // for sync response, this does not actually send, only generates messageId and caches the message if err := participant.SendConnectResponse(connectResponse); err != nil { prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "send_response").Add(1) diff --git a/pkg/rtc/signalling/errors.go b/pkg/rtc/signalling/errors.go new file mode 100644 index 000000000..048b11732 --- /dev/null +++ b/pkg/rtc/signalling/errors.go @@ -0,0 +1,26 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package signalling + +import ( + "errors" +) + +var ( + ErrInvalidMessageType = errors.New("invalid message type") + ErrNameExceedsLimits = errors.New("name length exceeds limits") + ErrMetadataExceedsLimits = errors.New("metadata size exceeds limits") + ErrAttributesExceedsLimits = errors.New("attributes size exceeds limits") +) diff --git a/pkg/rtc/signalling/interfaces.go b/pkg/rtc/signalling/interfaces.go index db5bb42e1..1f8ac9a06 100644 --- a/pkg/rtc/signalling/interfaces.go +++ b/pkg/rtc/signalling/interfaces.go @@ -23,6 +23,10 @@ import ( "google.golang.org/protobuf/proto" ) +type ParticipantSignalHandler interface { + HandleRequest(msg proto.Message) error +} + type ParticipantSignaller interface { SetResponseSink(sink routing.MessageSink) GetResponseSink() routing.MessageSink diff --git a/pkg/rtc/signalling/signalfragment.go b/pkg/rtc/signalling/signalfragment.go index 911d4b38c..588284046 100644 --- a/pkg/rtc/signalling/signalfragment.go +++ b/pkg/rtc/signalling/signalfragment.go @@ -138,6 +138,7 @@ func (s *SignalFragment) Reassemble(fragment *livekit.Fragment) []byte { for _, fr := range re.fragments { data = append(data, fr.Data...) } + delete(s.reassemblies, fragment.PacketId) // fully re-assembled, can be deleted from cache return data } diff --git a/pkg/rtc/signalhandler.go b/pkg/rtc/signalling/signalhandler.go similarity index 56% rename from pkg/rtc/signalhandler.go rename to pkg/rtc/signalling/signalhandler.go index a348476a9..63ef6bbdd 100644 --- a/pkg/rtc/signalhandler.go +++ b/pkg/rtc/signalling/signalhandler.go @@ -12,43 +12,75 @@ // See the License for the specific language governing permissions and // limitations under the License. -package rtc +package signalling import ( + "fmt" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/pkg/rtc/types" ) -func HandleParticipantSignal(participant types.LocalParticipant, req *livekit.SignalRequest) error { - participant.UpdateLastSeenSignal() +var _ ParticipantSignalHandler = (*signalhandler)(nil) + +type SignalHandlerParams struct { + Logger logger.Logger + Participant types.LocalParticipant +} + +type signalhandler struct { + signalhandlerUnimplemented + + params SignalHandlerParams +} + +func NewSignalHandler(params SignalHandlerParams) ParticipantSignalHandler { + return &signalhandler{ + params: params, + } +} + +// SIGNALLING-V2-TODO: consolidate base message handling for messages common to different signalling versions +func (s *signalhandler) HandleRequest(msg proto.Message) error { + req, ok := msg.(*livekit.SignalRequest) + if !ok { + s.params.Logger.Warnw( + "unknown message type", nil, + "messageType", fmt.Sprintf("%T", msg), + ) + return ErrInvalidMessageType + } + s.params.Participant.UpdateLastSeenSignal() switch msg := req.GetMessage().(type) { case *livekit.SignalRequest_Offer: - participant.HandleOffer(FromProtoSessionDescription(msg.Offer)) + s.params.Participant.HandleOffer(FromProtoSessionDescription(msg.Offer)) case *livekit.SignalRequest_Answer: - participant.HandleAnswer(FromProtoSessionDescription(msg.Answer)) + s.params.Participant.HandleAnswer(FromProtoSessionDescription(msg.Answer)) case *livekit.SignalRequest_Trickle: candidateInit, err := FromProtoTrickle(msg.Trickle) if err != nil { - participant.GetLogger().Warnw("could not decode trickle", err) - return nil + s.params.Logger.Warnw("could not decode trickle", err) + return err } - participant.AddICECandidate(candidateInit, msg.Trickle.Target) + s.params.Participant.AddICECandidate(candidateInit, msg.Trickle.Target) case *livekit.SignalRequest_AddTrack: - participant.GetLogger().Debugw("add track request", "trackID", msg.AddTrack.Cid) - participant.AddTrack(msg.AddTrack) + s.params.Logger.Debugw("add track request", "trackID", msg.AddTrack.Cid) + s.params.Participant.AddTrack(msg.AddTrack) case *livekit.SignalRequest_Mute: - participant.SetTrackMuted(livekit.TrackID(msg.Mute.Sid), msg.Mute.Muted, false) + s.params.Participant.SetTrackMuted(livekit.TrackID(msg.Mute.Sid), msg.Mute.Muted, false) case *livekit.SignalRequest_Subscription: // allow participant to indicate their interest in the subscription // permission check happens later in SubscriptionManager - participant.HandleUpdateSubscriptions( + s.params.Participant.HandleUpdateSubscriptions( livekit.StringsAsIDs[livekit.TrackID](msg.Subscription.TrackSids), msg.Subscription.ParticipantTracks, msg.Subscription.Subscribe, @@ -56,7 +88,7 @@ func HandleParticipantSignal(participant types.LocalParticipant, req *livekit.Si case *livekit.SignalRequest_TrackSetting: for _, sid := range livekit.StringsAsIDs[livekit.TrackID](msg.TrackSetting.TrackSids) { - participant.UpdateSubscribedTrackSettings(sid, msg.TrackSetting) + s.params.Participant.UpdateSubscribedTrackSettings(sid, msg.TrackSetting) } case *livekit.SignalRequest_Leave: @@ -69,31 +101,31 @@ func HandleParticipantSignal(participant types.LocalParticipant, req *livekit.Si case livekit.DisconnectReason_USER_REJECTED: reason = types.ParticipantCloseReasonUserRejected } - participant.GetLogger().Debugw("client leaving room", "reason", reason) - participant.HandleLeaveRequest(reason) + s.params.Logger.Debugw("client leaving room", "reason", reason) + s.params.Participant.HandleLeaveRequest(reason) case *livekit.SignalRequest_SubscriptionPermission: - err := participant.HandleUpdateSubscriptionPermission(msg.SubscriptionPermission) + err := s.params.Participant.HandleUpdateSubscriptionPermission(msg.SubscriptionPermission) if err != nil { - participant.GetLogger().Warnw( + s.params.Logger.Warnw( "could not update subscription permission", err, "permissions", msg.SubscriptionPermission, ) } case *livekit.SignalRequest_SyncState: - err := participant.HandleSyncState(msg.SyncState) + err := s.params.Participant.HandleSyncState(msg.SyncState) if err != nil { - participant.GetLogger().Warnw( + s.params.Logger.Warnw( "could not sync state", err, "state", msg.SyncState, ) } case *livekit.SignalRequest_Simulate: - err := participant.HandleSimulateScenario(msg.Simulate) + err := s.params.Participant.HandleSimulateScenario(msg.Simulate) if err != nil { - participant.GetLogger().Warnw( + s.params.Logger.Warnw( "could not simulate scenario", err, "simulate", msg.Simulate, ) @@ -101,7 +133,7 @@ func HandleParticipantSignal(participant types.LocalParticipant, req *livekit.Si case *livekit.SignalRequest_PingReq: if msg.PingReq.Rtt > 0 { - participant.UpdateSignalingRTT(uint32(msg.PingReq.Rtt)) + s.params.Participant.UpdateSignalingRTT(uint32(msg.PingReq.Rtt)) } case *livekit.SignalRequest_UpdateMetadata: @@ -109,23 +141,23 @@ func HandleParticipantSignal(participant types.LocalParticipant, req *livekit.Si RequestId: msg.UpdateMetadata.RequestId, Reason: livekit.RequestResponse_OK, } - if participant.ClaimGrants().Video.GetCanUpdateOwnMetadata() { - if err := participant.CheckMetadataLimits( + if s.params.Participant.ClaimGrants().Video.GetCanUpdateOwnMetadata() { + if err := s.params.Participant.CheckMetadataLimits( msg.UpdateMetadata.Name, msg.UpdateMetadata.Metadata, msg.UpdateMetadata.Attributes, ); err == nil { if msg.UpdateMetadata.Name != "" { - participant.SetName(msg.UpdateMetadata.Name) + s.params.Participant.SetName(msg.UpdateMetadata.Name) } if msg.UpdateMetadata.Metadata != "" { - participant.SetMetadata(msg.UpdateMetadata.Metadata) + s.params.Participant.SetMetadata(msg.UpdateMetadata.Metadata) } if msg.UpdateMetadata.Attributes != nil { - participant.SetAttributes(msg.UpdateMetadata.Attributes) + s.params.Participant.SetAttributes(msg.UpdateMetadata.Attributes) } } else { - participant.GetLogger().Warnw("could not update metadata", err) + s.params.Logger.Warnw("could not update metadata", err) switch err { case ErrNameExceedsLimits: @@ -146,16 +178,16 @@ func HandleParticipantSignal(participant types.LocalParticipant, req *livekit.Si requestResponse.Reason = livekit.RequestResponse_NOT_ALLOWED requestResponse.Message = "does not have permission to update own metadata" } - participant.SendRequestResponse(requestResponse) + s.params.Participant.SendRequestResponse(requestResponse) case *livekit.SignalRequest_UpdateAudioTrack: - if err := participant.UpdateAudioTrack(msg.UpdateAudioTrack); err != nil { - participant.GetLogger().Warnw("could not update audio track", err, "update", msg.UpdateAudioTrack) + if err := s.params.Participant.UpdateAudioTrack(msg.UpdateAudioTrack); err != nil { + s.params.Logger.Warnw("could not update audio track", err, "update", msg.UpdateAudioTrack) } case *livekit.SignalRequest_UpdateVideoTrack: - if err := participant.UpdateVideoTrack(msg.UpdateVideoTrack); err != nil { - participant.GetLogger().Warnw("could not update video track", err, "update", msg.UpdateVideoTrack) + if err := s.params.Participant.UpdateVideoTrack(msg.UpdateVideoTrack); err != nil { + s.params.Logger.Warnw("could not update video track", err, "update", msg.UpdateVideoTrack) } } diff --git a/pkg/rtc/signalling/signalhandlerunimplemented.go b/pkg/rtc/signalling/signalhandlerunimplemented.go new file mode 100644 index 000000000..ebd76be05 --- /dev/null +++ b/pkg/rtc/signalling/signalhandlerunimplemented.go @@ -0,0 +1,27 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package signalling + +import ( + "google.golang.org/protobuf/proto" +) + +var _ ParticipantSignalHandler = (*signalhandlerUnimplemented)(nil) + +type signalhandlerUnimplemented struct{} + +func (u *signalhandlerUnimplemented) HandleRequest(msg proto.Message) error { + return nil +} diff --git a/pkg/rtc/signalling/signalhandlerv2.go b/pkg/rtc/signalling/signalhandlerv2.go new file mode 100644 index 000000000..633549793 --- /dev/null +++ b/pkg/rtc/signalling/signalhandlerv2.go @@ -0,0 +1,101 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package signalling + +import ( + "fmt" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "google.golang.org/protobuf/proto" + + "github.com/livekit/livekit-server/pkg/rtc/types" + + "go.uber.org/atomic" +) + +var _ ParticipantSignalHandler = (*signalhandlerv2)(nil) + +type SignalHandlerv2Params struct { + Logger logger.Logger + Participant types.LocalParticipant + Signalling ParticipantSignalling +} + +type signalhandlerv2 struct { + signalhandlerUnimplemented + + params SignalHandlerv2Params + + // SIGNALLING-V2-TODO: have to set this properly for `ConnectRequest` coming via sync HTTP path + lastProcessedRemoteMessageId atomic.Uint32 + signalFragment *SignalFragment +} + +func NewSignalHandlerv2(params SignalHandlerv2Params) ParticipantSignalHandler { + return &signalhandlerv2{ + params: params, + signalFragment: NewSignalFragment(SignalFragmentParams{ + Logger: params.Logger, + }), + } +} + +func (s *signalhandlerv2) HandleRequest(msg proto.Message) error { + req, ok := msg.(*livekit.Signalv2WireMessage) + if !ok { + s.params.Logger.Warnw( + "unknown message type", nil, + "messageType", fmt.Sprintf("%T", msg), + ) + return ErrInvalidMessageType + } + + // SIGNALLING-V2-TODO: check if this makes sense for data channel based signalling + s.params.Participant.UpdateLastSeenSignal() + + switch msg := req.GetMessage().(type) { + case *livekit.Signalv2WireMessage_Envelope: + for _, clientMessage := range msg.Envelope.ClientMessages { + if clientMessage.Sequencer.MessageId != s.lastProcessedRemoteMessageId.Load()+1 { + s.params.Logger.Infow( + "gap in message stream", + "last", s.lastProcessedRemoteMessageId.Load(), + "current", clientMessage.Sequencer.MessageId, + ) + } + + // SIGNALLING-V2-TODO: process messages + + s.params.Signalling.AckMessageId(clientMessage.Sequencer.LastProcessedRemoteMessageId) + s.params.Signalling.SetLastProcessedRemoteMessageId(clientMessage.Sequencer.MessageId) + } + + case *livekit.Signalv2WireMessage_Fragment: + bytes := s.signalFragment.Reassemble(msg.Fragment) + if len(bytes) != 0 { + wireMessage := &livekit.Signalv2WireMessage{} + err := proto.Unmarshal(bytes, wireMessage) + if err != nil { + s.params.Logger.Warnw("could not unmarshal re-assembled packet", err) + return err + } + + s.HandleRequest(wireMessage) + } + } + + return nil +} diff --git a/pkg/rtc/signalling/utils.go b/pkg/rtc/signalling/utils.go new file mode 100644 index 000000000..bbe71fdbf --- /dev/null +++ b/pkg/rtc/signalling/utils.go @@ -0,0 +1,66 @@ +// Copyright 2023 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package signalling + +import ( + "encoding/json" + + "github.com/livekit/protocol/livekit" + "github.com/pion/webrtc/v4" +) + +func ToProtoSessionDescription(sd webrtc.SessionDescription, id uint32) *livekit.SessionDescription { + return &livekit.SessionDescription{ + Type: sd.Type.String(), + Sdp: sd.SDP, + Id: id, + } +} + +func FromProtoSessionDescription(sd *livekit.SessionDescription) (webrtc.SessionDescription, uint32) { + var sdType webrtc.SDPType + switch sd.Type { + case webrtc.SDPTypeOffer.String(): + sdType = webrtc.SDPTypeOffer + case webrtc.SDPTypeAnswer.String(): + sdType = webrtc.SDPTypeAnswer + case webrtc.SDPTypePranswer.String(): + sdType = webrtc.SDPTypePranswer + case webrtc.SDPTypeRollback.String(): + sdType = webrtc.SDPTypeRollback + } + return webrtc.SessionDescription{ + Type: sdType, + SDP: sd.Sdp, + }, sd.Id +} + +func ToProtoTrickle(candidateInit webrtc.ICECandidateInit, target livekit.SignalTarget, final bool) *livekit.TrickleRequest { + data, _ := json.Marshal(candidateInit) + return &livekit.TrickleRequest{ + CandidateInit: string(data), + Target: target, + Final: final, + } +} + +func FromProtoTrickle(trickle *livekit.TrickleRequest) (webrtc.ICECandidateInit, error) { + ci := webrtc.ICECandidateInit{} + err := json.Unmarshal([]byte(trickle.CandidateInit), &ci) + if err != nil { + return webrtc.ICECandidateInit{}, err + } + return ci, nil +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index b18da362e..d1f814b94 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -532,6 +532,8 @@ type LocalParticipant interface { HandleSyncState(*livekit.SyncState) error HandleSimulateScenario(*livekit.SimulateScenario) error HandleLeaveRequest(reason ParticipantCloseReason) + + HandleSignalRequest(msg proto.Message) error } // Room is a container of participants, and can provide room-level actions diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index cc3c4df94..79bbb8e6f 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -574,6 +574,17 @@ type FakeLocalParticipant struct { handleReconnectAndSendResponseReturnsOnCall map[int]struct { result1 error } + HandleSignalRequestStub func(proto.Message) error + handleSignalRequestMutex sync.RWMutex + handleSignalRequestArgsForCall []struct { + arg1 proto.Message + } + handleSignalRequestReturns struct { + result1 error + } + handleSignalRequestReturnsOnCall map[int]struct { + result1 error + } HandleSignalSourceCloseStub func() handleSignalSourceCloseMutex sync.RWMutex handleSignalSourceCloseArgsForCall []struct { @@ -4313,6 +4324,67 @@ func (fake *FakeLocalParticipant) HandleReconnectAndSendResponseReturnsOnCall(i }{result1} } +func (fake *FakeLocalParticipant) HandleSignalRequest(arg1 proto.Message) error { + fake.handleSignalRequestMutex.Lock() + ret, specificReturn := fake.handleSignalRequestReturnsOnCall[len(fake.handleSignalRequestArgsForCall)] + fake.handleSignalRequestArgsForCall = append(fake.handleSignalRequestArgsForCall, struct { + arg1 proto.Message + }{arg1}) + stub := fake.HandleSignalRequestStub + fakeReturns := fake.handleSignalRequestReturns + fake.recordInvocation("HandleSignalRequest", []interface{}{arg1}) + fake.handleSignalRequestMutex.Unlock() + if stub != nil { + return stub(arg1) + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) HandleSignalRequestCallCount() int { + fake.handleSignalRequestMutex.RLock() + defer fake.handleSignalRequestMutex.RUnlock() + return len(fake.handleSignalRequestArgsForCall) +} + +func (fake *FakeLocalParticipant) HandleSignalRequestCalls(stub func(proto.Message) error) { + fake.handleSignalRequestMutex.Lock() + defer fake.handleSignalRequestMutex.Unlock() + fake.HandleSignalRequestStub = stub +} + +func (fake *FakeLocalParticipant) HandleSignalRequestArgsForCall(i int) proto.Message { + fake.handleSignalRequestMutex.RLock() + defer fake.handleSignalRequestMutex.RUnlock() + argsForCall := fake.handleSignalRequestArgsForCall[i] + return argsForCall.arg1 +} + +func (fake *FakeLocalParticipant) HandleSignalRequestReturns(result1 error) { + fake.handleSignalRequestMutex.Lock() + defer fake.handleSignalRequestMutex.Unlock() + fake.HandleSignalRequestStub = nil + fake.handleSignalRequestReturns = struct { + result1 error + }{result1} +} + +func (fake *FakeLocalParticipant) HandleSignalRequestReturnsOnCall(i int, result1 error) { + fake.handleSignalRequestMutex.Lock() + defer fake.handleSignalRequestMutex.Unlock() + fake.HandleSignalRequestStub = nil + if fake.handleSignalRequestReturnsOnCall == nil { + fake.handleSignalRequestReturnsOnCall = make(map[int]struct { + result1 error + }) + } + fake.handleSignalRequestReturnsOnCall[i] = struct { + result1 error + }{result1} +} + func (fake *FakeLocalParticipant) HandleSignalSourceClose() { fake.handleSignalSourceCloseMutex.Lock() fake.handleSignalSourceCloseArgsForCall = append(fake.handleSignalSourceCloseArgsForCall, struct { @@ -9236,6 +9308,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.handleReceiverReportMutex.RUnlock() fake.handleReconnectAndSendResponseMutex.RLock() defer fake.handleReconnectAndSendResponseMutex.RUnlock() + fake.handleSignalRequestMutex.RLock() + defer fake.handleSignalRequestMutex.RUnlock() fake.handleSignalSourceCloseMutex.RLock() defer fake.handleSignalSourceCloseMutex.RUnlock() fake.handleSimulateScenarioMutex.RLock() diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index bbf72067c..ed683840d 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -15,7 +15,6 @@ package rtc import ( - "encoding/json" "errors" "io" "net" @@ -80,50 +79,6 @@ func UnpackDataTrackLabel(packed string) (participantID livekit.ParticipantID, t return } -func ToProtoSessionDescription(sd webrtc.SessionDescription, id uint32) *livekit.SessionDescription { - return &livekit.SessionDescription{ - Type: sd.Type.String(), - Sdp: sd.SDP, - Id: id, - } -} - -func FromProtoSessionDescription(sd *livekit.SessionDescription) (webrtc.SessionDescription, uint32) { - var sdType webrtc.SDPType - switch sd.Type { - case webrtc.SDPTypeOffer.String(): - sdType = webrtc.SDPTypeOffer - case webrtc.SDPTypeAnswer.String(): - sdType = webrtc.SDPTypeAnswer - case webrtc.SDPTypePranswer.String(): - sdType = webrtc.SDPTypePranswer - case webrtc.SDPTypeRollback.String(): - sdType = webrtc.SDPTypeRollback - } - return webrtc.SessionDescription{ - Type: sdType, - SDP: sd.Sdp, - }, sd.Id -} - -func ToProtoTrickle(candidateInit webrtc.ICECandidateInit, target livekit.SignalTarget, final bool) *livekit.TrickleRequest { - data, _ := json.Marshal(candidateInit) - return &livekit.TrickleRequest{ - CandidateInit: string(data), - Target: target, - Final: final, - } -} - -func FromProtoTrickle(trickle *livekit.TrickleRequest) (webrtc.ICECandidateInit, error) { - ci := webrtc.ICECandidateInit{} - err := json.Unmarshal([]byte(trickle.CandidateInit), &ci) - if err != nil { - return webrtc.ICECandidateInit{}, err - } - return ci, nil -} - func ToProtoTrackKind(kind webrtc.RTPCodecType) livekit.TrackType { switch kind { case webrtc.RTPCodecTypeVideo: diff --git a/pkg/service/errors.go b/pkg/service/errors.go index 05c21be00..0ae8e9606 100644 --- a/pkg/service/errors.go +++ b/pkg/service/errors.go @@ -22,6 +22,7 @@ var ( ErrEgressNotFound = psrpc.NewErrorf(psrpc.NotFound, "egress does not exist") ErrEgressNotConnected = psrpc.NewErrorf(psrpc.Internal, "egress not connected (redis required)") ErrIdentityEmpty = psrpc.NewErrorf(psrpc.InvalidArgument, "identity cannot be empty") + ErrParticipantSidEmpty = psrpc.NewErrorf(psrpc.InvalidArgument, "participant sid cannot be empty") ErrIngressNotConnected = psrpc.NewErrorf(psrpc.Internal, "ingress not connected (redis required)") ErrIngressNotFound = psrpc.NewErrorf(psrpc.NotFound, "ingress does not exist") ErrIngressNonReusable = psrpc.NewErrorf(psrpc.InvalidArgument, "ingress is not reusable and cannot be modified") @@ -44,4 +45,5 @@ var ( ErrSIPTrunkNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip trunk does not exist") ErrSIPDispatchRuleNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip dispatch rule does not exist") ErrSIPParticipantNotFound = psrpc.NewErrorf(psrpc.NotFound, "requested sip participant does not exist") + ErrInvalidMessageType = psrpc.NewErrorf(psrpc.Internal, "invalid message type") ) diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index abf31a375..a13597e56 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -86,10 +86,11 @@ type RoomManager struct { rooms map[livekit.RoomName]*rtc.Room - roomServers utils.MultitonService[rpc.RoomTopic] - agentDispatchServers utils.MultitonService[rpc.RoomTopic] - participantServers utils.MultitonService[rpc.ParticipantTopic] - rtcRestParticipantServers utils.MultitonService[rpc.ParticipantTopic] + roomServers utils.MultitonService[rpc.RoomTopic] + agentDispatchServers utils.MultitonService[rpc.RoomTopic] + participantServers utils.MultitonService[rpc.ParticipantTopic] + httpSignalParticipantServers utils.MultitonService[rpc.ParticipantTopic] + rtcRestParticipantServers utils.MultitonService[rpc.ParticipantTopic] iceConfigCache *sutils.IceConfigCache[iceConfigCacheKey] @@ -238,6 +239,7 @@ func (r *RoomManager) Stop() { r.roomServers.Kill() r.agentDispatchServers.Kill() r.participantServers.Kill() + r.httpSignalParticipantServers.Kill() r.rtcRestParticipantServers.Kill() if r.rtcConfig != nil { @@ -708,6 +710,30 @@ func (r *RoomManager) HandleConnect( return nil, err } + var participantServerClosers utils.Closers + participantTopic := rpc.FormatParticipantTopic(room.Name(), participant.Identity()) + participantServer := must.Get(rpc.NewTypedParticipantServer(r, r.bus)) + participantServerClosers = append(participantServerClosers, utils.CloseFunc(r.participantServers.Replace(participantTopic, participantServer))) + if err := participantServer.RegisterAllParticipantTopics(participantTopic); err != nil { + participantServerClosers.Close() + pLogger.Errorw("could not join register participant topic", err) + _ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false) + return nil, err + } + + // SIGNALLING-V2-TODO: register HTTP + httpSignalParticipantServer := must.Get(rpc.NewTypedSignalv2ParticipantServer(signalv2ParticipantService{r}, r.bus)) + participantServerClosers = append( + participantServerClosers, + utils.CloseFunc(r.httpSignalParticipantServers.Replace(participantTopic, httpSignalParticipantServer)), + ) + if err := httpSignalParticipantServer.RegisterAllCommonTopics(participantTopic); err != nil { + participantServerClosers.Close() + pLogger.Errorw("could not join register participant topic for http signalling", err) + _ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false) + return nil, err + } + if err = r.roomStore.StoreParticipant(ctx, room.Name(), participant.ToProto()); err != nil { pLogger.Errorw("could not store participant", err) } @@ -898,8 +924,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.LocalPa return } - req := obj.(*livekit.SignalRequest) - if err := rtc.HandleParticipantSignal(participant, req); err != nil { + if err := participant.HandleSignalRequest(obj); err != nil { // more specific errors are already logged // treat errors returned as fatal return diff --git a/pkg/service/roommanager_service.go b/pkg/service/roommanager_service.go index a962d3fac..a8b42dfc9 100644 --- a/pkg/service/roommanager_service.go +++ b/pkg/service/roommanager_service.go @@ -107,6 +107,8 @@ func (s rtcRestService) Create(ctx context.Context, req *rpc.RTCRestCreateReques }, nil } +// ------------------------------------------- + type rtcRestParticipantService struct { *RoomManager } @@ -187,3 +189,40 @@ func (r rtcRestParticipantService) DeleteSession(ctx context.Context, req *rpc.R return &emptypb.Empty{}, nil } + +// -------------------------------- + +type signalv2ParticipantService struct { + *RoomManager +} + +func (s signalv2ParticipantService) RelaySignalv2Participant(ctx context.Context, req *rpc.RelaySignalv2ParticipantRequest) (*rpc.RelaySignalv2ParticipantResponse, error) { + room := s.RoomManager.GetRoom(ctx, livekit.RoomName(req.Room)) + if room == nil { + return nil, ErrRoomNotFound + } + + lp := room.GetParticipantByID(livekit.ParticipantID(req.ParticipantId)) + if lp == nil { + return nil, ErrParticipantNotFound + } + + err := lp.HandleSignalRequest(req.WireMessage) + if err != nil { + return nil, err + } + + var wireMessage *livekit.Signalv2WireMessage + pending := lp.SignalPendingMessages() + if pending != nil { + var ok bool + wireMessage, ok = pending.(*livekit.Signalv2WireMessage) + if !ok { + return nil, ErrInvalidMessageType + } + } + + return &rpc.RelaySignalv2ParticipantResponse{ + WireMessage: wireMessage, + }, nil +} diff --git a/pkg/service/rtcrestservice.go b/pkg/service/rtcrestservice.go index 47bd00442..9e4a5673a 100644 --- a/pkg/service/rtcrestservice.go +++ b/pkg/service/rtcrestservice.go @@ -453,7 +453,7 @@ func (s *RTCRestService) handleParticipantPatch(w http.ResponseWriter, r *http.R } pID := livekit.ParticipantID(r.PathValue("participant_id")) if pID == "" { - s.handleError("Patch", w, r, http.StatusUnauthorized, errors.New("participant ID cannot be empty")) + s.handleError("Patch", w, r, http.StatusBadRequest, errors.New("participant ID cannot be empty")) return } diff --git a/pkg/service/rtcv2service.go b/pkg/service/rtcv2service.go index b70942f4b..31d38b73e 100644 --- a/pkg/service/rtcv2service.go +++ b/pkg/service/rtcv2service.go @@ -23,9 +23,11 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/livekit-server/pkg/rtc" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/rpc" + "github.com/livekit/psrpc" "google.golang.org/protobuf/proto" ) @@ -35,7 +37,8 @@ var ( ) const ( - cRTCv2Path = "/rtc/v2" + cRTCv2Path = "/rtc/v2" + cRTCv2ParticipantIDPath = "/rtc/v2/{participant_id}" ) type RTCv2Service struct { @@ -44,22 +47,30 @@ type RTCv2Service struct { limits config.LimitConfig roomAllocator RoomAllocator router routing.MessageRouter + + topicFormatter rpc.TopicFormatter + signalv2ParticipantClient rpc.TypedSignalv2ParticipantClient } func NewRTCv2Service( config *config.Config, roomAllocator RoomAllocator, router routing.MessageRouter, + topicFormatter rpc.TopicFormatter, + signalv2ParticipantClient rpc.TypedSignalv2ParticipantClient, ) *RTCv2Service { return &RTCv2Service{ - limits: config.Limit, - router: router, - roomAllocator: roomAllocator, + limits: config.Limit, + router: router, + roomAllocator: roomAllocator, + topicFormatter: topicFormatter, + signalv2ParticipantClient: signalv2ParticipantClient, } } func (s *RTCv2Service) SetupRoutes(mux *http.ServeMux) { mux.HandleFunc("POST "+cRTCv2Path, s.handlePost) + mux.HandleFunc("PATCH "+cRTCv2ParticipantIDPath, s.handleParticipantPatch) } func (s *RTCv2Service) validateInternal( @@ -197,3 +208,98 @@ func (s *RTCv2Service) handlePost(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) } + +func (s *RTCv2Service) handleParticipantPatch(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Content-type") != "application/x-protobuf" { + HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("unsupported content-type: %s", r.Header.Get("Content-type"))) + return + } + + claims := GetGrants(r.Context()) + if claims == nil || claims.Video == nil { + HandleErrorJson(w, r, http.StatusUnauthorized, rtc.ErrPermissionDenied) + return + } + + roomName, err := EnsureJoinPermission(r.Context()) + if err != nil { + HandleErrorJson(w, r, http.StatusUnauthorized, err) + return + } + if roomName == "" { + HandleErrorJson(w, r, http.StatusUnauthorized, ErrNoRoomName) + return + } + + participantIdentity := livekit.ParticipantIdentity(claims.Identity) + if participantIdentity == "" { + HandleErrorJson(w, r, http.StatusUnauthorized, ErrIdentityEmpty) + return + } + + pID := livekit.ParticipantID(r.PathValue("participant_id")) + if pID == "" { + HandleErrorJson(w, r, http.StatusBadRequest, ErrParticipantSidEmpty) + return + } + + body, err := ioutil.ReadAll(r.Body) + if err != nil { + HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("could not read request body: %w", err)) + return + } + + wireMessage := &livekit.Signalv2WireMessage{} + err = proto.Unmarshal(body, wireMessage) + if err != nil { + HandleErrorJson(w, r, http.StatusBadRequest, fmt.Errorf("could not unmarshal request: %w", err)) + return + } + + res, err := s.signalv2ParticipantClient.RelaySignalv2Participant( + r.Context(), + s.topicFormatter.ParticipantTopic(r.Context(), roomName, participantIdentity), + &rpc.RelaySignalv2ParticipantRequest{ + Room: string(roomName), + ParticipantIdentity: string(participantIdentity), + ParticipantId: string(pID), + WireMessage: wireMessage, + }, + ) + if err != nil { + var pe psrpc.Error + if errors.As(err, &pe) { + switch pe.Code() { + case psrpc.NotFound: + HandleErrorJson(w, r, http.StatusNotFound, errors.New(pe.Error())) + + case psrpc.InvalidArgument: + HandleErrorJson(w, r, http.StatusBadRequest, errors.New(pe.Error())) + default: + HandleErrorJson(w, r, http.StatusInternalServerError, errors.New(pe.Error())) + } + } else { + HandleErrorJson(w, r, http.StatusInternalServerError, nil) + } + return + } + + logger.Debugw( + "participant response", + "room", roomName, + "participant", participantIdentity, + "pID", pID, + "participantResponse", logger.Proto(res), + ) + + marshalled, err := proto.Marshal(res.WireMessage) + if err != nil { + HandleErrorJson(w, r, http.StatusInternalServerError, err) + return + } + + w.Header().Add("Content-type", "application/x-protobuf") + w.Write(marshalled) + + w.WriteHeader(http.StatusOK) +} diff --git a/pkg/service/wire.go b/pkg/service/wire.go index d6c1945bb..7ac36a050 100644 --- a/pkg/service/wire.go +++ b/pkg/service/wire.go @@ -95,6 +95,7 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live rpc.NewTopicFormatter, rpc.NewTypedRoomClient, rpc.NewTypedParticipantClient, + rpc.NewTypedSignalv2ParticipantClient, rpc.NewTypedRTCRestParticipantClient, rpc.NewTypedAgentDispatchInternalClient, NewLocalRoomManager, diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 52a19fe8c..af1b4ba08 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -120,7 +120,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live } sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService) rtcService := NewRTCService(conf, roomAllocator, router, telemetryService) - rtCv2Service := NewRTCv2Service(conf, roomAllocator, router) + signalv2ParticipantClient, err := rpc.NewTypedSignalv2ParticipantClient(clientParams) + if err != nil { + return nil, err + } + rtCv2Service := NewRTCv2Service(conf, roomAllocator, router, topicFormatter, signalv2ParticipantClient) rtcRestParticipantClient, err := rpc.NewTypedRTCRestParticipantClient(clientParams) if err != nil { return nil, err diff --git a/test/client/client.go b/test/client/client.go index 5a921ecc1..acfd12916 100644 --- a/test/client/client.go +++ b/test/client/client.go @@ -41,6 +41,7 @@ import ( "github.com/livekit/protocol/logger" "github.com/livekit/livekit-server/pkg/rtc" + "github.com/livekit/livekit-server/pkg/rtc/signalling" "github.com/livekit/livekit-server/pkg/rtc/transport/transportfakes" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu/buffer" @@ -326,7 +327,7 @@ func NewRTCClient(conn *websocket.Conn, opts *Options) (*RTCClient, error) { ) return c.SendRequest(&livekit.SignalRequest{ Message: &livekit.SignalRequest_Answer{ - Answer: rtc.ToProtoSessionDescription(answer, answerId), + Answer: signalling.ToProtoSessionDescription(answer, answerId), }, }) }) @@ -395,15 +396,15 @@ func (c *RTCClient) handleSignalResponse(res *livekit.SignalResponse) error { // logger.Debugw("received server answer", // "participant", c.localParticipant.Identity, // "answer", msg.Answer.Sdp) - c.handleAnswer(rtc.FromProtoSessionDescription(msg.Answer)) + c.handleAnswer(signalling.FromProtoSessionDescription(msg.Answer)) case *livekit.SignalResponse_Offer: logger.Infow("received server offer", "participant", c.localParticipant.Identity, ) - desc, offerId := rtc.FromProtoSessionDescription(msg.Offer) + desc, offerId := signalling.FromProtoSessionDescription(msg.Offer) c.handleOffer(desc, offerId) case *livekit.SignalResponse_Trickle: - candidateInit, err := rtc.FromProtoTrickle(msg.Trickle) + candidateInit, err := signalling.FromProtoTrickle(msg.Trickle) if err != nil { return err } @@ -598,7 +599,7 @@ func (c *RTCClient) SendIceCandidate(ic *webrtc.ICECandidate, target livekit.Sig return c.SendRequest(&livekit.SignalRequest{ Message: &livekit.SignalRequest_Trickle{ - Trickle: rtc.ToProtoTrickle(prevIC.ToJSON(), target, ic == nil), + Trickle: signalling.ToProtoTrickle(prevIC.ToJSON(), target, ic == nil), }, }) } @@ -846,7 +847,7 @@ func (c *RTCClient) onOffer(offer webrtc.SessionDescription, offerId uint32) err } return c.SendRequest(&livekit.SignalRequest{ Message: &livekit.SignalRequest_Offer{ - Offer: rtc.ToProtoSessionDescription(offer, offerId), + Offer: signalling.ToProtoSessionDescription(offer, offerId), }, }) }