Support data packet topics (#213)

@@ -98,7 +98,11 @@ sealed class ParticipantEvent(open val participant: Participant) : Event() {
     /**
     /**
      * Received data published by another participant
      * Received data published by another participant
      */
      */
-    class DataReceived(override val participant: RemoteParticipant, val data: ByteArray) : ParticipantEvent(participant)
+    class DataReceived(
+        override val participant: RemoteParticipant,
+        val data: ByteArray,
+        val topic: String?
+    ) : ParticipantEvent(participant)
 
 
     /**
     /**
      * A track's stream state has changed.
      * A track's stream state has changed.

@@ -145,7 +145,8 @@ sealed class RoomEvent(val room: Room) : Event() {
      * @param data the published data
      * @param data the published data
      * @param participant the participant if available
      * @param participant the participant if available
      */
      */
-    class DataReceived(room: Room, val data: ByteArray, val participant: RemoteParticipant?) : RoomEvent(room)
+    class DataReceived(room: Room, val data: ByteArray, val participant: RemoteParticipant?, val topic: String?) :
+        RoomEvent(room)
 
 
     /**
     /**
      * The connection quality for a participant has changed.
      * The connection quality for a participant has changed.

@@ -722,10 +722,15 @@ constructor(
     override fun onUserPacket(packet: LivekitModels.UserPacket, kind: LivekitModels.DataPacket.Kind) {
     override fun onUserPacket(packet: LivekitModels.UserPacket, kind: LivekitModels.DataPacket.Kind) {
         val participant = remoteParticipants[packet.participantSid]
         val participant = remoteParticipants[packet.participantSid]
         val data = packet.payload.toByteArray()
         val data = packet.payload.toByteArray()
+        val topic = if (packet.hasTopic()) {
+            packet.topic
+        } else {
+            null
+        }
 
 
         listener?.onDataReceived(data, participant, this)
         listener?.onDataReceived(data, participant, this)
-        eventBus.postEvent(RoomEvent.DataReceived(this, data, participant), coroutineScope)
-        participant?.onDataReceived(data)
+        eventBus.postEvent(RoomEvent.DataReceived(this, data, participant, topic), coroutineScope)
+        participant?.onDataReceived(data, topic)
     }
     }
 
 
     /**
     /**

@@ -270,6 +270,8 @@ constructor(
         val wasConnected = isConnected
         val wasConnected = isConnected
 
 
         if (wasConnected) {
         if (wasConnected) {
+            // onClosing/onClosed will not be called after onFailure.
+            // Handle websocket closure here.
             handleWebSocketClose(
             handleWebSocketClose(
                 reason = reason ?: response?.toString() ?: t.localizedMessage ?: "websocket failure",
                 reason = reason ?: response?.toString() ?: t.localizedMessage ?: "websocket failure",
                 code = response?.code ?: CLOSE_REASON_WEBSOCKET_FAILURE
                 code = response?.code ?: CLOSE_REASON_WEBSOCKET_FAILURE
@@ -390,7 +392,7 @@ constructor(
                     quality = LivekitModels.VideoQuality.HIGH
                     quality = LivekitModels.VideoQuality.HIGH
                 }
                 }
 
 
-                if(fps != null){
+                if (fps != null) {
                     setFps(fps)
                     setFps(fps)
                 }
                 }
             }
             }

@@ -443,12 +443,14 @@ internal constructor(
      * @param data payload to send
      * @param data payload to send
      * @param reliability for delivery guarantee, use RELIABLE. for fastest delivery without guarantee, use LOSSY
      * @param reliability for delivery guarantee, use RELIABLE. for fastest delivery without guarantee, use LOSSY
      * @param destination list of participant SIDs to deliver the payload, null to deliver to everyone
      * @param destination list of participant SIDs to deliver the payload, null to deliver to everyone
+     * @param topic the topic under which the message was published
      */
      */
     @Suppress("unused")
     @Suppress("unused")
     suspend fun publishData(
     suspend fun publishData(
         data: ByteArray,
         data: ByteArray,
         reliability: DataPublishReliability = DataPublishReliability.RELIABLE,
         reliability: DataPublishReliability = DataPublishReliability.RELIABLE,
-        destination: List<String>? = null
+        destination: List<String>? = null,
+        topic: String? = null,
     ) {
     ) {
         if (data.size > RTCEngine.MAX_DATA_PACKET_SIZE) {
         if (data.size > RTCEngine.MAX_DATA_PACKET_SIZE) {
             throw IllegalArgumentException("cannot publish data larger than " + RTCEngine.MAX_DATA_PACKET_SIZE)
             throw IllegalArgumentException("cannot publish data larger than " + RTCEngine.MAX_DATA_PACKET_SIZE)
@@ -458,11 +460,15 @@ internal constructor(
             DataPublishReliability.RELIABLE -> LivekitModels.DataPacket.Kind.RELIABLE
             DataPublishReliability.RELIABLE -> LivekitModels.DataPacket.Kind.RELIABLE
             DataPublishReliability.LOSSY -> LivekitModels.DataPacket.Kind.LOSSY
             DataPublishReliability.LOSSY -> LivekitModels.DataPacket.Kind.LOSSY
         }
         }
-        val packetBuilder = LivekitModels.UserPacket.newBuilder()
-            .setPayload(ByteString.copyFrom(data))
-            .setParticipantSid(sid)
-        if (destination != null) {
-            packetBuilder.addAllDestinationSids(destination)
+        val packetBuilder = LivekitModels.UserPacket.newBuilder().apply {
+            payload = ByteString.copyFrom(data)
+            participantSid = sid
+            if (topic != null) {
+                setTopic(topic)
+            }
+            if (destination != null) {
+                addAllDestinationSids(destination)
+            }
         }
         }
         val dataPacket = LivekitModels.DataPacket.newBuilder()
         val dataPacket = LivekitModels.DataPacket.newBuilder()
             .setUser(packetBuilder)
             .setUser(packetBuilder)

@@ -179,9 +179,9 @@ class RemoteParticipant(
     }
     }
 
 
     // Internal methods just for posting events.
     // Internal methods just for posting events.
-    internal fun onDataReceived(data: ByteArray) {
+    internal fun onDataReceived(data: ByteArray, topic: String?) {
         listener?.onDataReceived(data, this)
         listener?.onDataReceived(data, this)
-        eventBus.postEvent(ParticipantEvent.DataReceived(this, data), scope)
+        eventBus.postEvent(ParticipantEvent.DataReceived(this, data, topic), scope)
     }
     }
 
 
     companion object {
     companion object {