Return streamInfo from datastream send helper methods (#741)

* Return streamInfo from datastream send helper methods

* changeset

+---
+"client-sdk-android": minor
+---
+
+Return streamInfo from datastream send helper methods

@@ -60,32 +60,36 @@ interface OutgoingDataStreamManager {
      * Send text through a data stream.
      * Send text through a data stream.
      */
      */
     @CheckResult
     @CheckResult
-    suspend fun sendText(text: String, options: StreamTextOptions = StreamTextOptions()): Result<Unit> {
+    suspend fun sendText(text: String, options: StreamTextOptions = StreamTextOptions()): Result<TextStreamInfo> {
         val sender = streamText(options)
         val sender = streamText(options)
         val result = sender.write(text)
         val result = sender.write(text)
 
 
         if (result.isFailure) {
         if (result.isFailure) {
-            sender.close(result.exceptionOrNull()?.message ?: "Unknown error.")
+            val exception = result.exceptionOrNull() ?: Exception("Unknown error.")
+            sender.close(exception.message)
+            return Result.failure(exception)
         } else {
         } else {
             sender.close()
             sender.close()
+            return Result.success(sender.info)
         }
         }
-        return result
     }
     }
 
 
     /**
     /**
      * Send a file through a data stream.
      * Send a file through a data stream.
      */
      */
     @CheckResult
     @CheckResult
-    suspend fun sendFile(file: File, options: StreamBytesOptions = StreamBytesOptions()): Result<Unit> {
+    suspend fun sendFile(file: File, options: StreamBytesOptions = StreamBytesOptions()): Result<ByteStreamInfo> {
         val sender = streamBytes(options)
         val sender = streamBytes(options)
         val result = sender.writeFile(file)
         val result = sender.writeFile(file)
 
 
         if (result.isFailure) {
         if (result.isFailure) {
-            sender.close(result.exceptionOrNull()?.message ?: "Unknown error.")
+            val exception = result.exceptionOrNull() ?: Exception("Unknown error.")
+            sender.close(exception.message)
+            return Result.failure(exception)
         } else {
         } else {
             sender.close()
             sender.close()
+            return Result.success(sender.info)
         }
         }
-        return result
     }
     }
 }
 }
 
 

@@ -108,6 +108,96 @@ class RoomIncomingDataStreamMockE2ETest : MockE2ETest() {
     }
     }
 
 
     @Test
     @Test
+    fun textStreamThroughReadNext() = runTest {
+        connect()
+        val subPeerConnection = component.rtcEngine().getSubscriberPeerConnection() as MockPeerConnection
+        val subDataChannel = MockDataChannel(RTCEngine.RELIABLE_DATA_CHANNEL_LABEL)
+        subPeerConnection.observer?.onDataChannel(subDataChannel)
+
+        val scope = CoroutineScope(currentCoroutineContext())
+        val collectedData = mutableListOf<String>()
+        var finished = false
+        room.registerTextStreamHandler("topic") { reader, _ ->
+            scope.launch {
+                try {
+                    while (true) {
+                        collectedData.add(reader.readNext())
+                    }
+                } catch (exception: NoSuchElementException) {
+                    // end of stream
+                }
+                finished = true
+            }
+        }
+
+        val textStreamHeader = with(createStreamHeader().toBuilder()) {
+            streamHeader = with(streamHeader.toBuilder()) {
+                clearByteHeader()
+                textHeader = with(TextHeader.newBuilder()) {
+                    operationType = OperationType.CREATE
+                    generated = false
+                    build()
+                }
+                build()
+            }
+            build()
+        }
+        subDataChannel.observer?.onMessage(textStreamHeader.wrap())
+        subDataChannel.observer?.onMessage(createStreamChunk(0, "hello".toByteArray()).wrap())
+        subDataChannel.observer?.onMessage(createStreamChunk(1, "world".toByteArray()).wrap())
+        subDataChannel.observer?.onMessage(createStreamChunk(2, "!".toByteArray()).wrap())
+        subDataChannel.observer?.onMessage(createStreamTrailer().wrap())
+
+        assertTrue(finished)
+        assertEquals(3, collectedData.size)
+        assertEquals("hello", collectedData[0])
+        assertEquals("world", collectedData[1])
+        assertEquals("!", collectedData[2])
+    }
+
+    @Test
+    fun textStreamThroughReadAll() = runTest {
+        connect()
+        val subPeerConnection = component.rtcEngine().getSubscriberPeerConnection() as MockPeerConnection
+        val subDataChannel = MockDataChannel(RTCEngine.RELIABLE_DATA_CHANNEL_LABEL)
+        subPeerConnection.observer?.onDataChannel(subDataChannel)
+
+        val scope = CoroutineScope(currentCoroutineContext())
+        val collectedData = mutableListOf<String>()
+        var finished = false
+        room.registerTextStreamHandler("topic") { reader, _ ->
+            scope.launch {
+                collectedData.addAll(reader.readAll())
+                finished = true
+            }
+        }
+
+        val textStreamHeader = with(createStreamHeader().toBuilder()) {
+            streamHeader = with(streamHeader.toBuilder()) {
+                clearByteHeader()
+                textHeader = with(TextHeader.newBuilder()) {
+                    operationType = OperationType.CREATE
+                    generated = false
+                    build()
+                }
+                build()
+            }
+            build()
+        }
+        subDataChannel.observer?.onMessage(textStreamHeader.wrap())
+        subDataChannel.observer?.onMessage(createStreamChunk(0, "hello".toByteArray()).wrap())
+        subDataChannel.observer?.onMessage(createStreamChunk(1, "world".toByteArray()).wrap())
+        subDataChannel.observer?.onMessage(createStreamChunk(2, "!".toByteArray()).wrap())
+        subDataChannel.observer?.onMessage(createStreamTrailer().wrap())
+
+        assertTrue(finished)
+        assertEquals(3, collectedData.size)
+        assertEquals("hello", collectedData[0])
+        assertEquals("world", collectedData[1])
+        assertEquals("!", collectedData[2])
+    }
+
+    @Test
     fun dataStreamTerminated() = runTest {
     fun dataStreamTerminated() = runTest {
         connect()
         connect()
         val subPeerConnection = component.rtcEngine().getSubscriberPeerConnection() as MockPeerConnection
         val subPeerConnection = component.rtcEngine().getSubscriberPeerConnection() as MockPeerConnection