fix: expo/fetch not closing streams

Vali-98 · Dec 5, 2024 · ab6fbde · ab6fbde
1 parent bdcb08f
commit ab6fbde
Show file tree

Hide file tree

Showing 3 changed files with 107 additions and 4 deletions.
diff --git a/constants/API/APIBuilder.ts b/constants/API/APIBuilder.ts
@@ -96,7 +96,6 @@ const readableStreamResponse = async (
 
     useInference.getState().setAbort(async () => {
         Logger.debug('Running abort')
-        closeStream()
         sse.abort()
     })
 

diff --git a/constants/SSEFetch.ts b/constants/SSEFetch.ts
@@ -1,3 +1,4 @@
+import { Logger } from '@globals'
 import { fetch } from 'expo/fetch'
 
 type SSEValues = {
@@ -25,21 +26,27 @@ export class SSEFetch {
         this.abortController = new AbortController()
         const body = values.method === 'POST' ? { body: values.body } : {}
 
-        fetch(values.endpoint, {
+        const res = await fetch(values.endpoint, {
             signal: this.abortController.signal,
             method: values.method,
             headers: values.headers,
             ...body,
-        }).then(async (res) => {
+        })
+        try {
             if (res.status !== 200 || !res.body) return this.onError()
             this.closeStream = res.body.cancel
             for await (const chunk of res.body) {
                 const data = this.decoder.decode(chunk)
                 const output = parseSSE(data)
                 output.forEach((item) => this.onEvent(item))
             }
+        } catch (e) {
+            if (this.abortController.signal.aborted) {
+                Logger.debug('Abort caught')
+            }
+        } finally {
             this.onClose()
-        })
+        }
     }
 
     public setOnEvent(callback: (data: string) => void) {

diff --git a/patches/expo+52.0.14.patch b/patches/expo+52.0.14.patch
@@ -0,0 +1,97 @@
+diff --git a/node_modules/expo/android/src/main/java/expo/modules/fetch/NativeResponse.kt b/node_modules/expo/android/src/main/java/expo/modules/fetch/NativeResponse.kt
+index 8254d15..64554a4 100644
+--- a/node_modules/expo/android/src/main/java/expo/modules/fetch/NativeResponse.kt
++++ b/node_modules/expo/android/src/main/java/expo/modules/fetch/NativeResponse.kt
+@@ -68,6 +68,7 @@ internal class NativeResponse(appContext: AppContext, private val coroutineScope
+     if (isInvalidState(ResponseState.BODY_STREAMING_STARTED)) {
+       return
+     }
++    
+     state = ResponseState.BODY_STREAMING_CANCELLED
+   }
+
+@@ -118,7 +119,6 @@ internal class NativeResponse(appContext: AppContext, private val coroutineScope
+       val stream = response.body?.source() ?: return@launch
+       pumpResponseBodyStream(stream)
+       response.close()
+-
+       if ([email protected] == ResponseState.BODY_STREAMING_STARTED) {
+         emit("didComplete")
+       }
+@@ -157,7 +157,7 @@ internal class NativeResponse(appContext: AppContext, private val coroutineScope
+     )
+   }
+
+-  private fun pumpResponseBodyStream(stream: BufferedSource) {
++  /*private fun pumpResponseBodyStream(stream: BufferedSource) {
+     while (!stream.exhausted()) {
+       if (isInvalidState(
+           ResponseState.RESPONSE_RECEIVED,
+@@ -165,6 +165,7 @@ internal class NativeResponse(appContext: AppContext, private val coroutineScope
+           ResponseState.BODY_STREAMING_CANCELLED
+         )
+       ) {
++        Log.w(TAG, "State is Invalid")
+         break
+       }
+       if (state == ResponseState.RESPONSE_RECEIVED) {
+@@ -175,7 +176,45 @@ internal class NativeResponse(appContext: AppContext, private val coroutineScope
+         break
+       }
+     }
+-  }
++  }*/
++
++  private fun pumpResponseBodyStream(stream: BufferedSource) {
++    try {
++        while (!stream.exhausted()) {
++            if (isInvalidState(
++                    ResponseState.RESPONSE_RECEIVED,
++                    ResponseState.BODY_STREAMING_STARTED,
++                    ResponseState.BODY_STREAMING_CANCELLED
++                )
++            ) {
++                break
++            }
++            if (state == ResponseState.RESPONSE_RECEIVED) {
++                sink.appendBufferBody(stream.buffer.readByteArray())
++            } else if (state == ResponseState.BODY_STREAMING_STARTED) {
++                emit("didReceiveResponseData", stream.buffer.readByteArray())
++            } else {
++                break
++            }
++        }
++    } catch (e: IOException) {
++        Log.e(TAG, "Error while pumping response body stream: ${e.message}", e)
++        state = ResponseState.ERROR_RECEIVED
++        error = e
++        emit("didFailWithError", e)
++    } catch (e: Exception) {
++        Log.e(TAG, "Unexpected error while pumping response body stream: ${e.message}", e)
++        state = ResponseState.ERROR_RECEIVED
++        error = e
++        emit("didFailWithError", e)
++    } finally {
++        try {
++            stream.close()
++        } catch (e: IOException) {
++            Log.w(TAG, "Error while closing the stream: ${e.message}", e)
++        }
++    }
++}
+
+   //endregion Internals
+
+diff --git a/node_modules/expo/src/winter/fetch/fetch.ts b/node_modules/expo/src/winter/fetch/fetch.ts
+index e7a515b..0a0f8ea 100644
+--- a/node_modules/expo/src/winter/fetch/fetch.ts
++++ b/node_modules/expo/src/winter/fetch/fetch.ts
+@@ -38,7 +38,7 @@ export async function fetch(url: string, init?: FetchRequestInit): Promise<Fetch
+       throw new FetchError(String(e));
+     }
+   } finally {
+-    init?.signal?.removeEventListener('abort', abortHandler);
++    // init?.signal?.removeEventListener('abort', abortHandler);
+   }
+   return response;
+ }