dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-07-28 18:39:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Signal handling interfaces and participant specific HTTP PATCH. (#3804)

Browse Source 
* HTTP PATCH on participant sid

* rename

* signal handler

* new files

* clean up

* clean up reassembly

* handle messages on server side

* deps

* staticcheck

* clean up
This commit is contained in:
Raja Subramanian
2025-07-20 22:47:36 +05:30
committed by GitHub
parent 7837c8e595
commit 0a1bfd3092
25 changed files with 592 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+2 -2
View File
@@ -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
go.sum
+4 -4
View File
@@ -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=
pkg/rtc/errors.go
-3
View File
@@ -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")
pkg/rtc/participant.go
+14 -8
View File
@@ -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)
}
pkg/rtc/participant_internal_test.go
+4 -3
View File
@@ -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)
pkg/rtc/participant_signal.go
+4 -3
View File
@@ -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 {
pkg/rtc/room.go
+2 -1
View File
@@ -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)
pkg/rtc/signalling/errors.go
+26
View File
@@ -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")
)
pkg/rtc/signalling/interfaces.go
+4
View File
@@ -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
pkg/rtc/signalling/signalfragment.go
+1
View File
@@ -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
}
pkg/rtc/signalhandler.go → pkg/rtc/signalling/signalhandler.go
+65 -33
View File
@@ -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)
}
}
pkg/rtc/signalling/signalhandlerunimplemented.go
+27
View File
@@ -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
}
pkg/rtc/signalling/signalhandlerv2.go
+101
View File
@@ -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
}
pkg/rtc/signalling/utils.go
+66
View File
@@ -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
}
pkg/rtc/types/interfaces.go
+2
View File
@@ -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
pkg/rtc/types/typesfakes/fake_local_participant.go
+74
View File
@@ -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()
pkg/rtc/utils.go
-45
View File
@@ -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:
pkg/service/errors.go
+2
View File
@@ -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")
)
pkg/service/roommanager.go
+31 -6
View File
@@ -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
pkg/service/roommanager_service.go
+39
View File
@@ -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
}
pkg/service/rtcrestservice.go
+1 -1
View File
@@ -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
}
pkg/service/rtcv2service.go
+110 -4
View File
@@ -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)
}
pkg/service/wire.go
+1
View File
@@ -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,
pkg/service/wire_gen.go
+5 -1
View File
@@ -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
test/client/client.go
+7 -6
View File
@@ -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),
},
})
}