Fix Subscriber primary data channel (#33)

* Properly receive data channels through subscriber

* clean up log

* compose sample app data sending and receiving

* sample app send message example

 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
   <component name="CompilerConfiguration">
-    <bytecodeTargetLevel target="1.8">
-      <module name="livekit-android.livekit-android-sdk" target="11" />
-      <module name="livekit-android.sample-app" target="11" />
-      <module name="livekit-android.sample-app-common" target="11" />
-      <module name="livekit-android.sample-app-compose" target="11" />
-    </bytecodeTargetLevel>
+    <bytecodeTargetLevel target="1.8" />
   </component>
 </project>
\ No newline at end of file

@@ -12,7 +12,6 @@ class PublisherTransportObserver(
     private val client: SignalClient,
 ) : PeerConnection.Observer, PeerConnectionTransport.Listener {
 
-    var dataChannelListener: ((DataChannel?) -> Unit)? = null
     var iceConnectionChangeListener: ((newState: PeerConnection.IceConnectionState?) -> Unit)? =
         null
 
@@ -63,7 +62,6 @@ class PublisherTransportObserver(
     }
 
     override fun onDataChannel(dataChannel: DataChannel?) {
-        dataChannelListener?.invoke(dataChannel)
     }
 
     override fun onTrack(transceiver: RtpTransceiver?) {

@@ -179,10 +179,7 @@ internal constructor(
 
         if (joinResponse.subscriberPrimary) {
             // in subscriber primary mode, server side opens sub data channels.
-            publisherObserver.dataChannelListener = onDataChannel@{ dataChannel: DataChannel? ->
-                if (dataChannel == null) {
-                    return@onDataChannel
-                }
+            subscriberObserver.dataChannelListener = onDataChannel@{ dataChannel: DataChannel ->
                 when (dataChannel.label()) {
                     RELIABLE_DATA_CHANNEL_LABEL -> reliableDataChannelSub = dataChannel
                     LOSSY_DATA_CHANNEL_LABEL -> lossyDataChannelSub = dataChannel
@@ -190,9 +187,9 @@ internal constructor(
                 }
                 dataChannel.registerObserver(this)
             }
-            publisherObserver.iceConnectionChangeListener = iceConnectionStateListener
-        } else {
             subscriberObserver.iceConnectionChangeListener = iceConnectionStateListener
+        } else {
+            publisherObserver.iceConnectionChangeListener = iceConnectionStateListener
         }
 
         // data channels
@@ -282,7 +279,7 @@ internal constructor(
             if (hasPublished) {
                 publisher.negotiate(
                     getPublisherOfferConstraints().apply {
-                        with(mandatory){
+                        with(mandatory) {
                             add(
                                 MediaConstraints.KeyValuePair(
                                     MediaConstraintKeys.ICE_RESTART,
@@ -323,7 +320,7 @@ internal constructor(
         channel.send(buf)
     }
 
-    private suspend fun ensurePublisherConnected(){
+    private suspend fun ensurePublisherConnected() {
         if (!isSubscriberPrimary) {
             return
         }
@@ -557,7 +554,7 @@ internal constructor(
     }
 }
 
- internal enum class IceState {
+internal enum class IceState {
     DISCONNECTED,
     RECONNECTING,
     CONNECTED,

@@ -12,6 +12,7 @@ class SubscriberTransportObserver(
     private val client: SignalClient,
 ) : PeerConnection.Observer {
 
+    var dataChannelListener: ((DataChannel) -> Unit)? = null
     var iceConnectionChangeListener: ((PeerConnection.IceConnectionState?) -> Unit)? = null
 
     override fun onIceCandidate(candidate: IceCandidate) {
@@ -34,7 +35,7 @@ class SubscriberTransportObserver(
     }
 
     override fun onDataChannel(channel: DataChannel) {
-        LKLog.v { "onDataChannel" }
+        dataChannelListener?.invoke(channel)
     }
 
     override fun onStandardizedIceConnectionChange(newState: PeerConnection.IceConnectionState?) {

@@ -311,7 +311,7 @@ internal constructor(
             }
 
             fun addEncoding(videoEncoding: VideoEncoding, scale: Double) {
-                if(encodings.size >= VIDEO_RIDS.size) {
+                if (encodings.size >= VIDEO_RIDS.size) {
                     throw IllegalStateException("Attempting to add more encodings than we have rids for!")
                 }
                 val rid = VIDEO_RIDS[encodings.size]
@@ -348,7 +348,7 @@ internal constructor(
         // presets assume width is longest size
         val longestSize = max(width, height)
         val preset = presets
-            .firstOrNull { it.capture.width >= longestSize}
+            .firstOrNull { it.capture.width >= longestSize }
             ?: presets.last()
 
         return preset.encoding
@@ -435,7 +435,11 @@ internal constructor(
      * @param destination list of participant SIDs to deliver the payload, null to deliver to everyone
      */
     @Suppress("unused")
-    suspend fun publishData(data: ByteArray, reliability: DataPublishReliability, destination: List<String>?) {
+    suspend fun publishData(
+        data: ByteArray,
+        reliability: DataPublishReliability = DataPublishReliability.RELIABLE,
+        destination: List<String>? = null
+    ) {
         if (data.size > RTCEngine.MAX_DATA_PACKET_SIZE) {
             throw IllegalArgumentException("cannot publish data larger than " + RTCEngine.MAX_DATA_PACKET_SIZE)
         }
@@ -444,16 +448,16 @@ internal constructor(
             DataPublishReliability.RELIABLE -> LivekitModels.DataPacket.Kind.RELIABLE
             DataPublishReliability.LOSSY -> LivekitModels.DataPacket.Kind.LOSSY
         }
-        val packetBuilder = LivekitModels.UserPacket.newBuilder().
-                setPayload(ByteString.copyFrom(data)).
-                setParticipantSid(sid)
+        val packetBuilder = LivekitModels.UserPacket.newBuilder()
+            .setPayload(ByteString.copyFrom(data))
+            .setParticipantSid(sid)
         if (destination != null) {
             packetBuilder.addAllDestinationSids(destination)
         }
-        val dataPacket = LivekitModels.DataPacket.newBuilder().
-            setUser(packetBuilder).
-            setKind(kind).
-            build()
+        val dataPacket = LivekitModels.DataPacket.newBuilder()
+            .setUser(packetBuilder)
+            .setKind(kind)
+            .build()
 
         engine.sendData(dataPacket)
     }

@@ -17,6 +17,8 @@ import io.livekit.android.util.flow
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.launch
+import okio.Utf8
+import java.nio.charset.Charset
 
 @OptIn(ExperimentalCoroutinesApi::class)
 class CallViewModel(
@@ -69,6 +71,9 @@ class CallViewModel(
     private val mutableScreencastEnabled = MutableLiveData(false)
     val screenshareEnabled = mutableScreencastEnabled.hide()
 
+    private val mutableDataReceived = MutableSharedFlow<String>()
+    val dataReceived = mutableDataReceived
+
     init {
         viewModelScope.launch {
             try {
@@ -95,6 +100,11 @@ class CallViewModel(
                     room.events.collect {
                         when (it) {
                             is RoomEvent.FailedToConnect -> mutableError.value = it.error
+                            is RoomEvent.DataReceived -> {
+                                val identity = it.participant.identity ?: ""
+                                val message = it.data.toString(Charsets.UTF_8)
+                                mutableDataReceived.emit("$identity: $message")
+                            }
                         }
                     }
                 }
@@ -186,8 +196,15 @@ class CallViewModel(
     fun dismissError() {
         mutableError.value = null
     }
+
+    fun sendData(message: String) {
+        viewModelScope.launch {
+            room.value?.localParticipant?.publishData(message.toByteArray(Charsets.UTF_8))
+        }
+    }
 }
 
 private fun <T> LiveData<T>.hide(): LiveData<T> = this
 
-private fun <T> MutableStateFlow<T>.hide(): StateFlow<T> = this
\ No newline at end of file
+private fun <T> MutableStateFlow<T>.hide(): StateFlow<T> = this
+private fun <T> Flow<T>.hide(): Flow<T> = this
\ No newline at end of file

+<vector xmlns:android="http://schemas.android.com/apk/res/android"
+    android:width="24dp"
+    android:height="24dp"
+    android:viewportWidth="24"
+    android:viewportHeight="24"
+    android:tint="?attr/colorControlNormal"
+    android:autoMirrored="true">
+  <path
+      android:fillColor="@android:color/white"
+      android:pathData="M20,2L4,2c-1.1,0 -1.99,0.9 -1.99,2L2,22l4,-4h14c1.1,0 2,-0.9 2,-2L22,4c0,-1.1 -0.9,-2 -2,-2zM6,9h12v2L6,11L6,9zM14,14L6,14v-2h8v2zM18,8L6,8L6,6h12v2z"/>
+</vector>

@@ -86,15 +86,6 @@ class CallActivity : AppCompatActivity() {
             Timber.v { "Audio focus request failed" }
         }
 
-        lifecycleScope.launchWhenStarted {
-            viewModel.error.collect {
-                if(it != null){
-                    Toast.makeText(this@CallActivity, "Error: $it", Toast.LENGTH_LONG).show()
-                    viewModel.dismissError()
-                }
-            }
-        }
-
         // Setup compose view.
         setContent {
             val room by viewModel.room.collectAsState()
@@ -114,11 +105,30 @@ class CallActivity : AppCompatActivity() {
                 videoEnabled,
                 flipButtonEnabled,
                 screencastEnabled,
-                onExitClick = { finish() }
+                onExitClick = { finish() },
+                onSendMessage = { viewModel.sendData(it) }
             )
         }
     }
 
+    override fun onResume() {
+        super.onResume()
+        lifecycleScope.launchWhenResumed {
+            viewModel.error.collect {
+                if (it != null) {
+                    Toast.makeText(this@CallActivity, "Error: $it", Toast.LENGTH_LONG).show()
+                    viewModel.dismissError()
+                }
+            }
+        }
+
+        lifecycleScope.launchWhenResumed {
+            viewModel.dataReceived.collect {
+                Toast.makeText(this@CallActivity, "Data received: $it", Toast.LENGTH_LONG).show()
+            }
+        }
+    }
+
     private fun requestMediaProjection() {
         val mediaProjectionManager =
             getSystemService(MEDIA_PROJECTION_SERVICE) as MediaProjectionManager
@@ -141,7 +151,8 @@ class CallActivity : AppCompatActivity() {
         screencastEnabled: Boolean = false,
         onExitClick: () -> Unit = {},
         error: Throwable? = null,
-        onSnackbarDismiss: () -> Unit = {}
+        onSnackbarDismiss: () -> Unit = {},
+        onSendMessage: (String) -> Unit = {},
     ) {
         AppTheme(darkTheme = true) {
             ConstraintLayout(
@@ -216,7 +227,8 @@ class CallActivity : AppCompatActivity() {
                     Surface(
                         onClick = { viewModel.setMicEnabled(!micEnabled) },
                         indication = rememberRipple(false),
-                        modifier = Modifier.size(controlSize)
+                        modifier = Modifier
+                            .size(controlSize)
                             .padding(controlPadding)
                     ) {
                         val resource =
@@ -230,7 +242,8 @@ class CallActivity : AppCompatActivity() {
                     Surface(
                         onClick = { viewModel.setCameraEnabled(!videoEnabled) },
                         indication = rememberRipple(false),
-                        modifier = Modifier.size(controlSize)
+                        modifier = Modifier
+                            .size(controlSize)
                             .padding(controlPadding)
                     ) {
                         val resource =
@@ -244,7 +257,8 @@ class CallActivity : AppCompatActivity() {
                     Surface(
                         onClick = { viewModel.flipCamera() },
                         indication = rememberRipple(false),
-                        modifier = Modifier.size(controlSize)
+                        modifier = Modifier
+                            .size(controlSize)
                             .padding(controlPadding)
                     ) {
                         Icon(
@@ -262,7 +276,8 @@ class CallActivity : AppCompatActivity() {
                             }
                         },
                         indication = rememberRipple(false),
-                        modifier = Modifier.size(controlSize)
+                        modifier = Modifier
+                            .size(controlSize)
                             .padding(controlPadding)
                     ) {
                         val resource =
@@ -273,10 +288,65 @@ class CallActivity : AppCompatActivity() {
                             tint = Color.White,
                         )
                     }
+
+                    var showMessageDialog by remember { mutableStateOf(false) }
+                    var messageToSend by remember { mutableStateOf("") }
+                    Surface(
+                        onClick = { showMessageDialog = true },
+                        indication = rememberRipple(false),
+                        modifier = Modifier
+                            .size(controlSize)
+                            .padding(controlPadding)
+                    ) {
+                        Icon(
+                            painterResource(id = R.drawable.baseline_chat_24),
+                            contentDescription = "Send Message",
+                            tint = Color.White,
+                        )
+                    }
+
+                    if (showMessageDialog) {
+                        AlertDialog(
+                            onDismissRequest = {
+                                showMessageDialog = false
+                                messageToSend = ""
+                            },
+                            title = {
+                                Text(text = "Send Message")
+                            },
+                            text = {
+                                OutlinedTextField(
+                                    value = messageToSend,
+                                    onValueChange = { messageToSend = it },
+                                    label = { Text("Message") },
+                                    modifier = Modifier.fillMaxWidth(),
+                                )
+                            },
+                            confirmButton = {
+                                Button(
+                                    onClick = {
+                                        onSendMessage(messageToSend)
+                                        showMessageDialog = false
+                                        messageToSend = ""
+                                    }
+                                ) { Text("Send") }
+                            },
+                            dismissButton = {
+                                Button(
+                                    onClick = {
+                                        showMessageDialog = false
+                                        messageToSend = ""
+                                    }
+                                ) { Text("Cancel") }
+                            },
+                            backgroundColor = Color.Black,
+                        )
+                    }
                     Surface(
                         onClick = { onExitClick() },
                         indication = rememberRipple(false),
-                        modifier = Modifier.size(controlSize)
+                        modifier = Modifier
+                            .size(controlSize)
                             .padding(controlPadding)
                     ) {
                         Icon(
@@ -290,7 +360,7 @@ class CallActivity : AppCompatActivity() {
                 // Snack bar for errors
                 val scaffoldState = rememberScaffoldState()
                 val scope = rememberCoroutineScope()
-                if(error != null) {
+                if (error != null) {
                     Scaffold(
                         scaffoldState = scaffoldState,
                         floatingActionButton = {
@@ -307,7 +377,10 @@ class CallActivity : AppCompatActivity() {
                         content = { innerPadding ->
                             Text(
                                 text = "Body content",
-                                modifier = Modifier.padding(innerPadding).fillMaxSize().wrapContentSize()
+                                modifier = Modifier
+                                    .padding(innerPadding)
+                                    .fillMaxSize()
+                                    .wrapContentSize()
                             )
                         }
                     )

 package io.livekit.android.sample
 
 import android.app.Activity
+import android.content.DialogInterface
 import android.media.AudioManager
 import android.media.projection.MediaProjectionManager
 import android.os.Bundle
 import android.os.Parcelable
 import android.view.View
+import android.widget.EditText
 import android.widget.Toast
 import androidx.activity.result.contract.ActivityResultContracts
+import androidx.appcompat.app.AlertDialog
 import androidx.appcompat.app.AppCompatActivity
 import androidx.lifecycle.lifecycleScope
 import androidx.recyclerview.widget.LinearLayoutManager
@@ -72,15 +75,6 @@ class CallActivity : AppCompatActivity() {
                 }
         }
 
-        lifecycleScope.launchWhenStarted {
-            viewModel.error.collect {
-                if(it != null){
-                    Toast.makeText(this@CallActivity, "Error: $it", Toast.LENGTH_LONG).show()
-                    viewModel.dismissError()
-                }
-            }
-        }
-
         // speaker view setup
         lifecycleScope.launchWhenCreated {
             viewModel.room.filterNotNull().take(1)
@@ -161,6 +155,19 @@ class CallActivity : AppCompatActivity() {
             )
         }
 
+        binding.message.setOnClickListener {
+            val editText = EditText(this)
+            AlertDialog.Builder(this)
+                .setTitle("Send Message")
+                .setView(editText)
+                .setPositiveButton("Send") { dialog, _ ->
+                    viewModel.sendData(editText.text?.toString() ?: "")
+                }
+                .setNegativeButton("Cancel") { _, _ -> }
+                .create()
+                .show()
+        }
+
         binding.exit.setOnClickListener { finish() }
 
         // Grab audio focus for video call
@@ -184,6 +191,24 @@ class CallActivity : AppCompatActivity() {
         }
     }
 
+    override fun onResume() {
+        super.onResume()
+        lifecycleScope.launchWhenResumed {
+            viewModel.error.collect {
+                if (it != null) {
+                    Toast.makeText(this@CallActivity, "Error: $it", Toast.LENGTH_LONG).show()
+                    viewModel.dismissError()
+                }
+            }
+        }
+
+        lifecycleScope.launchWhenResumed {
+            viewModel.dataReceived.collect {
+                Toast.makeText(this@CallActivity, "Data received: $it", Toast.LENGTH_LONG).show()
+            }
+        }
+    }
+
     private fun requestMediaProjection() {
         val mediaProjectionManager =
             getSystemService(MEDIA_PROJECTION_SERVICE) as MediaProjectionManager

@@ -10,9 +10,9 @@
         android:layout_height="0dp"
         android:background="@color/no_video_background"
         app:layout_constraintBottom_toTopOf="@id/audience_row"
-        app:layout_constraintTop_toTopOf="parent"
         app:layout_constraintEnd_toEndOf="parent"
-        app:layout_constraintStart_toStartOf="parent">
+        app:layout_constraintStart_toStartOf="parent"
+        app:layout_constraintTop_toTopOf="parent">
 
         <ImageView
             android:layout_width="120dp"
@@ -105,6 +105,14 @@
             android:src="@drawable/baseline_cast_24" />
 
         <ImageView
+            android:id="@+id/message"
+            android:layout_width="@dimen/control_size"
+            android:layout_height="@dimen/control_size"
+            android:background="?android:attr/selectableItemBackground"
+            android:padding="@dimen/control_padding"
+            android:src="@drawable/baseline_chat_24" />
+
+        <ImageView
             android:id="@+id/exit"
             android:layout_width="@dimen/control_size"
             android:layout_height="@dimen/control_size"