[Functions] Several functions improvements (#6796)

These improvements are both in the SDK itself, and in the code used for
the actual functions running in the cloud during integration testing:

A summary of the changes is:
- [Backend functions] Update the dependencies and node runtime version
- [Backend functions] Add logging in case the client does not support
streaming
- [SDK] Use a more robust handle of the media type
- [SDK] Fix issue introduced in #6773 that missed a return in case of
error
- [SDK] Other minor code fixes

b/404814020

Author: rlazo

firebase-functions/CHANGELOG.md

@@ -1,5 +1,5 @@
 # Unreleased
-
+* [fixed] Fixed issue that caused the SDK to crash when trying to stream a function that does not exist.
 
 # 21.2.0
 * [feature] Streaming callable functions are now supported.
@@ -235,4 +235,3 @@ updates.
 optional region to override the default "us-central1".
 * [feature] New `useFunctionsEmulator` method allows testing against a local
 instance of the [Cloud Functions Emulator](https://firebase.google.com/docs/functions/local-emulator).
-

firebase-functions/src/androidTest/backend/functions/index.js

@@ -154,6 +154,9 @@ exports.genStream = functionsV2.https.onCall(async (request, response) => {
       response.sendChunk(chunk);
     }
   }
+  else {
+      console.log("CLIENT DOES NOT SUPPORT STEAMING");
+  }
   return streamData.join(' ');
 });
 
@@ -225,6 +228,8 @@ exports.genStreamLargeData = functionsV2.https.onCall(
         response.sendChunk(chunk);
         await sleep(100);
       }
+    } else {
+      console.log("CLIENT DOES NOT SUPPORT STEAMING")
     }
     return "Stream Completed";
   }

firebase-functions/src/androidTest/backend/functions/package.json

@@ -2,11 +2,11 @@
   "name": "functions",
   "description": "Cloud Functions for Firebase",
   "dependencies": {
-    "firebase-admin": "11.8.0",
-    "firebase-functions": "4.4.0"
+    "firebase-admin": "13.2.0",
+    "firebase-functions": "6.3.2"
   },
   "private": true,
   "engines": {
-    "node": "18"
+    "node": "22"
   }
 }

firebase-functions/src/androidTest/java/com/google/firebase/functions/StreamTests.kt

@@ -116,11 +116,11 @@ class StreamTests {
       throwable = e
     }
 
+    assertThat(throwable).isNull()
     assertThat(messages.map { it.message.data.toString() })
       .containsExactly("hello", "world", "this", "is", "cool")
     assertThat(result).isNotNull()
     assertThat(result!!.result.data.toString()).isEqualTo("hello world this is cool")
-    assertThat(throwable).isNull()
     assertThat(isComplete).isTrue()
   }
 
@@ -196,6 +196,7 @@ class StreamTests {
     function.stream(mapOf("data" to "test")).subscribe(subscriber)
 
     withTimeout(2000) { delay(500) }
+    assertThat(subscriber.throwable).isNull()
     assertThat(subscriber.messages).isEmpty()
     assertThat(subscriber.result).isNull()
   }

firebase-functions/src/main/java/com/google/firebase/functions/PublisherStream.kt

@@ -136,12 +136,20 @@ internal class PublisherStream(
           MediaType.parse("application/json"),
           JSONObject(mapOf("data" to serializer.encode(data))).toString()
         )
-      val requestBuilder =
-        Request.Builder().url(url).post(requestBody).header("Accept", "text/event-stream")
-      context?.authToken?.let { requestBuilder.header("Authorization", "Bearer $it") }
-      context?.instanceIdToken?.let { requestBuilder.header("Firebase-Instance-ID-Token", it) }
-      context?.appCheckToken?.let { requestBuilder.header("X-Firebase-AppCheck", it) }
-      val request = requestBuilder.build()
+      val request =
+        Request.Builder()
+          .url(url)
+          .post(requestBody)
+          .apply {
+            header("Accept", "text/event-stream")
+            header("Content-Type", "application/json")
+            context?.apply {
+              authToken?.let { header("Authorization", "Bearer $it") }
+              instanceIdToken?.let { header("Firebase-Instance-ID-Token", it) }
+              appCheckToken?.let { header("X-Firebase-AppCheck", it) }
+            }
+          }
+          .build()
       val call = configuredClient.newCall(request)
       activeCall = call
 
@@ -206,6 +214,9 @@ internal class PublisherStream(
             eventBuffer.append(dataChunk.trim()).append("\n")
           }
         }
+        if (eventBuffer.isNotEmpty()) {
+          processEvent(eventBuffer.toString())
+        }
       } catch (e: Exception) {
         notifyError(
           FirebaseFunctionsException(
@@ -296,9 +307,11 @@ internal class PublisherStream(
   private fun validateResponse(response: Response) {
     if (response.isSuccessful) return
 
-    val htmlContentType = "text/html; charset=utf-8"
     val errorMessage: String
-    if (response.code() == 404 && response.header("Content-Type") == htmlContentType) {
+    if (
+      response.code() == 404 &&
+        MediaType.parse(response.header("Content-Type") ?: "")?.subtype() == "html"
+    ) {
       errorMessage = """URL not found. Raw response: ${response.body()?.string()}""".trimMargin()
       notifyError(
         FirebaseFunctionsException(
@@ -307,6 +320,7 @@ internal class PublisherStream(
           null
         )
       )
+      return
     }
 
     val text = response.body()?.string() ?: ""