feat: fetch android with streams (#49)

* init

* tmp: works

* update example

* clean-up

* clean-up

* tmp

* temporary fix on Android

Author: grabbou

example/tests/benchmark.tsx

@@ -76,7 +76,7 @@ const TESTS = [
   },
 ]
 
-const BASE_URL = Platform.OS === 'android' ? 'ws://10.0.2.2' : 'ws://localhost'
+export const BASE_URL = Platform.OS === 'android' ? '10.0.2.2' : 'localhost'
 
 export function BenchmarkUI() {
   return (
@@ -278,7 +278,7 @@ async function runSingleTest(opts: TestCase): Promise<TestResult> {
   const { Ws, port, messageCount, testCase, payload } = opts
 
   return new Promise((resolve) => {
-    const ws = new Ws(`${BASE_URL}:${port}`)
+    const ws = new Ws(`ws://${BASE_URL}:${port}`)
     let outgoingTime: number
     let incomingTime: number
     let received = 0

example/tests/filesystem.tsx

@@ -7,6 +7,8 @@ import {
   showOpenFilePicker,
 } from 'react-native-fast-io'
 
+import { BASE_URL } from './benchmark'
+
 export function FileSystemUI() {
   const [file, setFile] = useState<File | null>(null)
 
@@ -21,10 +23,9 @@ export function FileSystemUI() {
     if (!file) {
       return
     }
-
     const body = compression ? file.stream().pipeThrough(new CompressionStream(compression)) : file
 
-    await fetch('http://localhost:3002/upload', {
+    await fetch(`http://${BASE_URL}:3002/upload`, {
       method: 'POST',
       body,
     })

packages/react-native-fast-io/android/src/main/java/com/margelo/nitro/fastio/HybridDuplexStream.kt

@@ -1,14 +1,29 @@
 package com.margelo.nitro.fastio
 
+import java.io.PipedInputStream
+import java.io.PipedOutputStream
+
 class HybridDuplexStream : HybridDuplexStreamSpec() {
-    override var inputStream: HybridInputStreamSpec
-        get() = throw NotImplementedError("HybridDuplexStream.inputStream getter not implemented")
-        set(_) = throw NotImplementedError("HybridDuplexStream.inputStream setter not implemented")
+    private val pipeIn = PipedInputStream(HybridStreamFactory.BUFFER_SIZE)
+    private val pipeOut = PipedOutputStream(pipeIn)
+
+    override var inputStream: HybridInputStreamSpec = HybridInputStream(pipeIn).also {
+        System.err.println("Created input stream wrapper")
+    }
+    
+    override var outputStream: HybridOutputStreamSpec = HybridOutputStream(pipeOut).also {
+        System.err.println("Created output stream wrapper")
+    }
 
-    override var outputStream: HybridOutputStreamSpec
-        get() = throw NotImplementedError("HybridDuplexStream.outputStream getter not implemented")
-        set(_) = throw NotImplementedError("HybridDuplexStream.outputStream setter not implemented")
+    override val memorySize: Long = 0L
 
-    override val memorySize: Long
-        get() = 0L
+    fun close() {
+        try {
+            System.err.println("Closing duplex stream")
+            pipeOut.close()
+            pipeIn.close()
+        } catch (e: Exception) {
+            System.err.println("Error closing duplex stream: ${e.message}")
+        }
+    }
 } 

packages/react-native-fast-io/android/src/main/java/com/margelo/nitro/fastio/HybridInputStream.kt

@@ -2,29 +2,28 @@ package com.margelo.nitro.fastio
 
 import com.margelo.nitro.core.ArrayBuffer
 import com.margelo.nitro.core.Promise
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import java.io.InputStream
 
-class HybridInputStream(public val stream: InputStream) : HybridInputStreamSpec() {
+class HybridInputStream(val stream: InputStream) : HybridInputStreamSpec() {
     override fun read(): Promise<ArrayBuffer> {
-        return Promise.async {
+        return Promise.async(CoroutineScope(Dispatchers.IO)) {
             val bytes = ByteArray(HybridStreamFactory.BUFFER_SIZE)
             val bytesRead = stream.read(bytes, 0, bytes.size)
 
             when {
                 bytesRead == -1 -> {
-                    // End of stream
-                    ArrayBuffer.allocate(0)
+                    val emptyBuffer = ArrayBuffer.allocate(0)
+                    emptyBuffer
                 }
                 bytesRead > 0 -> {
                     val arrayBuffer = ArrayBuffer.allocate(bytesRead)
-
                     val destBuffer = arrayBuffer.getBuffer(false)
                     destBuffer.put(bytes, 0, bytesRead)
-
                     arrayBuffer
                 }
                 else -> {
-                    // Error case
                     throw Error("Unexpected error reading stream")
                 }
             }

packages/react-native-fast-io/android/src/main/java/com/margelo/nitro/fastio/HybridNetwork.kt

@@ -1,10 +1,34 @@
 package com.margelo.nitro.fastio
 
 import com.margelo.nitro.core.Promise
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import java.net.HttpURLConnection
+import java.net.URL
 
 class HybridNetwork : HybridNetworkSpec() {
     override fun request(opts: RequestOptions): Promise<Unit> {
-        throw NotImplementedError("HybridNetwork.request() not implemented")
+        Promise.async(CoroutineScope(Dispatchers.IO)) {
+            val connection = URL(opts.url).openConnection() as HttpURLConnection
+            connection.apply {
+                requestMethod = opts.method.name.uppercase()
+                doInput = true
+                doOutput = opts.body != null
+
+                opts.body?.let { hybridStream ->
+                    (hybridStream as HybridInputStream).stream.use { input ->
+                        outputStream.buffered().use { output ->
+                            input.copyTo(output, HybridStreamFactory.BUFFER_SIZE)
+                        }
+                    }
+                }
+
+                if (responseCode !in 200..299) {
+                    throw Error("HTTP Error: $responseCode")
+                }
+            }
+        }
+        return Promise.resolved(Unit)
     }
 
     override val memorySize: Long

packages/react-native-fast-io/android/src/main/java/com/margelo/nitro/fastio/HybridOutputStream.kt

@@ -5,7 +5,6 @@ import com.margelo.nitro.core.Promise
 import java.io.OutputStream
 
 class HybridOutputStream(private val stream: OutputStream) : HybridOutputStreamSpec() {
-    
     override fun write(buffer: ArrayBuffer): Promise<Unit> {
         val byteBuffer = buffer.getBuffer(false)
         val bytes = ByteArray(buffer.size)

packages/react-native-fast-io/src/w3c/streams.ts

@@ -11,7 +11,9 @@ export const toReadableStream = (inputStream: InputStream) => {
         controller.close()
         return
       }
-      controller.enqueue(new Uint8Array(buffer))
+
+      const chunk = new Uint8Array(buffer)
+      controller.enqueue(chunk)
     },
     cancel() {
       inputStream.close()