diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 3f82ec85b..fd2d251ca 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -358,10 +358,6 @@ func (p *ParticipantImpl) GetClientConfiguration() *livekit.ClientConfiguration return p.params.ClientConf } -func (p *ParticipantImpl) GetICEConnectionType() types.ICEConnectionType { - return p.TransportManager.GetICEConnectionType() -} - func (p *ParticipantImpl) GetBufferFactory() *buffer.Factory { return p.params.Config.BufferFactory } @@ -582,7 +578,7 @@ func (p *ParticipantImpl) OnClaimsChanged(callback func(types.LocalParticipant)) func (p *ParticipantImpl) HandleSignalSourceClose() { p.TransportManager.SetSignalSourceValid(false) - if !p.TransportManager.HasPublisherEverConnected() && !p.TransportManager.HasSubscriberEverConnected() { + if !p.HasConnected() { reason := types.ParticipantCloseReasonJoinFailed _ = p.Close(false, reason, false) } @@ -1708,6 +1704,10 @@ func (p *ParticipantImpl) GetPendingTrack(trackID livekit.TrackID) *livekit.Trac return nil } +func (p *ParticipantImpl) HasConnected() bool { + return p.TransportManager.HasSubscriberEverConnected() || p.TransportManager.HasPublisherEverConnected() +} + func (p *ParticipantImpl) sendTrackPublished(cid string, ti *livekit.TrackInfo) { p.pubLogger.Debugw("sending track published", "cid", cid, "trackInfo", logger.Proto(ti)) _ = p.writeMessage(&livekit.SignalResponse{ diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 09904506f..d832a61a3 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -17,9 +17,11 @@ package rtc import ( "context" "errors" + "fmt" "io" "math" "sort" + "strings" "sync" "time" @@ -325,11 +327,6 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me // it's important to set this before connection, we don't want to miss out on any published tracks participant.OnTrackPublished(r.onTrackPublished) participant.OnStateChange(func(p types.LocalParticipant, oldState livekit.ParticipantInfo_State) { - r.Logger.Infow("participant state changed", - "state", p.State(), - "participant", p.Identity(), - "pID", p.ID(), - "oldState", oldState) if r.onParticipantChanged != nil { r.onParticipantChanged(participant) } @@ -343,15 +340,24 @@ func (r *Room) Join(participant types.LocalParticipant, requestSource routing.Me // start the workers once connectivity is established p.Start() + meta := &livekit.AnalyticsClientMeta{ + ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds()), + } + cds := participant.GetICEConnectionDetails() + for _, cd := range cds { + if cd.Type != types.ICEConnectionTypeUnknown { + meta.ConnectionType = string(cd.Type) + break + } + } r.telemetry.ParticipantActive(context.Background(), r.ToProto(), p.ToProto(), - &livekit.AnalyticsClientMeta{ - ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds()), - ConnectionType: string(p.GetICEConnectionType()), - }, + meta, false, ) + + p.GetLogger().Infow("participant active", connectionDetailsFields(cds)...) } else if state == livekit.ParticipantInfo_DISCONNECTED { // remove participant from room go r.RemoveParticipant(p.Identity(), p.ID(), types.ParticipantCloseReasonStateDisconnected) @@ -495,24 +501,27 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, requestSource routing func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livekit.ParticipantID, reason types.ParticipantCloseReason) { r.lock.Lock() p, ok := r.participants[identity] - if ok { - if pID != "" && p.ID() != pID { - // participant session has been replaced - r.lock.Unlock() - return - } + if !ok { + r.lock.Unlock() + return + } - delete(r.participants, identity) - delete(r.participantOpts, identity) - delete(r.participantRequestSources, identity) - delete(r.hasPublished, identity) - if !p.Hidden() { - r.protoRoom.NumParticipants-- - } + if pID != "" && p.ID() != pID { + // participant session has been replaced + r.lock.Unlock() + return + } + + delete(r.participants, identity) + delete(r.participantOpts, identity) + delete(r.participantRequestSources, identity) + delete(r.hasPublished, identity) + if !p.Hidden() { + r.protoRoom.NumParticipants-- } immediateChange := false - if (p != nil && p.IsRecorder()) || r.protoRoom.ActiveRecording { + if p.IsRecorder() { activeRecording := false for _, op := range r.participants { if op.IsRecorder() { @@ -524,14 +533,16 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek if r.protoRoom.ActiveRecording != activeRecording { r.protoRoom.ActiveRecording = activeRecording immediateChange = true - } } r.lock.Unlock() r.protoProxy.MarkDirty(immediateChange) - if !ok { - return + if !p.HasConnected() { + fields := append(connectionDetailsFields(p.GetICEConnectionDetails()), + "reason", reason.String(), + ) + p.GetLogger().Infow("removing participant without connection", fields...) } // send broadcast only if it's not already closed @@ -551,7 +562,6 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, pID livek p.OnSubscribeStatusChanged(nil) // close participant as well - r.Logger.Debugw("closing participant for removal", "pID", p.ID(), "participant", p.Identity()) _ = p.Close(true, reason, false) r.leftAt.Store(time.Now().Unix()) @@ -1416,3 +1426,39 @@ func BroadcastDataPacketForRoom(r types.Room, source types.LocalParticipant, dp } }) } + +func connectionDetailsFields(cds []*types.ICEConnectionDetails) []interface{} { + var fields []interface{} + connectionType := types.ICEConnectionTypeUnknown + for _, cd := range cds { + candidates := make([]string, 0, len(cd.Remote)+len(cd.Local)) + for _, c := range cd.Local { + cStr := "[local]" + if c.Selected { + cStr += "[selected]" + } else if c.Filtered { + cStr += "[filtered]" + } + cStr += " " + c.Local.String() + candidates = append(candidates, cStr) + } + for _, c := range cd.Remote { + cStr := "[remote]" + if c.Selected { + cStr += "[selected]" + } else if c.Filtered { + cStr += "[filtered]" + } + cStr += " " + c.Remote.String() + candidates = append(candidates, cStr) + } + if len(candidates) > 0 { + fields = append(fields, fmt.Sprintf("%sCandidates", strings.ToLower(cd.Transport.String())), candidates) + } + if cd.Type != types.ICEConnectionTypeUnknown { + connectionType = cd.Type + } + } + fields = append(fields, "connectionType", connectionType) + return fields +} diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 4230b9dfd..dede7cdf9 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -23,7 +23,6 @@ import ( "github.com/bep/debounce" "github.com/pion/dtls/v2/pkg/crypto/elliptic" - "github.com/pion/ice/v2" "github.com/pion/interceptor" "github.com/pion/interceptor/pkg/cc" "github.com/pion/interceptor/pkg/gcc" @@ -35,13 +34,6 @@ import ( "github.com/pkg/errors" "go.uber.org/atomic" - sutils "github.com/livekit/livekit-server/pkg/utils" - "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/logger" - "github.com/livekit/protocol/logger/pionlogger" - lksdp "github.com/livekit/protocol/sdp" - "github.com/livekit/protocol/utils" - "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu/pacer" @@ -49,6 +41,11 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/streamallocator" "github.com/livekit/livekit-server/pkg/telemetry" "github.com/livekit/livekit-server/pkg/telemetry/prometheus" + sutils "github.com/livekit/livekit-server/pkg/utils" + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/logger/pionlogger" + lksdp "github.com/livekit/protocol/sdp" ) const ( @@ -94,7 +91,6 @@ const ( signalICEGatheringComplete signal = iota signalLocalICECandidate signalRemoteICECandidate - signalLogICECandidates signalSendOffer signalRemoteDescriptionReceived signalICERestart @@ -108,8 +104,6 @@ func (s signal) String() string { return "LOCAL_ICE_CANDIDATE" case signalRemoteICECandidate: return "REMOTE_ICE_CANDIDATE" - case signalLogICECandidates: - return "LOG_ICE_CANDIDATES" case signalSendOffer: return "SEND_OFFER" case signalRemoteDescriptionReceived: @@ -234,11 +228,7 @@ type PCTransport struct { currentOfferIceCredential string // ice user:pwd, for publish side ice restart checking pendingRestartIceOffer *webrtc.SessionDescription - // for cleaner logging - allowedLocalCandidates *utils.DedupedSlice[string] - allowedRemoteCandidates *utils.DedupedSlice[string] - filteredLocalCandidates *utils.DedupedSlice[string] - filteredRemoteCandidates *utils.DedupedSlice[string] + connectionDetails *types.ICEConnectionDetails } type TransportParams struct { @@ -251,6 +241,7 @@ type TransportParams struct { Telemetry telemetry.TelemetryService EnabledCodecs []*livekit.Codec Logger logger.Logger + Transport livekit.SignalTarget SimTracks map[uint32]SimulcastTrackInfo ClientInfo ClientInfo IsOfferer bool @@ -393,10 +384,7 @@ func NewPCTransport(params TransportParams) (*PCTransport, error) { eventCh: make(chan event, 100), previousTrackDescription: make(map[string]*trackDescription), canReuseTransceiver: true, - allowedLocalCandidates: utils.NewDedupedSlice[string](maxICECandidates), - allowedRemoteCandidates: utils.NewDedupedSlice[string](maxICECandidates), - filteredLocalCandidates: utils.NewDedupedSlice[string](maxICECandidates), - filteredRemoteCandidates: utils.NewDedupedSlice[string](maxICECandidates), + connectionDetails: types.NewICEConnectionDetails(params.Transport, params.Logger), } if params.IsSendSide { t.streamAllocator = streamallocator.NewStreamAllocator(streamallocator.StreamAllocatorParams{ @@ -564,12 +552,6 @@ func (t *PCTransport) getSelectedPair() (*webrtc.ICECandidatePair, error) { return iceTransport.GetSelectedCandidatePair() } -func (t *PCTransport) logICECandidates() { - t.postEvent(event{ - signal: signalLogICECandidates, - }) -} - func (t *PCTransport) setConnectedAt(at time.Time) bool { t.lock.Lock() t.connectedAt = at @@ -629,6 +611,14 @@ func (t *PCTransport) onICEConnectionStateChange(state webrtc.ICEConnectionState switch state { case webrtc.ICEConnectionStateConnected: t.setICEConnectedAt(time.Now()) + go func() { + pair, err := t.getSelectedPair() + if err != nil { + t.params.Logger.Warnw("failed to get selected candidate pair", err) + return + } + t.connectionDetails.SetSelectedPair(pair) + }() case webrtc.ICEConnectionStateChecking: t.setICEStartedAt(time.Now()) @@ -905,6 +895,10 @@ func (t *PCTransport) HasEverConnected() bool { return !t.firstConnectedAt.IsZero() } +func (t *PCTransport) GetICEConnectionDetails() *types.ICEConnectionDetails { + return t.connectionDetails +} + func (t *PCTransport) WriteRTCP(pkts []rtcp.Packet) error { return t.pc.WriteRTCP(pkts) } @@ -1198,44 +1192,6 @@ func (t *PCTransport) SetChannelCapacityOfStreamAllocator(channelCapacity int64) t.streamAllocator.SetChannelCapacity(channelCapacity) } -func (t *PCTransport) GetICEConnectionType() types.ICEConnectionType { - unknown := types.ICEConnectionTypeUnknown - if t.pc == nil { - return unknown - } - p, err := t.getSelectedPair() - if err != nil || p == nil { - return unknown - } - - if p.Remote.Typ == webrtc.ICECandidateTypeRelay { - return types.ICEConnectionTypeTURN - } else if p.Remote.Typ == webrtc.ICECandidateTypePrflx { - // if the remote relay candidate pings us *before* we get a relay candidate, - // Pion would have created a prflx candidate with the same address as the relay candidate. - // to report an accurate connection type, we'll compare to see if existing relay candidates match - t.lock.RLock() - allowedRemoteCandidates := t.allowedRemoteCandidates.Get() - t.lock.RUnlock() - - for _, ci := range allowedRemoteCandidates { - candidateValue := strings.TrimPrefix(ci, "candidate:") - candidate, err := ice.UnmarshalCandidate(candidateValue) - if err == nil && candidate.Type() == ice.CandidateTypeRelay { - if p.Remote.Address == candidate.Address() && - p.Remote.Port == uint16(candidate.Port()) && - p.Remote.Protocol.String() == candidate.NetworkType().NetworkShort() { - return types.ICEConnectionTypeTURN - } - } - } - } - if p.Remote.Protocol == webrtc.ICEProtocolTCP { - return types.ICEConnectionTypeTCP - } - return types.ICEConnectionTypeUDP -} - func (t *PCTransport) preparePC(previousAnswer webrtc.SessionDescription) error { // sticky data channel to first m-lines, if someday we don't send sdp without media streams to // client's subscribe pc after joining, should change this step @@ -1453,7 +1409,6 @@ func (t *PCTransport) processEvents() { t.clearSignalStateCheckTimer() t.params.Logger.Debugw("leaving events processor") - t.handleLogICECandidates(nil) } func (t *PCTransport) handleEvent(e *event) error { @@ -1464,8 +1419,6 @@ func (t *PCTransport) handleEvent(e *event) error { return t.handleLocalICECandidate(e) case signalRemoteICECandidate: return t.handleRemoteICECandidate(e) - case signalLogICECandidates: - return t.handleLogICECandidates(e) case signalSendOffer: return t.handleSendOffer(e) case signalRemoteDescriptionReceived: @@ -1537,33 +1490,28 @@ func (t *PCTransport) localDescriptionSent() error { func (t *PCTransport) clearLocalDescriptionSent() { t.cacheLocalCandidates = true t.cachedLocalCandidates = nil - - t.allowedLocalCandidates.Clear() - t.lock.Lock() - t.allowedRemoteCandidates.Clear() - t.lock.Unlock() - t.filteredLocalCandidates.Clear() - t.filteredRemoteCandidates.Clear() + t.connectionDetails.Clear() } func (t *PCTransport) handleLocalICECandidate(e *event) error { c := e.data.(*webrtc.ICECandidate) filtered := false - if t.preferTCP.Load() && c != nil && c.Protocol != webrtc.ICEProtocolTCP { - cstr := c.String() - t.params.Logger.Debugw("filtering out local candidate", "candidate", cstr) - t.filteredLocalCandidates.Add(cstr) - filtered = true + if c != nil { + if t.preferTCP.Load() && c.Protocol != webrtc.ICEProtocolTCP { + t.params.Logger.Debugw("filtering out local candidate", + "candidate", func() interface{} { + return c.String() + }) + filtered = true + } + t.connectionDetails.AddLocalCandidate(c, filtered) } if filtered { return nil } - if c != nil { - t.allowedLocalCandidates.Add(c.String()) - } if t.cacheLocalCandidates { t.cachedLocalCandidates = append(t.cachedLocalCandidates, c) return nil @@ -1582,18 +1530,14 @@ func (t *PCTransport) handleRemoteICECandidate(e *event) error { filtered := false if t.preferTCP.Load() && !strings.Contains(c.Candidate, "tcp") { t.params.Logger.Debugw("filtering out remote candidate", "candidate", c.Candidate) - t.filteredRemoteCandidates.Add(c.Candidate) filtered = true } + t.connectionDetails.AddRemoteCandidate(*c, filtered) if filtered { return nil } - t.lock.Lock() - t.allowedRemoteCandidates.Add(c.Candidate) - t.lock.Unlock() - if t.pc.RemoteDescription() == nil { t.pendingRemoteCandidates = append(t.pendingRemoteCandidates, c) return nil @@ -1606,30 +1550,6 @@ func (t *PCTransport) handleRemoteICECandidate(e *event) error { return nil } -func (t *PCTransport) handleLogICECandidates(_ *event) error { - lc := t.allowedLocalCandidates.Get() - rc := t.allowedRemoteCandidates.Get() - var fields []interface{} - if len(lc) != 0 || len(rc) != 0 { - fields = append(fields, - "lc", lc, - "rc", rc, - "lc_filtered", t.filteredLocalCandidates.Get(), - "rc_filtered", t.filteredRemoteCandidates.Get(), - ) - - } - if pair, err := t.getSelectedPair(); err == nil { - fields = append(fields, "selected_pair", pair) - } - - if len(fields) > 0 { - t.params.Logger.Infow("ice candidates", fields...) - } - - return nil -} - func (t *PCTransport) setNegotiationState(state NegotiationState) { t.negotiationState = state if onNegotiationStateChanged := t.getOnNegotiationStateChanged(); onNegotiationStateChanged != nil { diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 6805bd430..21591f3c2 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -124,6 +124,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro Logger: LoggerWithPCTarget(params.Logger, livekit.SignalTarget_PUBLISHER), SimTracks: params.SimTracks, ClientInfo: params.ClientInfo, + Transport: livekit.SignalTarget_PUBLISHER, }) if err != nil { return nil, err @@ -159,6 +160,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro IsSendSide: true, AllowPlayoutDelay: params.AllowPlayoutDelay, DataChannelMaxBufferedAmount: params.DataChannelMaxBufferedAmount, + Transport: livekit.SignalTarget_SUBSCRIBER, }) if err != nil { return nil, err @@ -554,8 +556,15 @@ func (t *TransportManager) SubscriberAsPrimary() bool { return t.params.SubscriberAsPrimary } -func (t *TransportManager) GetICEConnectionType() types.ICEConnectionType { - return t.getTransport(true).GetICEConnectionType() +func (t *TransportManager) GetICEConnectionDetails() []*types.ICEConnectionDetails { + details := make([]*types.ICEConnectionDetails, 0, 2) + for _, pc := range []*PCTransport{t.publisher, t.subscriber} { + cd := pc.GetICEConnectionDetails() + if cd.HasCandidates() { + details = append(details, cd.Clone()) + } + } + return details } func (t *TransportManager) getTransport(isPrimary bool) *PCTransport { diff --git a/pkg/rtc/types/ice.go b/pkg/rtc/types/ice.go new file mode 100644 index 000000000..531e1ad43 --- /dev/null +++ b/pkg/rtc/types/ice.go @@ -0,0 +1,254 @@ +/* + * 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 types + +import ( + "strings" + "sync" + + "github.com/pion/ice/v2" + "github.com/pion/webrtc/v3" + "golang.org/x/exp/slices" + + "github.com/livekit/protocol/livekit" + "github.com/livekit/protocol/logger" +) + +type ICEConnectionType string + +const ( + ICEConnectionTypeUDP ICEConnectionType = "udp" + ICEConnectionTypeTCP ICEConnectionType = "tcp" + ICEConnectionTypeTURN ICEConnectionType = "turn" + ICEConnectionTypeUnknown ICEConnectionType = "unknown" +) + +type ICECandidateExtended struct { + // only one of local or remote is set. This is due to type foo in Pion + Local *webrtc.ICECandidate + Remote ice.Candidate + Selected bool + Filtered bool +} + +type ICEConnectionDetails struct { + Local []*ICECandidateExtended + Remote []*ICECandidateExtended + Transport livekit.SignalTarget + Type ICEConnectionType + lock sync.Mutex + logger logger.Logger +} + +func NewICEConnectionDetails(transport livekit.SignalTarget, l logger.Logger) *ICEConnectionDetails { + d := &ICEConnectionDetails{ + Transport: transport, + Type: ICEConnectionTypeUnknown, + logger: l, + } + return d +} + +func (d *ICEConnectionDetails) HasCandidates() bool { + d.lock.Lock() + defer d.lock.Unlock() + return len(d.Local) > 0 || len(d.Remote) > 0 +} + +// Clone returns a copy of the ICEConnectionDetails, where fields can be read without locking +func (d *ICEConnectionDetails) Clone() *ICEConnectionDetails { + d.lock.Lock() + defer d.lock.Unlock() + clone := &ICEConnectionDetails{ + Transport: d.Transport, + Type: d.Type, + logger: d.logger, + Local: make([]*ICECandidateExtended, 0, len(d.Local)), + Remote: make([]*ICECandidateExtended, 0, len(d.Remote)), + } + for _, c := range d.Local { + clone.Local = append(clone.Local, &ICECandidateExtended{ + Local: c.Local, + Filtered: c.Filtered, + }) + } + for _, c := range d.Remote { + clone.Remote = append(clone.Remote, &ICECandidateExtended{ + Remote: c.Remote, + Filtered: c.Filtered, + }) + } + return clone +} + +func (d *ICEConnectionDetails) AddLocalCandidate(c *webrtc.ICECandidate, filtered bool) { + d.lock.Lock() + defer d.lock.Unlock() + compFn := func(e *ICECandidateExtended) bool { + return isCandidateEqualTo(e.Local, c) + } + if slices.ContainsFunc[[]*ICECandidateExtended, *ICECandidateExtended](d.Local, compFn) { + return + } + d.Local = append(d.Local, &ICECandidateExtended{ + Local: c, + Filtered: filtered, + }) +} + +func (d *ICEConnectionDetails) AddRemoteCandidate(c webrtc.ICECandidateInit, filtered bool) { + candidate, err := unmarshalICECandidate(c) + if err != nil { + d.logger.Errorw("could not unmarshal candidate", err, "candidate", c) + return + } + + d.lock.Lock() + defer d.lock.Unlock() + compFn := func(e *ICECandidateExtended) bool { + return isICECandidateEqualTo(e.Remote, candidate) + } + if slices.ContainsFunc[[]*ICECandidateExtended, *ICECandidateExtended](d.Remote, compFn) { + return + } + d.Remote = append(d.Remote, &ICECandidateExtended{ + Remote: candidate, + Filtered: filtered, + }) +} + +func (d *ICEConnectionDetails) Clear() { + d.lock.Lock() + defer d.lock.Unlock() + d.Local = nil + d.Remote = nil + d.Type = ICEConnectionTypeUnknown +} + +func (d *ICEConnectionDetails) SetSelectedPair(pair *webrtc.ICECandidatePair) { + d.lock.Lock() + defer d.lock.Unlock() + remoteIdx := slices.IndexFunc[[]*ICECandidateExtended, *ICECandidateExtended](d.Remote, func(e *ICECandidateExtended) bool { + return isICECandidateEqualToCandidate(e.Remote, pair.Remote) + }) + if remoteIdx < 0 { + // it's possible for prflx candidates to be generated by Pion, we'll add them + candidate, err := unmarshalICECandidate(pair.Remote.ToJSON()) + if err != nil { + d.logger.Errorw("could not unmarshal remote candidate", err, "candidate", pair.Remote) + return + } + d.Remote = append(d.Remote, &ICECandidateExtended{ + Remote: candidate, + Filtered: false, + }) + remoteIdx = len(d.Remote) - 1 + } + remote := d.Remote[remoteIdx] + remote.Selected = true + + localIdx := slices.IndexFunc[[]*ICECandidateExtended, *ICECandidateExtended](d.Local, func(e *ICECandidateExtended) bool { + return isCandidateEqualTo(e.Local, pair.Local) + }) + if localIdx < 0 { + d.logger.Errorw("could not match local candidate", nil, "local", pair.Local) + // should not happen + return + } + local := d.Local[localIdx] + local.Selected = true + + d.Type = ICEConnectionTypeUDP + if pair.Remote.Protocol == webrtc.ICEProtocolTCP { + d.Type = ICEConnectionTypeTCP + } + if pair.Remote.Typ == webrtc.ICECandidateTypeRelay { + d.Type = ICEConnectionTypeTURN + } else if pair.Remote.Typ == webrtc.ICECandidateTypePrflx { + // if the remote relay candidate pings us *before* we get a relay candidate, + // Pion would have created a prflx candidate with the same address as the relay candidate. + // to report an accurate connection type, we'll compare to see if existing relay candidates match + for _, other := range d.Remote { + or := other.Remote + if or.Type() == ice.CandidateTypeRelay && + pair.Remote.Address == or.Address() && + pair.Remote.Port == uint16(or.Port()) && + pair.Remote.Protocol.String() == or.NetworkType().NetworkShort() { + d.Type = ICEConnectionTypeTURN + } + } + } +} + +func isCandidateEqualTo(c1, c2 *webrtc.ICECandidate) bool { + if c1 == nil && c2 == nil { + return true + } + if (c1 == nil && c2 != nil) || (c1 != nil && c2 == nil) { + return false + } + return c1.Typ == c2.Typ && + c1.Protocol == c2.Protocol && + c1.Address == c2.Address && + c1.Port == c2.Port && + c1.Component == c2.Component && + c1.Foundation == c2.Foundation && + c1.Priority == c2.Priority && + c1.RelatedAddress == c2.RelatedAddress && + c1.RelatedPort == c2.RelatedPort && + c1.TCPType == c2.TCPType +} + +func isICECandidateEqualTo(c1, c2 ice.Candidate) bool { + if c1 == nil && c2 == nil { + return true + } + if (c1 == nil && c2 != nil) || (c1 != nil && c2 == nil) { + return false + } + return c1.Type() == c2.Type() && + c1.NetworkType() == c2.NetworkType() && + c1.Address() == c2.Address() && + c1.Port() == c2.Port() && + c1.Component() == c2.Component() && + c1.Foundation() == c2.Foundation() && + c1.Priority() == c2.Priority() && + c1.RelatedAddress().Equal(c2.RelatedAddress()) && + c1.TCPType() == c2.TCPType() +} + +func isICECandidateEqualToCandidate(c1 ice.Candidate, c2 *webrtc.ICECandidate) bool { + if c1 == nil && c2 == nil { + return true + } + if (c1 == nil && c2 != nil) || (c1 != nil && c2 == nil) { + return false + } + return c1.Type().String() == c2.Typ.String() && + c1.NetworkType().NetworkShort() == c2.Protocol.String() && + c1.Address() == c2.Address && + c1.Port() == int(c2.Port) && + c1.Component() == c2.Component && + c1.Foundation() == c2.Foundation && + c1.Priority() == c2.Priority && + c1.TCPType().String() == c2.TCPType +} + +func unmarshalICECandidate(c webrtc.ICECandidateInit) (ice.Candidate, error) { + candidateValue := strings.TrimPrefix(c.Candidate, "candidate:") + return ice.UnmarshalCandidate(candidateValue) +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 266e24b8b..77e395d85 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -285,15 +285,6 @@ type Participant interface { // ------------------------------------------------------- -type ICEConnectionType string - -const ( - ICEConnectionTypeUDP ICEConnectionType = "udp" - ICEConnectionTypeTCP ICEConnectionType = "tcp" - ICEConnectionTypeTURN ICEConnectionType = "turn" - ICEConnectionTypeUnknown ICEConnectionType = "unknown" -) - type AddTrackParams struct { Stereo bool Red bool @@ -320,10 +311,11 @@ type LocalParticipant interface { SubscriberAsPrimary() bool GetClientInfo() *livekit.ClientInfo GetClientConfiguration() *livekit.ClientConfiguration - GetICEConnectionType() ICEConnectionType GetBufferFactory() *buffer.Factory GetPlayoutDelayConfig() *livekit.PlayoutDelay GetPendingTrack(trackID livekit.TrackID) *livekit.TrackInfo + GetICEConnectionDetails() []*ICEConnectionDetails + HasConnected() bool SetResponseSink(sink routing.MessageSink) CloseSignalConnection(reason SignallingCloseReason) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index a01a3834b..0cd75616a 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -233,15 +233,15 @@ type FakeLocalParticipant struct { getConnectionQualityReturnsOnCall map[int]struct { result1 *livekit.ConnectionQualityInfo } - GetICEConnectionTypeStub func() types.ICEConnectionType - getICEConnectionTypeMutex sync.RWMutex - getICEConnectionTypeArgsForCall []struct { + GetICEConnectionDetailsStub func() []*types.ICEConnectionDetails + getICEConnectionDetailsMutex sync.RWMutex + getICEConnectionDetailsArgsForCall []struct { } - getICEConnectionTypeReturns struct { - result1 types.ICEConnectionType + getICEConnectionDetailsReturns struct { + result1 []*types.ICEConnectionDetails } - getICEConnectionTypeReturnsOnCall map[int]struct { - result1 types.ICEConnectionType + getICEConnectionDetailsReturnsOnCall map[int]struct { + result1 []*types.ICEConnectionDetails } GetLoggerStub func() logger.Logger getLoggerMutex sync.RWMutex @@ -371,6 +371,16 @@ type FakeLocalParticipant struct { handleSignalSourceCloseMutex sync.RWMutex handleSignalSourceCloseArgsForCall []struct { } + HasConnectedStub func() bool + hasConnectedMutex sync.RWMutex + hasConnectedArgsForCall []struct { + } + hasConnectedReturns struct { + result1 bool + } + hasConnectedReturnsOnCall map[int]struct { + result1 bool + } HasPermissionStub func(livekit.TrackID, livekit.ParticipantIdentity) bool hasPermissionMutex sync.RWMutex hasPermissionArgsForCall []struct { @@ -2062,15 +2072,15 @@ func (fake *FakeLocalParticipant) GetConnectionQualityReturnsOnCall(i int, resul }{result1} } -func (fake *FakeLocalParticipant) GetICEConnectionType() types.ICEConnectionType { - fake.getICEConnectionTypeMutex.Lock() - ret, specificReturn := fake.getICEConnectionTypeReturnsOnCall[len(fake.getICEConnectionTypeArgsForCall)] - fake.getICEConnectionTypeArgsForCall = append(fake.getICEConnectionTypeArgsForCall, struct { +func (fake *FakeLocalParticipant) GetICEConnectionDetails() []*types.ICEConnectionDetails { + fake.getICEConnectionDetailsMutex.Lock() + ret, specificReturn := fake.getICEConnectionDetailsReturnsOnCall[len(fake.getICEConnectionDetailsArgsForCall)] + fake.getICEConnectionDetailsArgsForCall = append(fake.getICEConnectionDetailsArgsForCall, struct { }{}) - stub := fake.GetICEConnectionTypeStub - fakeReturns := fake.getICEConnectionTypeReturns - fake.recordInvocation("GetICEConnectionType", []interface{}{}) - fake.getICEConnectionTypeMutex.Unlock() + stub := fake.GetICEConnectionDetailsStub + fakeReturns := fake.getICEConnectionDetailsReturns + fake.recordInvocation("GetICEConnectionDetails", []interface{}{}) + fake.getICEConnectionDetailsMutex.Unlock() if stub != nil { return stub() } @@ -2080,38 +2090,38 @@ func (fake *FakeLocalParticipant) GetICEConnectionType() types.ICEConnectionType return fakeReturns.result1 } -func (fake *FakeLocalParticipant) GetICEConnectionTypeCallCount() int { - fake.getICEConnectionTypeMutex.RLock() - defer fake.getICEConnectionTypeMutex.RUnlock() - return len(fake.getICEConnectionTypeArgsForCall) +func (fake *FakeLocalParticipant) GetICEConnectionDetailsCallCount() int { + fake.getICEConnectionDetailsMutex.RLock() + defer fake.getICEConnectionDetailsMutex.RUnlock() + return len(fake.getICEConnectionDetailsArgsForCall) } -func (fake *FakeLocalParticipant) GetICEConnectionTypeCalls(stub func() types.ICEConnectionType) { - fake.getICEConnectionTypeMutex.Lock() - defer fake.getICEConnectionTypeMutex.Unlock() - fake.GetICEConnectionTypeStub = stub +func (fake *FakeLocalParticipant) GetICEConnectionDetailsCalls(stub func() []*types.ICEConnectionDetails) { + fake.getICEConnectionDetailsMutex.Lock() + defer fake.getICEConnectionDetailsMutex.Unlock() + fake.GetICEConnectionDetailsStub = stub } -func (fake *FakeLocalParticipant) GetICEConnectionTypeReturns(result1 types.ICEConnectionType) { - fake.getICEConnectionTypeMutex.Lock() - defer fake.getICEConnectionTypeMutex.Unlock() - fake.GetICEConnectionTypeStub = nil - fake.getICEConnectionTypeReturns = struct { - result1 types.ICEConnectionType +func (fake *FakeLocalParticipant) GetICEConnectionDetailsReturns(result1 []*types.ICEConnectionDetails) { + fake.getICEConnectionDetailsMutex.Lock() + defer fake.getICEConnectionDetailsMutex.Unlock() + fake.GetICEConnectionDetailsStub = nil + fake.getICEConnectionDetailsReturns = struct { + result1 []*types.ICEConnectionDetails }{result1} } -func (fake *FakeLocalParticipant) GetICEConnectionTypeReturnsOnCall(i int, result1 types.ICEConnectionType) { - fake.getICEConnectionTypeMutex.Lock() - defer fake.getICEConnectionTypeMutex.Unlock() - fake.GetICEConnectionTypeStub = nil - if fake.getICEConnectionTypeReturnsOnCall == nil { - fake.getICEConnectionTypeReturnsOnCall = make(map[int]struct { - result1 types.ICEConnectionType +func (fake *FakeLocalParticipant) GetICEConnectionDetailsReturnsOnCall(i int, result1 []*types.ICEConnectionDetails) { + fake.getICEConnectionDetailsMutex.Lock() + defer fake.getICEConnectionDetailsMutex.Unlock() + fake.GetICEConnectionDetailsStub = nil + if fake.getICEConnectionDetailsReturnsOnCall == nil { + fake.getICEConnectionDetailsReturnsOnCall = make(map[int]struct { + result1 []*types.ICEConnectionDetails }) } - fake.getICEConnectionTypeReturnsOnCall[i] = struct { - result1 types.ICEConnectionType + fake.getICEConnectionDetailsReturnsOnCall[i] = struct { + result1 []*types.ICEConnectionDetails }{result1} } @@ -2811,6 +2821,59 @@ func (fake *FakeLocalParticipant) HandleSignalSourceCloseCalls(stub func()) { fake.HandleSignalSourceCloseStub = stub } +func (fake *FakeLocalParticipant) HasConnected() bool { + fake.hasConnectedMutex.Lock() + ret, specificReturn := fake.hasConnectedReturnsOnCall[len(fake.hasConnectedArgsForCall)] + fake.hasConnectedArgsForCall = append(fake.hasConnectedArgsForCall, struct { + }{}) + stub := fake.HasConnectedStub + fakeReturns := fake.hasConnectedReturns + fake.recordInvocation("HasConnected", []interface{}{}) + fake.hasConnectedMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) HasConnectedCallCount() int { + fake.hasConnectedMutex.RLock() + defer fake.hasConnectedMutex.RUnlock() + return len(fake.hasConnectedArgsForCall) +} + +func (fake *FakeLocalParticipant) HasConnectedCalls(stub func() bool) { + fake.hasConnectedMutex.Lock() + defer fake.hasConnectedMutex.Unlock() + fake.HasConnectedStub = stub +} + +func (fake *FakeLocalParticipant) HasConnectedReturns(result1 bool) { + fake.hasConnectedMutex.Lock() + defer fake.hasConnectedMutex.Unlock() + fake.HasConnectedStub = nil + fake.hasConnectedReturns = struct { + result1 bool + }{result1} +} + +func (fake *FakeLocalParticipant) HasConnectedReturnsOnCall(i int, result1 bool) { + fake.hasConnectedMutex.Lock() + defer fake.hasConnectedMutex.Unlock() + fake.HasConnectedStub = nil + if fake.hasConnectedReturnsOnCall == nil { + fake.hasConnectedReturnsOnCall = make(map[int]struct { + result1 bool + }) + } + fake.hasConnectedReturnsOnCall[i] = struct { + result1 bool + }{result1} +} + func (fake *FakeLocalParticipant) HasPermission(arg1 livekit.TrackID, arg2 livekit.ParticipantIdentity) bool { fake.hasPermissionMutex.Lock() ret, specificReturn := fake.hasPermissionReturnsOnCall[len(fake.hasPermissionArgsForCall)] @@ -6156,8 +6219,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getClientInfoMutex.RUnlock() fake.getConnectionQualityMutex.RLock() defer fake.getConnectionQualityMutex.RUnlock() - fake.getICEConnectionTypeMutex.RLock() - defer fake.getICEConnectionTypeMutex.RUnlock() + fake.getICEConnectionDetailsMutex.RLock() + defer fake.getICEConnectionDetailsMutex.RUnlock() fake.getLoggerMutex.RLock() defer fake.getLoggerMutex.RUnlock() fake.getPacerMutex.RLock() @@ -6186,6 +6249,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.handleReconnectAndSendResponseMutex.RUnlock() fake.handleSignalSourceCloseMutex.RLock() defer fake.handleSignalSourceCloseMutex.RUnlock() + fake.hasConnectedMutex.RLock() + defer fake.hasConnectedMutex.RUnlock() fake.hasPermissionMutex.RLock() defer fake.hasPermissionMutex.RUnlock() fake.hiddenMutex.RLock()