Skip to content

Commit 5d70f3b

Browse files
davidliudavidliu
authored
Implement data streams (#625)
* Implement data streams * Emit warning logs for unhandled datastreams * Fix stream closing
1 parent cd89fb3 commit 5d70f3b

File tree

26 files changed

+1692
-49
lines changed

26 files changed

+1692
-49
lines changed

‎.changeset/wise-cherries-speak.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
---
2+
"client-sdk-android": minor
3+
---
4+
5+
Implement data streams feature

‎.idea/dictionaries/davidliu.xml‎

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎livekit-android-sdk/src/main/java/io/livekit/android/coroutines/FlowExt.kt‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
/*
2+
* Copyright 2025 LiveKit, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "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+
* http://www.apache.org/licenses/LICENSE-2.0
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.livekit.android.coroutines
18+
19+
import kotlinx.coroutines.cancel
20+
import kotlinx.coroutines.coroutineScope
21+
import kotlinx.coroutines.currentCoroutineContext
22+
import kotlinx.coroutines.flow.Flow
23+
import kotlinx.coroutines.flow.collect
24+
import kotlinx.coroutines.flow.flow
25+
import kotlinx.coroutines.flow.takeWhile
26+
import kotlinx.coroutines.launch
27+
import kotlin.coroutines.cancellation.CancellationException
28+
29+
fun <T> Flow<T>.takeUntilSignal(signal: Flow<Unit?>): Flow<T> = flow {
30+
try {
31+
coroutineScope {
32+
launch {
33+
signal.takeWhile { it == null }.collect()
34+
this@coroutineScope.cancel()
35+
}
36+
37+
collect {
38+
emit(it)
39+
}
40+
}
41+
} catch (e: CancellationException) {
42+
// ignore
43+
}
44+
}
45+
46+
fun <T> Flow<T>.cancelOnSignal(signal: Flow<Unit?>): Flow<T> = flow {
47+
coroutineScope {
48+
launch {
49+
signal.takeWhile { it == null }.collect()
50+
currentCoroutineContext().cancel()
51+
}
52+
53+
collect {
54+
emit(it)
55+
}
56+
currentCoroutineContext().cancel()
57+
}
58+
}

‎livekit-android-sdk/src/main/java/io/livekit/android/dagger/InternalBindsModule.kt‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* Copyright 2025 LiveKit, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "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+
* http://www.apache.org/licenses/LICENSE-2.0
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.livekit.android.dagger
18+
19+
import dagger.Binds
20+
import dagger.Module
21+
import io.livekit.android.room.datastream.incoming.IncomingDataStreamManager
22+
import io.livekit.android.room.datastream.incoming.IncomingDataStreamManagerImpl
23+
import io.livekit.android.room.datastream.outgoing.OutgoingDataStreamManager
24+
import io.livekit.android.room.datastream.outgoing.OutgoingDataStreamManagerImpl
25+
26+
/**
27+
* @suppress
28+
*/
29+
@Module
30+
abstract class InternalBindsModule {
31+
@Binds
32+
abstract fun incomingDataStreamManager(manager: IncomingDataStreamManagerImpl): IncomingDataStreamManager
33+
34+
@Binds
35+
abstract fun outgoingDataStreamManager(manager: OutgoingDataStreamManagerImpl): OutgoingDataStreamManager
36+
}

‎livekit-android-sdk/src/main/java/io/livekit/android/dagger/LiveKitComponent.kt‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2023-2024 LiveKit, Inc.
2+
* Copyright 2023-2025 LiveKit, Inc.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -38,6 +38,7 @@ import javax.inject.Singleton
3838
OverridesModule::class,
3939
AudioHandlerModule::class,
4040
MemoryModule::class,
41+
InternalBindsModule::class,
4142
],
4243
)
4344
interface LiveKitComponent {

‎livekit-android-sdk/src/main/java/io/livekit/android/room/RTCEngine.kt‎

Lines changed: 38 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ import io.livekit.android.util.LKLog
3838
import io.livekit.android.util.flowDelegate
3939
import io.livekit.android.util.nullSafe
4040
import io.livekit.android.util.withCheckLock
41+
import io.livekit.android.webrtc.DataChannelManager
4142
import io.livekit.android.webrtc.RTCStatsGetter
4243
import io.livekit.android.webrtc.copy
4344
import io.livekit.android.webrtc.isConnected
@@ -165,6 +166,10 @@ internal constructor(
165166
private var reliableDataChannelSub: DataChannel? = null
166167
private var lossyDataChannel: DataChannel? = null
167168
private var lossyDataChannelSub: DataChannel? = null
169+
private var reliableDataChannelManager: DataChannelManager? = null
170+
private var reliableDataChannelSubManager: DataChannelManager? = null
171+
private var lossyDataChannelManager: DataChannelManager? = null
172+
private var lossyDataChannelSubManager: DataChannelManager? = null
168173

169174
private var isSubscriberPrimary = false
170175
private var isClosed = true
@@ -406,19 +411,17 @@ internal constructor(
406411
subscriber?.closeBlocking()
407412
subscriber = null
408413

409-
fun DataChannel?.completeDispose() {
410-
this?.unregisterObserver()
411-
this?.close()
412-
this?.dispose()
413-
}
414-
415-
reliableDataChannel?.completeDispose()
414+
reliableDataChannelManager?.dispose()
415+
reliableDataChannelManager = null
416416
reliableDataChannel = null
417-
reliableDataChannelSub?.completeDispose()
417+
reliableDataChannelSubManager?.dispose()
418+
reliableDataChannelSubManager = null
418419
reliableDataChannelSub = null
419-
lossyDataChannel?.completeDispose()
420+
lossyDataChannelManager?.dispose()
421+
lossyDataChannelManager = null
420422
lossyDataChannel = null
421-
lossyDataChannelSub?.completeDispose()
423+
lossyDataChannelSubManager?.dispose()
424+
lossyDataChannelSubManager = null
422425
lossyDataChannelSub = null
423426
isSubscriberPrimary = false
424427
}
@@ -634,6 +637,22 @@ internal constructor(
634637
channel.send(buf)
635638
}
636639

640+
internal suspend fun waitForBufferStatusLow(kind: LivekitModels.DataPacket.Kind) {
641+
ensurePublisherConnected(kind)
642+
val manager = when (kind) {
643+
LivekitModels.DataPacket.Kind.RELIABLE -> reliableDataChannelManager
644+
LivekitModels.DataPacket.Kind.LOSSY -> lossyDataChannelManager
645+
LivekitModels.DataPacket.Kind.UNRECOGNIZED -> {
646+
throw IllegalArgumentException()
647+
}
648+
}
649+
650+
if (manager == null) {
651+
throw IllegalStateException("Not connected!")
652+
}
653+
manager.waitForBufferedAmountLow(DATA_CHANNEL_LOW_THRESHOLD.toLong())
654+
}
655+
637656
private suspend fun ensurePublisherConnected(kind: LivekitModels.DataPacket.Kind) {
638657
if (!isSubscriberPrimary) {
639658
return
@@ -802,6 +821,7 @@ internal constructor(
802821
fun onTranscriptionReceived(transcription: LivekitModels.Transcription)
803822
fun onLocalTrackSubscribed(trackSubscribed: LivekitRtc.TrackSubscribed)
804823
fun onRpcPacketReceived(dp: LivekitModels.DataPacket)
824+
fun onDataStreamPacket(dp: LivekitModels.DataPacket)
805825
}
806826

807827
companion object {
@@ -817,11 +837,13 @@ internal constructor(
817837
*/
818838
@VisibleForTesting
819839
const val LOSSY_DATA_CHANNEL_LABEL = "_lossy"
820-
internal const val MAX_DATA_PACKET_SIZE = 15360 // 15 KB
840+
internal const val MAX_DATA_PACKET_SIZE = 15 * 1024 // 15 KB
821841
private const val MAX_RECONNECT_RETRIES = 10
822842
private const val MAX_RECONNECT_TIMEOUT = 60 * 1000
823843
private const val MAX_ICE_CONNECT_TIMEOUT_MS = 20000
824844

845+
private const val DATA_CHANNEL_LOW_THRESHOLD = 64 * 1024 // 64 KB
846+
825847
internal val CONN_CONSTRAINTS = MediaConstraints().apply {
826848
with(optional) {
827849
add(MediaConstraints.KeyValuePair("DtlsSrtpKeyAgreement", "true"))
@@ -1079,16 +1101,11 @@ internal constructor(
10791101
LKLog.v { "invalid value for data packet" }
10801102
}
10811103

1082-
LivekitModels.DataPacket.ValueCase.STREAM_HEADER -> {
1083-
// TODO
1084-
}
1085-
1086-
LivekitModels.DataPacket.ValueCase.STREAM_CHUNK -> {
1087-
// TODO
1088-
}
1089-
1090-
LivekitModels.DataPacket.ValueCase.STREAM_TRAILER -> {
1091-
// TODO
1104+
LivekitModels.DataPacket.ValueCase.STREAM_HEADER,
1105+
LivekitModels.DataPacket.ValueCase.STREAM_CHUNK,
1106+
LivekitModels.DataPacket.ValueCase.STREAM_TRAILER,
1107+
-> {
1108+
listener?.onDataStreamPacket(dp)
10921109
}
10931110
}
10941111
}

‎livekit-android-sdk/src/main/java/io/livekit/android/room/Room.kt‎

Lines changed: 26 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ import io.livekit.android.e2ee.E2EEOptions
4242
import io.livekit.android.events.*
4343
import io.livekit.android.memory.CloseableManager
4444
import io.livekit.android.renderer.TextureViewRenderer
45+
import io.livekit.android.room.datastream.incoming.IncomingDataStreamManager
4546
import io.livekit.android.room.metrics.collectMetrics
4647
import io.livekit.android.room.network.NetworkCallbackManagerFactory
4748
import io.livekit.android.room.participant.*
@@ -106,7 +107,8 @@ constructor(
106107
private val regionUrlProviderFactory: RegionUrlProvider.Factory,
107108
private val connectionWarmer: ConnectionWarmer,
108109
private val audioRecordPrewarmer: AudioRecordPrewarmer,
109-
) : RTCEngine.Listener, ParticipantListener {
110+
private val incomingDataStreamManager: IncomingDataStreamManager,
111+
) : RTCEngine.Listener, ParticipantListener, IncomingDataStreamManager by incomingDataStreamManager {
110112

111113
private lateinit var coroutineScope: CoroutineScope
112114
private val eventBus = BroadcastEventBus<RoomEvent>()
@@ -907,6 +909,7 @@ constructor(
907909
name = null
908910
isRecording = false
909911
sidToIdentity.clear()
912+
incomingDataStreamManager.clearOpenStreams()
910913
}
911914

912915
private fun sendSyncState() {
@@ -1190,6 +1193,28 @@ constructor(
11901193
participant?.onDataReceived(data, topic)
11911194
}
11921195

1196+
/**
1197+
* @suppress
1198+
*/
1199+
override fun onDataStreamPacket(dp: LivekitModels.DataPacket) {
1200+
when (dp.valueCase) {
1201+
LivekitModels.DataPacket.ValueCase.STREAM_HEADER -> {
1202+
incomingDataStreamManager.handleStreamHeader(dp.streamHeader, Participant.Identity(dp.participantIdentity))
1203+
}
1204+
1205+
LivekitModels.DataPacket.ValueCase.STREAM_CHUNK -> {
1206+
incomingDataStreamManager.handleDataChunk(dp.streamChunk)
1207+
}
1208+
1209+
LivekitModels.DataPacket.ValueCase.STREAM_TRAILER -> {
1210+
incomingDataStreamManager.handleStreamTrailer(dp.streamTrailer)
1211+
}
1212+
1213+
// Ignore other cases.
1214+
else -> {}
1215+
}
1216+
}
1217+
11931218
/**
11941219
* @suppress
11951220
*/

‎livekit-android-sdk/src/main/java/io/livekit/android/room/datastream/StreamException.kt‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
/*
2+
* Copyright 2025 LiveKit, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "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+
* http://www.apache.org/licenses/LICENSE-2.0
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.livekit.android.room.datastream
18+
19+
sealed class StreamException(message: String? = null) : Exception(message) {
20+
class AlreadyOpenedException : StreamException()
21+
class AbnormalEndException(message: String?) : StreamException(message)
22+
class DecodeFailedException : StreamException()
23+
class LengthExceededException : StreamException()
24+
class IncompleteException : StreamException()
25+
class TerminatedException : StreamException()
26+
class UnknownStreamException : StreamException()
27+
class NotDirectoryException : StreamException()
28+
class FileInfoUnavailableException : StreamException()
29+
}

0 commit comments

Comments
 (0)