Skip to content

Commit 6c062f4

Browse files
rahul-lohraaleksandar-apostolov
rahul-lohra
and
aleksandar-apostolov
authored
Improve video analytics by correctly sending ice state (#1724)
* improve: fix sending correct ice state * improve: correctly update analytics stage * improve: refactor * fix: remove grace period to wait for ice state * fix: refactor * fix: refactor --------- Co-authored-by: Aleksandar Apostolov <apostolov.alexandar@gmail.com>
1 parent d0e1456 commit 6c062f4

8 files changed

Lines changed: 361 additions & 125 deletions

File tree

‎stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/CallAnalytics.kt‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,9 @@ import io.getstream.video.android.core.analytics.call.observer.PeerConnectionAna
2929
import io.getstream.video.android.core.analytics.call.observer.SfuAnalytics
3030
import io.getstream.video.android.core.analytics.call.observer.SfuAnalyticsStateHolder
3131
import io.getstream.video.android.core.analytics.call.observer.VideoAnalytics
32+
import io.getstream.video.android.core.analytics.call.observer.VideoAnalyticsIceState
3233
import io.getstream.video.android.core.analytics.call.observer.model.Stage
34+
import io.getstream.video.android.core.analytics.call.observer.toVideoAnalyticsIceState
3335
import io.getstream.video.android.core.analytics.reporting.ClientEventReporter
3436
import io.getstream.video.android.core.analytics.reporting.model.AnalyticsCallAbortReason
3537
import io.getstream.video.android.core.call.RtcSession
@@ -130,8 +132,12 @@ internal class CallAnalytics(
130132
Pair(AnalyticsCallAbortReason.CUSTOM, callLeaveReason.message)
131133
}
132134
}
133-
val publisherIceState = session.value?.publisher?.value?.iceState?.value
134-
val subscriberIceState = session.value?.subscriber?.value?.iceState?.value
135+
val publisherIceState =
136+
session.value?.publisher?.value?.iceState?.value?.toVideoAnalyticsIceState()
137+
?: VideoAnalyticsIceState.NOT_CONNECTED
138+
val subscriberIceState =
139+
session.value?.subscriber?.value?.iceState?.value?.toVideoAnalyticsIceState()
140+
?: VideoAnalyticsIceState.NOT_CONNECTED
135141
eventReporter.abortAllPostCallInFlight(
136142
publisherIceState,
137143
subscriberIceState,

‎stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/observer/PeerConnectionAnalytics.kt‎

Lines changed: 89 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -21,11 +21,15 @@ import io.getstream.video.android.core.analytics.call.observer.model.Stage
2121
import io.getstream.video.android.core.analytics.reporting.ClientEventReporter
2222
import io.getstream.video.android.core.analytics.reporting.model.PeerConnectionRole
2323
import io.getstream.video.android.core.call.RtcSession
24+
import io.getstream.video.android.core.call.connection.StreamPeerConnection
2425
import kotlinx.coroutines.CoroutineScope
26+
import kotlinx.coroutines.Job
2527
import kotlinx.coroutines.flow.StateFlow
28+
import kotlinx.coroutines.flow.distinctUntilChanged
2629
import kotlinx.coroutines.flow.filter
2730
import kotlinx.coroutines.flow.filterNotNull
2831
import kotlinx.coroutines.flow.flatMapLatest
32+
import kotlinx.coroutines.flow.map
2933
import kotlinx.coroutines.launch
3034
import org.webrtc.PeerConnection
3135

@@ -39,7 +43,7 @@ internal class PeerConnectionAnalytics(
3943
val stateHolder: PeerConnectionAnalyticsStateHolder = PeerConnectionAnalyticsStateHolder(),
4044
) {
4145

42-
val allowedStates = listOf(
46+
val allowedPcStates = listOf(
4347
PeerConnection.PeerConnectionState.CONNECTING,
4448
PeerConnection.PeerConnectionState.FAILED,
4549
PeerConnection.PeerConnectionState.CONNECTED,
@@ -49,71 +53,91 @@ internal class PeerConnectionAnalytics(
4953
stateHolder.state.value.peerConnectionObserverJob?.cancel()
5054
val peerConnectionObserverJob = observerScope.launch {
5155
stateHolder.state.value.publisherJob?.cancel()
52-
val publisherJob = launch {
53-
session.filterNotNull()
54-
.flatMapLatest { it.publisher.filterNotNull() }
55-
.flatMapLatest {
56-
it.state.filter { it ->
57-
allowedStates.contains(
58-
it,
59-
)
60-
}.filterNotNull()
61-
}.filter {
62-
val existingStage = stateHolder.state.value.publisherStage
63-
val newStage = getStage(it)
64-
val isExistingStageAndNewStageAreCompleted = (existingStage == Stage.COMPLETED && newStage == Stage.COMPLETED)
65-
!isExistingStageAndNewStageAreCompleted
66-
}
67-
.collect { state ->
68-
val publisherStage = getStage(state)
69-
publisherStage?.let {
70-
stateHolder.updatePublisherStage(publisherStage)
71-
observerScope.launch {
72-
session.value?.publisher?.value?.let { publisher ->
73-
onPeerConnectionStateChanged(
74-
publisher.hashCode(),
75-
role = PeerConnectionRole.PUBLISH,
76-
iceState = publisher.iceState.value,
77-
peerConnectionState = state,
78-
)
79-
}
80-
}
81-
}
82-
}
83-
}
56+
val publisherJob = observeConnection(
57+
session = session,
58+
role = PeerConnectionRole.PUBLISH,
59+
connectionOf = { it.publisher },
60+
currentStage = { stateHolder.state.value.publisherStage },
61+
updateStage = stateHolder::updatePublisherStage,
62+
)
8463
stateHolder.updatePublisherJob(publisherJob)
64+
8565
stateHolder.state.value.subscriberJob?.cancel()
86-
val subscriberJob = launch {
87-
session.filterNotNull()
88-
.flatMapLatest { it.subscriber.filterNotNull() }
89-
.flatMapLatest {
90-
it.state.filter { it -> allowedStates.contains(it) }.filterNotNull()
91-
}.filter {
92-
val existingStage = stateHolder.state.value.subscriberStage
66+
val subscriberJob = observeConnection(
67+
session = session,
68+
role = PeerConnectionRole.SUBSCRIBE,
69+
connectionOf = { it.subscriber },
70+
currentStage = { stateHolder.state.value.subscriberStage },
71+
updateStage = stateHolder::updateSubscriberStage,
72+
)
73+
stateHolder.updateSubscriberJob(subscriberJob)
74+
}
75+
stateHolder.updatePeerConnectionObserverJob(peerConnectionObserverJob)
76+
}
77+
78+
/**
79+
* Observes a single peer connection ([connectionOf]) for analytics.
80+
*
81+
* Each allowed peer-connection state ([allowedPcStates]) is reported together with the ICE
82+
* state as it stands at that moment ([StreamPeerConnection.iceState] value, mapped via
83+
* [toVideoAnalyticsIceState]). So a CONNECTED peer connection reports CONNECTED only when its
84+
* ICE is already connected, otherwise NOT_CONNECTED. The upstream filter drops a transition
85+
* that would merely repeat an already-completed stage (e.g. CONNECTED after FAILED), and
86+
* [distinctUntilChanged] avoids emitting the same combination twice.
87+
*
88+
* The stage is updated in lockstep with the reported event (in the terminal collector), not
89+
* when the raw peer-connection state changes. It therefore stays IN_PROGRESS while the
90+
* peer-connection session is open and only flips to COMPLETED once the completed event is
91+
* actually emitted, which keeps the call-leave flow's "is a stage still in progress?" check
92+
* correct.
93+
*/
94+
private fun CoroutineScope.observeConnection(
95+
session: StateFlow<RtcSession?>,
96+
role: PeerConnectionRole,
97+
connectionOf: (RtcSession) -> StateFlow<StreamPeerConnection?>,
98+
currentStage: () -> Stage,
99+
updateStage: (Stage) -> Unit,
100+
): Job = launch {
101+
session.filterNotNull()
102+
.flatMapLatest { connectionOf(it).filterNotNull() }
103+
.flatMapLatest { connection ->
104+
connection.state
105+
.filter { allowedPcStates.contains(it) }
106+
.filterNotNull()
107+
.filter {
108+
// Skip a peer-connection state that would only repeat an already-completed
109+
// stage (e.g. a CONNECTED after a FAILED). Gating here — before the ICE read
110+
// and the collector — means we don't do any work for such transitions.
111+
val existingStage = currentStage()
93112
val newStage = getStage(it)
94113
val isExistingStageAndNewStageAreCompleted = (existingStage == Stage.COMPLETED && newStage == Stage.COMPLETED)
95114
!isExistingStageAndNewStageAreCompleted
96115
}
97-
.collect { state ->
98-
val stage = getStage(state)
99-
stage?.let {
100-
stateHolder.updateSubscriberStage(stage)
101-
observerScope.launch {
102-
session.value?.subscriber?.value?.let { subscriber ->
103-
onPeerConnectionStateChanged(
104-
subscriber.hashCode(),
105-
role = PeerConnectionRole.SUBSCRIBE,
106-
iceState = subscriber.iceState.value,
107-
peerConnectionState = state,
108-
)
109-
}
110-
}
111-
}
116+
.map { pcState ->
117+
PeerConnectionSnapshot(
118+
connection.hashCode(),
119+
pcState,
120+
connection.iceState.value.toVideoAnalyticsIceState(),
121+
)
112122
}
113123
}
114-
stateHolder.updateSubscriberJob(subscriberJob)
115-
}
116-
stateHolder.updatePeerConnectionObserverJob(peerConnectionObserverJob)
124+
.distinctUntilChanged()
125+
.collect { snapshot ->
126+
// Update the stage in lockstep with the reported event (not when the raw
127+
// peer-connection state changed), so it only flips to COMPLETED once the completed
128+
// event is actually emitted. The COMPLETED -> COMPLETED case is already dropped by
129+
// the upstream filter, so no extra guard is needed here.
130+
val pcAnalyticsStage = getStage(snapshot.peerConnectionState)
131+
pcAnalyticsStage?.let {
132+
updateStage(pcAnalyticsStage)
133+
onPeerConnectionStateChanged(
134+
peerConnectionHashCode = snapshot.peerConnectionHashCode,
135+
role = role,
136+
iceState = snapshot.iceState,
137+
peerConnectionState = snapshot.peerConnectionState,
138+
)
139+
}
140+
}
117141
}
118142

119143
private fun getStage(peerConnectionState: PeerConnection.PeerConnectionState): Stage? {
@@ -135,7 +159,7 @@ internal class PeerConnectionAnalytics(
135159
internal fun onPeerConnectionStateChanged(
136160
peerConnectionHashCode: Int,
137161
role: PeerConnectionRole,
138-
iceState: PeerConnection.IceConnectionState?,
162+
iceState: VideoAnalyticsIceState,
139163
peerConnectionState: PeerConnection.PeerConnectionState?,
140164
) {
141165
reporter.onPeerConnectionStateChanged(
@@ -162,3 +186,9 @@ internal class PeerConnectionAnalytics(
162186
observePeerConnections(session)
163187
}
164188
}
189+
190+
private data class PeerConnectionSnapshot(
191+
val peerConnectionHashCode: Int,
192+
val peerConnectionState: PeerConnection.PeerConnectionState,
193+
val iceState: VideoAnalyticsIceState,
194+
)

‎stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/call/observer/VideoAnalyticsIceState.kt‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
/*
2+
* Copyright (c) 2014-2026 Stream.io Inc. All rights reserved.
3+
*
4+
* Licensed under the Stream License;
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://github.com/GetStream/stream-video-android/blob/main/LICENSE
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package io.getstream.video.android.core.analytics.call.observer
18+
19+
import org.webrtc.PeerConnection
20+
21+
internal enum class VideoAnalyticsIceState(val text: String) {
22+
CONNECTED("CONNECTED"), FAILED("FAILED"), NOT_CONNECTED("NOT_CONNECTED")
23+
}
24+
25+
internal fun PeerConnection.IceConnectionState?.toVideoAnalyticsIceState(): VideoAnalyticsIceState {
26+
return when (this) {
27+
PeerConnection.IceConnectionState.CONNECTED,
28+
PeerConnection.IceConnectionState.COMPLETED,
29+
-> VideoAnalyticsIceState.CONNECTED
30+
31+
PeerConnection.IceConnectionState.FAILED -> VideoAnalyticsIceState.FAILED
32+
33+
else -> VideoAnalyticsIceState.NOT_CONNECTED
34+
}
35+
}

‎stream-video-android-core/src/main/kotlin/io/getstream/video/android/core/analytics/reporting/ClientEventFactory.kt‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ package io.getstream.video.android.core.analytics.reporting
1818

1919
import io.getstream.android.video.generated.models.ClientEvent
2020
import io.getstream.video.android.core.StreamVideo
21+
import io.getstream.video.android.core.analytics.call.observer.VideoAnalyticsIceState
2122
import io.getstream.video.android.core.analytics.call.observer.model.JoinReason
2223
import io.getstream.video.android.core.analytics.reporting.model.EventOutcome
2324
import io.getstream.video.android.core.analytics.reporting.model.EventStage
@@ -46,7 +47,7 @@ internal class ClientEventFactory(val sdkVersion: String, val userAgent: () -> S
4647
sfuId: String? = null,
4748
peerConnection: PeerConnectionRole? = null,
4849
wasPreviouslyConnected: Boolean? = null,
49-
iceState: PeerConnection.IceConnectionState? = null,
50+
iceState: VideoAnalyticsIceState? = null,
5051
peerConnectionState: PeerConnection.PeerConnectionState? = null,
5152
userSessionId: String? = null,
5253
screenShareAllowed: Boolean? = null,
@@ -66,7 +67,7 @@ internal class ClientEventFactory(val sdkVersion: String, val userAgent: () -> S
6667
userId = StreamVideo.Companion.instanceOrNull()?.userId,
6768
callSessionId = callSessionId,
6869
elapsedTime = elapsedTime?.toInt(),
69-
iceState = iceState?.name,
70+
iceState = iceState?.text,
7071
outcome = outcome?.value,
7172
peerConnection = peerConnection?.value,
7273
previouslyConnectedTimestamp = null,

0 commit comments

Comments
 (0)