Reliable data channel (#738)

* e2e reliablility for data channel

* fix tests

* changeset

* add .idea to gitignore

* change to patch change

+---
+"client-sdk-android": patch
+---
+
+E2E reliability for data channels with resending after reconnects

 *.iml
 *.iml
 .gradle
 .gradle
 /local.properties
 /local.properties
-/.idea/caches
-/.idea/libraries
-/.idea/modules.xml
-/.idea/workspace.xml
-/.idea/navEditor.xml
-/.idea/assetWizardSettings.xml
-/.idea/deploymentTargetDropDown.xml
-/.idea/misc.xml
-/.idea/gradle.xml
-/.idea/runConfigurations.xml
-/.idea/deploymentTargetSelector.xml
+/.idea
 .DS_Store
 .DS_Store
 /build
 /build
 /captures
 /captures

 <component name="ProjectCodeStyleConfiguration">
 <component name="ProjectCodeStyleConfiguration">
   <code_scheme name="Project" version="173">
   <code_scheme name="Project" version="173">
+    <JavaCodeStyleSettings>
+      <option name="IMPORT_LAYOUT_TABLE">
+        <value>
+          <package name="" withSubpackages="true" static="false" module="true" />
+          <package name="android" withSubpackages="true" static="true" />
+          <package name="androidx" withSubpackages="true" static="true" />
+          <package name="com" withSubpackages="true" static="true" />
+          <package name="junit" withSubpackages="true" static="true" />
+          <package name="net" withSubpackages="true" static="true" />
+          <package name="org" withSubpackages="true" static="true" />
+          <package name="java" withSubpackages="true" static="true" />
+          <package name="javax" withSubpackages="true" static="true" />
+          <package name="" withSubpackages="true" static="true" />
+          <emptyLine />
+          <package name="android" withSubpackages="true" static="false" />
+          <emptyLine />
+          <package name="androidx" withSubpackages="true" static="false" />
+          <emptyLine />
+          <package name="com" withSubpackages="true" static="false" />
+          <emptyLine />
+          <package name="junit" withSubpackages="true" static="false" />
+          <emptyLine />
+          <package name="net" withSubpackages="true" static="false" />
+          <emptyLine />
+          <package name="org" withSubpackages="true" static="false" />
+          <emptyLine />
+          <package name="java" withSubpackages="true" static="false" />
+          <emptyLine />
+          <package name="javax" withSubpackages="true" static="false" />
+          <emptyLine />
+          <package name="" withSubpackages="true" static="false" />
+          <emptyLine />
+        </value>
+      </option>
+    </JavaCodeStyleSettings>
     <JetCodeStyleSettings>
     <JetCodeStyleSettings>
       <option name="CODE_STYLE_DEFAULTS" value="KOTLIN_OFFICIAL" />
       <option name="CODE_STYLE_DEFAULTS" value="KOTLIN_OFFICIAL" />
     </JetCodeStyleSettings>
     </JetCodeStyleSettings>

@@ -299,6 +299,7 @@ enum class DisconnectReason {
     USER_REJECTED,
     USER_REJECTED,
     SIP_TRUNK_FAILURE,
     SIP_TRUNK_FAILURE,
     CONNECTION_TIMEOUT,
     CONNECTION_TIMEOUT,
+    MEDIA_FAILURE,
 }
 }
 
 
 /**
 /**
@@ -320,6 +321,7 @@ fun LivekitModels.DisconnectReason?.convert(): DisconnectReason {
         LivekitModels.DisconnectReason.USER_REJECTED -> DisconnectReason.USER_REJECTED
         LivekitModels.DisconnectReason.USER_REJECTED -> DisconnectReason.USER_REJECTED
         LivekitModels.DisconnectReason.SIP_TRUNK_FAILURE -> DisconnectReason.SIP_TRUNK_FAILURE
         LivekitModels.DisconnectReason.SIP_TRUNK_FAILURE -> DisconnectReason.SIP_TRUNK_FAILURE
         LivekitModels.DisconnectReason.CONNECTION_TIMEOUT -> DisconnectReason.CONNECTION_TIMEOUT
         LivekitModels.DisconnectReason.CONNECTION_TIMEOUT -> DisconnectReason.CONNECTION_TIMEOUT
+        LivekitModels.DisconnectReason.MEDIA_FAILURE -> DisconnectReason.MEDIA_FAILURE
         LivekitModels.DisconnectReason.UNKNOWN_REASON,
         LivekitModels.DisconnectReason.UNKNOWN_REASON,
         LivekitModels.DisconnectReason.UNRECOGNIZED,
         LivekitModels.DisconnectReason.UNRECOGNIZED,
         null,
         null,

@@ -36,10 +36,13 @@ import io.livekit.android.util.CloseableCoroutineScope
 import io.livekit.android.util.Either
 import io.livekit.android.util.Either
 import io.livekit.android.util.FlowObservable
 import io.livekit.android.util.FlowObservable
 import io.livekit.android.util.LKLog
 import io.livekit.android.util.LKLog
+import io.livekit.android.util.TTLMap
 import io.livekit.android.util.flowDelegate
 import io.livekit.android.util.flowDelegate
 import io.livekit.android.util.nullSafe
 import io.livekit.android.util.nullSafe
 import io.livekit.android.util.withCheckLock
 import io.livekit.android.util.withCheckLock
 import io.livekit.android.webrtc.DataChannelManager
 import io.livekit.android.webrtc.DataChannelManager
+import io.livekit.android.webrtc.DataPacketBuffer
+import io.livekit.android.webrtc.DataPacketItem
 import io.livekit.android.webrtc.RTCStatsGetter
 import io.livekit.android.webrtc.RTCStatsGetter
 import io.livekit.android.webrtc.copy
 import io.livekit.android.webrtc.copy
 import io.livekit.android.webrtc.isConnected
 import io.livekit.android.webrtc.isConnected
@@ -171,6 +174,11 @@ internal constructor(
     private var lossyDataChannelManager: DataChannelManager? = null
     private var lossyDataChannelManager: DataChannelManager? = null
     private var lossyDataChannelSubManager: DataChannelManager? = null
     private var lossyDataChannelSubManager: DataChannelManager? = null
 
 
+    private val reliableStateLock = Object()
+    private var reliableDataSequence: Int = 1
+    private val reliableMessageBuffer = DataPacketBuffer(RELIABLE_RETRY_AMOUNT)
+    private val reliableReceivedState = TTLMap<String, Int>(RELIABLE_RECEIVE_STATE_TTL_MS)
+
     private var isSubscriberPrimary = false
     private var isSubscriberPrimary = false
     private var isClosed = true
     private var isClosed = true
 
 
@@ -403,6 +411,12 @@ internal constructor(
         abortPendingPublishTracks()
         abortPendingPublishTracks()
         closeResources(reason)
         closeResources(reason)
         connectionState = ConnectionState.DISCONNECTED
         connectionState = ConnectionState.DISCONNECTED
+
+        synchronized(reliableStateLock) {
+            reliableDataSequence = 1
+            reliableMessageBuffer.clear()
+            reliableReceivedState.clear()
+        }
     }
     }
 
 
     private fun closeResources(reason: String) {
     private fun closeResources(reason: String) {
@@ -506,6 +520,7 @@ internal constructor(
                     ReconnectType.FORCE_FULL_RECONNECT -> true
                     ReconnectType.FORCE_FULL_RECONNECT -> true
                 }
                 }
 
 
+                var lastMessageSeq: Int? = null
                 val connectOptions = connectOptions ?: ConnectOptions()
                 val connectOptions = connectOptions ?: ConnectOptions()
                 if (isFullReconnect) {
                 if (isFullReconnect) {
                     LKLog.v { "Attempting full reconnect." }
                     LKLog.v { "Attempting full reconnect." }
@@ -539,6 +554,7 @@ internal constructor(
                             val rtcConfig = makeRTCConfig(Either.Right(reconnectResponse), connectOptions)
                             val rtcConfig = makeRTCConfig(Either.Right(reconnectResponse), connectOptions)
                             subscriber?.updateRTCConfig(rtcConfig)
                             subscriber?.updateRTCConfig(rtcConfig)
                             publisher?.updateRTCConfig(rtcConfig)
                             publisher?.updateRTCConfig(rtcConfig)
+                            lastMessageSeq = reconnectResponse.lastMessageSeq
                         }
                         }
                         client.onReadyForResponses()
                         client.onReadyForResponses()
                     } catch (e: Exception) {
                     } catch (e: Exception) {
@@ -590,6 +606,9 @@ internal constructor(
                 if (connectionState == ConnectionState.CONNECTED &&
                 if (connectionState == ConnectionState.CONNECTED &&
                     (!hasPublished || publisher?.isConnected() == true)
                     (!hasPublished || publisher?.isConnected() == true)
                 ) {
                 ) {
+                    if (lastMessageSeq != null) {
+                        resendReliableMessagesForResume(lastMessageSeq)
+                    }
                     // Is connected, notify and return.
                     // Is connected, notify and return.
                     regionUrlProvider?.clearAttemptedRegions()
                     regionUrlProvider?.clearAttemptedRegions()
                     client.onPCConnected()
                     client.onPCConnected()
@@ -630,21 +649,62 @@ internal constructor(
 
 
     @CheckResult
     @CheckResult
     internal suspend fun sendData(dataPacket: LivekitModels.DataPacket): Result<Unit> {
     internal suspend fun sendData(dataPacket: LivekitModels.DataPacket): Result<Unit> {
-        try {
-            ensurePublisherConnected(dataPacket.kind)
+        ensurePublisherConnected(dataPacket.kind)
+
+        fun sendDataImpl(dataPacket: LivekitModels.DataPacket): Result<Unit> {
+            try {
+                // Redeclare to make variable
+                var dataPacket = dataPacket
+                if (dataPacket.kind == LivekitModels.DataPacket.Kind.RELIABLE) {
+                    dataPacket = dataPacket.toBuilder()
+                        .setSequence(reliableDataSequence)
+                        .build()
+                    reliableDataSequence++
+                }
 
 
-            val buf = DataChannel.Buffer(
-                ByteBuffer.wrap(dataPacket.toByteArray()),
-                true,
-            )
+                val byteBuffer = ByteBuffer.wrap(dataPacket.toByteArray())
+
+                if (dataPacket.kind == LivekitModels.DataPacket.Kind.RELIABLE) {
+                    reliableMessageBuffer.queue(DataPacketItem(byteBuffer, dataPacket.sequence))
+                    if (this.connectionState == ConnectionState.RECONNECTING) {
+                        return Result.success(Unit)
+                    }
+                }
+                val buf = DataChannel.Buffer(
+                    byteBuffer,
+                    true,
+                )
+                val channel = dataChannelForKind(dataPacket.kind)
+                    ?: throw RoomException.ConnectException("channel not established for ${dataPacket.kind.name}")
 
 
-            val channel = dataChannelForKind(dataPacket.kind)
-                ?: throw RoomException.ConnectException("channel not established for ${dataPacket.kind.name}")
+                channel.send(buf)
+            } catch (e: Exception) {
+                return Result.failure(e)
+            }
+            return Result.success(Unit)
+        }
 
 
-            channel.send(buf)
-        } catch (e: Exception) {
-            return Result.failure(e)
+        if (dataPacket.kind == LivekitModels.DataPacket.Kind.RELIABLE) {
+            synchronized(reliableStateLock) {
+                return sendDataImpl(dataPacket)
+            }
+        } else {
+            return sendDataImpl(dataPacket)
         }
         }
+    }
+
+    internal suspend fun resendReliableMessagesForResume(lastMessageSeq: Int): Result<Unit> {
+        ensurePublisherConnected(LivekitModels.DataPacket.Kind.RELIABLE)
+        val channel = dataChannelForKind(LivekitModels.DataPacket.Kind.RELIABLE)
+            ?: return Result.failure(NullPointerException("reliable channel not established!"))
+
+        synchronized(reliableStateLock) {
+            reliableMessageBuffer.popToSequence(lastMessageSeq)
+            reliableMessageBuffer.getAll().forEach { item ->
+                channel.send(DataChannel.Buffer(item.data, true))
+            }
+        }
+
         return Result.success(Unit)
         return Result.success(Unit)
     }
     }
 
 
@@ -858,7 +918,10 @@ internal constructor(
         private const val MAX_RECONNECT_TIMEOUT = 60 * 1000
         private const val MAX_RECONNECT_TIMEOUT = 60 * 1000
         private const val MAX_ICE_CONNECT_TIMEOUT_MS = 20000
         private const val MAX_ICE_CONNECT_TIMEOUT_MS = 20000
 
 
-        private const val DATA_CHANNEL_LOW_THRESHOLD = 64 * 1024 // 64 KB
+        private const val DATA_CHANNEL_LOW_THRESHOLD = 2 * 1024 * 1024 // 64 KB
+
+        private val RELIABLE_RECEIVE_STATE_TTL_MS = 30.seconds
+        private val RELIABLE_RETRY_AMOUNT = (DATA_CHANNEL_LOW_THRESHOLD * 1.25).toLong()
 
 
         internal val CONN_CONSTRAINTS = MediaConstraints().apply {
         internal val CONN_CONSTRAINTS = MediaConstraints().apply {
             with(optional) {
             with(optional) {
@@ -1080,6 +1143,17 @@ internal constructor(
             return
             return
         }
         }
         val dp = LivekitModels.DataPacket.parseFrom(ByteString.copyFrom(buffer.data))
         val dp = LivekitModels.DataPacket.parseFrom(ByteString.copyFrom(buffer.data))
+
+        if (dp.sequence > 0 && dp.participantSid.isNotEmpty()) {
+            synchronized(reliableStateLock) {
+                val lastSeq = reliableReceivedState[dp.participantSid]
+                if (lastSeq != null && dp.sequence <= lastSeq) {
+                    // ignore duplicate or out-of-order packets in reliable channel
+                    return
+                }
+                this.reliableReceivedState[dp.participantSid] = dp.sequence
+            }
+        }
         when (dp.valueCase) {
         when (dp.valueCase) {
             LivekitModels.DataPacket.ValueCase.SPEAKER -> {
             LivekitModels.DataPacket.ValueCase.SPEAKER -> {
                 listener?.onActiveSpeakersUpdate(dp.speaker.speakersList)
                 listener?.onActiveSpeakersUpdate(dp.speaker.speakersList)
@@ -1145,8 +1219,13 @@ internal constructor(
         subscription: LivekitRtc.UpdateSubscription,
         subscription: LivekitRtc.UpdateSubscription,
         publishedTracks: List<LivekitRtc.TrackPublishedResponse>,
         publishedTracks: List<LivekitRtc.TrackPublishedResponse>,
     ) {
     ) {
-        val answer = runBlocking {
-            subscriber?.withPeerConnection { localDescription?.toProtoSessionDescription() }
+        var answer: LivekitRtc.SessionDescription? = null
+        var offer: LivekitRtc.SessionDescription? = null
+        runBlocking {
+            subscriber?.withPeerConnection {
+                answer = localDescription?.toProtoSessionDescription()
+                offer = remoteDescription?.toProtoSessionDescription()
+            }
         }
         }
 
 
         val dataChannelInfos = LivekitModels.DataPacket.Kind.values()
         val dataChannelInfos = LivekitModels.DataPacket.Kind.values()
@@ -1160,13 +1239,25 @@ internal constructor(
                     .build()
                     .build()
             }
             }
 
 
+        val dataChannelReceiveStates = this.reliableReceivedState.map { (participantSid, sequence) ->
+            with(LivekitRtc.DataChannelReceiveState.newBuilder()) {
+                publisherSid = participantSid
+                lastSeq = sequence
+                build()
+            }
+        }
+
         val syncState = with(LivekitRtc.SyncState.newBuilder()) {
         val syncState = with(LivekitRtc.SyncState.newBuilder()) {
             if (answer != null) {
             if (answer != null) {
                 setAnswer(answer)
                 setAnswer(answer)
             }
             }
+            if (offer != null) {
+                setOffer(offer)
+            }
             setSubscription(subscription)
             setSubscription(subscription)
             addAllPublishTracks(publishedTracks)
             addAllPublishTracks(publishedTracks)
             addAllDataChannels(dataChannelInfos)
             addAllDataChannels(dataChannelInfos)
+            addAllDatachannelReceiveStates(dataChannelReceiveStates)
             build()
             build()
         }
         }
 
 

+/*
+ * Copyright 2025 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 io.livekit.android.util
+
+import android.os.SystemClock
+import kotlin.time.Duration
+import kotlin.time.DurationUnit
+
+/**
+ * @suppress
+ */
+class TTLMap<K, V>(
+    val ttl: Duration,
+    private val clock: () -> Long = { SystemClock.elapsedRealtime() },
+) : MutableMap<K, V> {
+
+    private data class TTLItem<V>(val value: V, val expiresAt: Long)
+
+    private val map = mutableMapOf<K, TTLItem<V>>()
+    private val lastCleanup = getNow()
+
+    override fun get(key: K): V? {
+        val item = map[key]
+
+        if (item == null) {
+            return null
+        }
+
+        if (item.expiresAt < getNow()) {
+            map.remove(key)
+            return null
+        }
+        return item.value
+    }
+
+    override val size: Int
+        get() {
+            cleanup()
+            return map.size
+        }
+
+    override fun containsKey(key: K): Boolean = get(key) != null
+    override fun containsValue(value: V): Boolean = values.contains(value)
+
+    override fun isEmpty(): Boolean {
+        cleanup()
+        return map.isEmpty()
+    }
+
+    private data class MutableEntry<K, V>(override val key: K, override var value: V) : MutableMap.MutableEntry<K, V> {
+        override fun setValue(newValue: V): V {
+            val old = this.value
+            this.value = newValue
+            return old
+        }
+    }
+
+    override val entries: MutableSet<MutableMap.MutableEntry<K, V>>
+        get() {
+            cleanup()
+            return map.entries
+                .map { (key, value) -> MutableEntry(key, value.value) }
+                .toMutableSet()
+        }
+
+    override val keys: MutableSet<K>
+        get() {
+            cleanup()
+            return map.keys
+        }
+
+    override val values: MutableCollection<V>
+        get() {
+            cleanup()
+            return map.values
+                .map { (value, _) -> value }
+                .toMutableList()
+        }
+
+    override fun clear() {
+        map.clear()
+    }
+
+    override fun put(key: K, value: V): V? {
+        val now = getNow()
+        val ttlMs = ttl.toLong(DurationUnit.MILLISECONDS)
+        if (now - lastCleanup > ttlMs / 2) {
+            cleanup()
+        }
+
+        val expiresAt = now + ttlMs
+
+        map[key] = TTLItem(value, expiresAt)
+        return value
+    }
+
+    override fun putAll(from: Map<out K, V>) {
+        from.iterator().forEach { (key, value) ->
+            put(key, value)
+        }
+    }
+
+    override fun remove(key: K): V? {
+        return map.remove(key)?.value
+    }
+
+    fun cleanup() {
+        val now = getNow()
+
+        val iterator = map.iterator()
+        while (iterator.hasNext()) {
+            val (_, entry) = iterator.next()
+            if (entry.expiresAt < now) {
+                iterator.remove()
+            }
+        }
+    }
+
+    private fun getNow() = clock()
+}

+/*
+ * Copyright 2025 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 io.livekit.android.webrtc
+
+import java.nio.ByteBuffer
+import java.util.Deque
+import java.util.LinkedList
+
+/** @suppress */
+data class DataPacketItem(val data: ByteBuffer, val sequence: Int)
+
+/** @suppress */
+class DataPacketBuffer(private val extraCapacity: Long = 0L) {
+    private val buffer: Deque<DataPacketItem> = LinkedList()
+    private var totalSize = 0L
+
+    @Synchronized
+    fun clear() {
+        buffer.clear()
+        totalSize = 0L
+    }
+
+    @Synchronized
+    fun queue(item: DataPacketItem) {
+        buffer.add(item)
+        totalSize += item.data.capacity()
+    }
+
+    @Synchronized
+    fun dequeue(): DataPacketItem? {
+        if (buffer.isEmpty()) {
+            return null
+        }
+        val item = buffer.removeFirst()
+        totalSize -= item.data.capacity()
+        return item
+    }
+
+    @Synchronized
+    fun getAll(): List<DataPacketItem> {
+        return buffer.toList()
+    }
+
+    /**
+     * Pops until [sequence] (inclusive).
+     */
+    @Synchronized
+    fun popToSequence(sequence: Int): List<DataPacketItem> {
+        val retList = mutableListOf<DataPacketItem>()
+        while (buffer.isNotEmpty()) {
+            val first = buffer.first
+            if (first.sequence <= sequence) {
+                val item = dequeue()
+                retList.add(item!!)
+            } else {
+                break
+            }
+        }
+
+        return retList
+    }
+
+    @Synchronized
+    fun trim(size: Long) {
+        while (buffer.isNotEmpty() && totalSize > size + extraCapacity) {
+            dequeue()
+        }
+    }
+
+    /**
+     * Returns the total byte size of the items in the buffer.
+     */
+    @Synchronized
+    fun byteSize() = totalSize
+
+    /**
+     * Returns the number of items in the buffer
+     */
+    @Synchronized
+    fun size() = buffer.size
+}

@@ -133,6 +133,7 @@ object TestData {
                 forceRelay = LivekitModels.ClientConfigSetting.ENABLED
                 forceRelay = LivekitModels.ClientConfigSetting.ENABLED
                 build()
                 build()
             }
             }
+            lastMessageSeq = 1
             build()
             build()
         }
         }
         build()
         build()

@@ -73,6 +73,7 @@ class ProtoConverterTest(
                 LivekitRtc.SessionDescription::class.java,
                 LivekitRtc.SessionDescription::class.java,
                 SessionDescription::class.java,
                 SessionDescription::class.java,
                 mapping = mapOf("sdp" to "description"),
                 mapping = mapOf("sdp" to "description"),
+                whitelist = listOf("id"),
             ),
             ),
         )
         )
 
 

 /*
 /*
- * Copyright 2023-2024 LiveKit, Inc.
+ * Copyright 2023-2025 LiveKit, Inc.
  *
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * you may not use this file except in compliance with the License.
@@ -16,7 +16,9 @@
 
 
 package io.livekit.android.room
 package io.livekit.android.room
 
 
+import io.livekit.android.room.track.DataPublishReliability
 import io.livekit.android.test.MockE2ETest
 import io.livekit.android.test.MockE2ETest
+import io.livekit.android.test.mock.MockDataChannel
 import io.livekit.android.test.mock.TestData
 import io.livekit.android.test.mock.TestData
 import io.livekit.android.test.mock.room.track.createMockLocalAudioTrack
 import io.livekit.android.test.mock.room.track.createMockLocalAudioTrack
 import io.livekit.android.test.util.toPBByteString
 import io.livekit.android.test.util.toPBByteString
@@ -25,6 +27,7 @@ import livekit.LivekitRtc
 import livekit.org.webrtc.PeerConnection
 import livekit.org.webrtc.PeerConnection
 import org.junit.Assert
 import org.junit.Assert
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
 import org.junit.Test
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runner.RunWith
 import org.robolectric.RobolectricTestRunner
 import org.robolectric.RobolectricTestRunner
@@ -77,6 +80,28 @@ class RoomReconnectionMockE2ETest : MockE2ETest() {
     }
     }
 
 
     @Test
     @Test
+    fun softReconnectResendsPackets() = runTest {
+        room.setReconnectionType(ReconnectType.FORCE_SOFT_RECONNECT)
+
+        connect()
+
+        for (i in 1..5) {
+            assertTrue(room.localParticipant.publishData(ByteArray(i), reliability = DataPublishReliability.RELIABLE).isSuccess)
+        }
+        disconnectPeerConnection()
+        // Wait so that the reconnect job properly starts first.
+        testScheduler.advanceTimeBy(1000)
+        reconnectWebsocket()
+        connectPeerConnection()
+
+        testScheduler.advanceUntilIdle()
+        val pubPeerConnection = getPublisherPeerConnection()
+        val pubDataChannel = pubPeerConnection.dataChannels[RTCEngine.RELIABLE_DATA_CHANNEL_LABEL] as MockDataChannel
+
+        assertEquals(5, pubDataChannel.sentBuffers.size)
+    }
+
+    @Test
     fun softReconnectConfiguration() = runTest {
     fun softReconnectConfiguration() = runTest {
         room.setReconnectionType(ReconnectType.FORCE_SOFT_RECONNECT)
         room.setReconnectionType(ReconnectType.FORCE_SOFT_RECONNECT)
         connect()
         connect()

@@ -20,10 +20,12 @@ import android.Manifest
 import android.app.Application
 import android.app.Application
 import android.content.Context
 import android.content.Context
 import androidx.test.core.app.ApplicationProvider
 import androidx.test.core.app.ApplicationProvider
+import com.google.protobuf.ByteString
 import io.livekit.android.audio.AudioProcessorInterface
 import io.livekit.android.audio.AudioProcessorInterface
 import io.livekit.android.events.ParticipantEvent
 import io.livekit.android.events.ParticipantEvent
 import io.livekit.android.events.RoomEvent
 import io.livekit.android.events.RoomEvent
 import io.livekit.android.room.DefaultsManager
 import io.livekit.android.room.DefaultsManager
+import io.livekit.android.room.RTCEngine
 import io.livekit.android.room.track.LocalVideoTrack
 import io.livekit.android.room.track.LocalVideoTrack
 import io.livekit.android.room.track.LocalVideoTrackOptions
 import io.livekit.android.room.track.LocalVideoTrackOptions
 import io.livekit.android.room.track.Track
 import io.livekit.android.room.track.Track
@@ -35,6 +37,7 @@ import io.livekit.android.test.assert.assertIsClassList
 import io.livekit.android.test.coroutines.toListUntilSignal
 import io.livekit.android.test.coroutines.toListUntilSignal
 import io.livekit.android.test.events.EventCollector
 import io.livekit.android.test.events.EventCollector
 import io.livekit.android.test.mock.MockAudioProcessingController
 import io.livekit.android.test.mock.MockAudioProcessingController
+import io.livekit.android.test.mock.MockDataChannel
 import io.livekit.android.test.mock.MockEglBase
 import io.livekit.android.test.mock.MockEglBase
 import io.livekit.android.test.mock.MockVideoCapturer
 import io.livekit.android.test.mock.MockVideoCapturer
 import io.livekit.android.test.mock.MockVideoStreamTrack
 import io.livekit.android.test.mock.MockVideoStreamTrack
@@ -55,6 +58,7 @@ import kotlinx.coroutines.test.StandardTestDispatcher
 import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.advanceUntilIdle
 import livekit.LivekitModels
 import livekit.LivekitModels
 import livekit.LivekitModels.AudioTrackFeature
 import livekit.LivekitModels.AudioTrackFeature
+import livekit.LivekitModels.DataPacket
 import livekit.LivekitRtc
 import livekit.LivekitRtc
 import livekit.LivekitRtc.SubscribedCodec
 import livekit.LivekitRtc.SubscribedCodec
 import livekit.LivekitRtc.SubscribedQuality
 import livekit.LivekitRtc.SubscribedQuality
@@ -647,4 +651,22 @@ class LocalParticipantMockE2ETest : MockE2ETest() {
         assertTrue(collectedList[1])
         assertTrue(collectedList[1])
         assertFalse(collectedList[2])
         assertFalse(collectedList[2])
     }
     }
+
+    @Test
+    fun publishData() = runTest {
+        connect()
+        val pubPeerConnection = getPublisherPeerConnection()
+        val pubDataChannel = pubPeerConnection.dataChannels[RTCEngine.RELIABLE_DATA_CHANNEL_LABEL] as MockDataChannel
+
+        val data = "hello".toByteArray()
+        assertTrue(room.localParticipant.publishData(data).isSuccess)
+
+        assertEquals(1, pubDataChannel.sentBuffers.size)
+
+        val headerPacket = DataPacket.parseFrom(ByteString.copyFrom(pubDataChannel.sentBuffers[0].data))
+        assertEquals(1, headerPacket.sequence)
+        assertTrue(headerPacket.hasUser())
+
+        assertTrue(headerPacket.user.payload.toByteArray().contentEquals(data))
+    }
 }
 }

+/*
+ * Copyright 2025 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 io.livekit.android.room.webrtc
+
+import io.livekit.android.test.BaseTest
+import io.livekit.android.webrtc.DataPacketBuffer
+import io.livekit.android.webrtc.DataPacketItem
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNotNull
+import org.junit.Before
+import org.junit.Test
+import java.nio.ByteBuffer
+
+class DataPacketBufferTest : BaseTest() {
+
+    lateinit var buffer: DataPacketBuffer
+
+    @Before
+    fun setup() {
+        buffer = DataPacketBuffer(5)
+    }
+
+    fun fillWithTestValues() {
+        for (i in 1..5) {
+            val bytes = ByteArray(i)
+            buffer.queue(DataPacketItem(ByteBuffer.wrap(bytes), i))
+        }
+    }
+
+    @Test
+    fun queue() {
+        fillWithTestValues()
+        assertEquals(15, buffer.byteSize())
+        assertEquals(5, buffer.size())
+    }
+
+    @Test
+    fun dequeue() {
+        fillWithTestValues()
+        val list = mutableListOf<DataPacketItem>()
+        for (i in 1..5) {
+            val item = buffer.dequeue()
+            assertNotNull(item)
+
+            list.add(item!!)
+        }
+
+        list.forEachIndexed { index, item ->
+            assertEquals(index + 1, item.sequence)
+            assertEquals(index + 1, item.data.capacity())
+        }
+    }
+
+    @Test
+    fun clear() {
+        fillWithTestValues()
+        buffer.clear()
+        assertEquals(0, buffer.byteSize())
+        assertEquals(0, buffer.size())
+    }
+
+    @Test
+    fun getAll() {
+        fillWithTestValues()
+        val list = buffer.getAll()
+
+        assertEquals(5, list.size)
+        list.forEachIndexed { index, item ->
+            assertEquals(index + 1, item.sequence)
+            assertEquals(index + 1, item.data.capacity())
+        }
+    }
+
+    @Test
+    fun trim() {
+        fillWithTestValues()
+        buffer.trim(9)
+        // extra capacity is set to 5, so only the 1st packet is dropped.
+        assertEquals(14, buffer.byteSize())
+        assertEquals(4, buffer.size())
+    }
+
+    @Test
+    fun popToSequence() {
+        fillWithTestValues()
+        buffer.popToSequence(3)
+
+        assertEquals(9, buffer.byteSize())
+        assertEquals(2, buffer.size())
+    }
+}

+/*
+ * Copyright 2025 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 io.livekit.android.util
+
+import io.livekit.android.test.BaseTest
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertNull
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import kotlin.time.Duration.Companion.milliseconds
+
+class TTLMapTest : BaseTest() {
+    lateinit var map: TTLMap<String, String>
+    var time = 0L
+
+    @Before
+    fun setup() {
+        map = TTLMap(TTL, { time })
+        time = 0L
+    }
+
+    fun expire() {
+        time += 1 + TTL.inWholeMilliseconds
+    }
+
+    @Test
+    fun getNormal() {
+        map[KEY] = VALUE
+        assertEquals(map[KEY], VALUE)
+    }
+
+    @Test
+    fun getExpired() {
+        map[KEY] = VALUE
+        expire()
+        assertNull("map key was not deleted!", map[KEY])
+    }
+
+    @Test
+    fun isEmptyExpired() {
+        map[KEY] = VALUE
+        expire()
+        assertTrue(map.isEmpty())
+    }
+
+    @Test
+    fun containsKeyNormal() {
+        map[KEY] = VALUE
+        assertTrue(map.containsKey(KEY))
+    }
+
+    @Test
+    fun containsKeyExpired() {
+        map[KEY] = VALUE
+        expire()
+        assertFalse(map.containsKey(KEY))
+    }
+
+    @Test
+    fun containsValueNormal() {
+        map[KEY] = VALUE
+        assertTrue(map.containsValue(VALUE))
+    }
+
+    @Test
+    fun containsValueExpired() {
+        map[KEY] = VALUE
+        expire()
+        assertFalse(map.containsValue(VALUE))
+    }
+
+    @Test
+    fun clear() {
+        map[KEY] = VALUE
+        map.clear()
+        assertTrue(map.isEmpty())
+    }
+
+    @Test
+    fun remove() {
+        map[KEY] = VALUE
+        map.remove(KEY)
+        assertFalse(map.containsKey(KEY))
+    }
+
+    @Test
+    fun cleanup() {
+        map[KEY] = VALUE
+        map.cleanup()
+        assertEquals(map[KEY], VALUE)
+    }
+
+    @Test
+    fun cleanupExpired() {
+        map[KEY] = VALUE
+        expire()
+        map.cleanup()
+        assertTrue(map.isEmpty())
+    }
+
+    @Test
+    fun sizeNormal() {
+        map[KEY] = VALUE
+        assertEquals(1, map.size)
+    }
+
+    @Test
+    fun sizeExpired() {
+        map[KEY] = VALUE
+        expire()
+        assertEquals(0, map.size)
+    }
+
+    companion object {
+        val TTL = 1000.milliseconds
+
+        val KEY = "hello"
+        val VALUE = "world"
+    }
+}

--Subproject commit 499c17c48063582ac2af0a021827fab18356cc29
++Subproject commit 7276243574cd9bd8a621700d2f696cda5fd1b6bd