Extend Firebase SDK with new APIs to consume streaming callable function response (#6602)

Extend Firebase SDK with new APIs to consume streaming callable function
response.

- Handling the server-sent event (SSE) parsing internally
- Providing proper error handling and connection management
- Maintaining memory efficiency for long-running streams

---------

Co-authored-by: Rodrigo Lazo <[email protected]>

Author: MustafaJadid2025

Diff for: firebase-functions/api.txt

@@ -84,6 +84,8 @@ package com.google.firebase.functions {
     method public com.google.android.gms.tasks.Task<com.google.firebase.functions.HttpsCallableResult> call(Object? data);
     method public long getTimeout();
     method public void setTimeout(long timeout, java.util.concurrent.TimeUnit units);
+    method public org.reactivestreams.Publisher<com.google.firebase.functions.StreamResponse> stream();
+    method public org.reactivestreams.Publisher<com.google.firebase.functions.StreamResponse> stream(Object? data = null);
     method public com.google.firebase.functions.HttpsCallableReference withTimeout(long timeout, java.util.concurrent.TimeUnit units);
     property public final long timeout;
   }
@@ -93,6 +95,21 @@ package com.google.firebase.functions {
     field public final Object? data;
   }
 
+  public abstract class StreamResponse {
+  }
+
+  public static final class StreamResponse.Message extends com.google.firebase.functions.StreamResponse {
+    ctor public StreamResponse.Message(com.google.firebase.functions.HttpsCallableResult message);
+    method public com.google.firebase.functions.HttpsCallableResult getMessage();
+    property public final com.google.firebase.functions.HttpsCallableResult message;
+  }
+
+  public static final class StreamResponse.Result extends com.google.firebase.functions.StreamResponse {
+    ctor public StreamResponse.Result(com.google.firebase.functions.HttpsCallableResult result);
+    method public com.google.firebase.functions.HttpsCallableResult getResult();
+    property public final com.google.firebase.functions.HttpsCallableResult result;
+  }
+
 }
 
 package com.google.firebase.functions.ktx {

Diff for: firebase-functions/firebase-functions.gradle.kts

@@ -112,6 +112,8 @@ dependencies {
   implementation(libs.okhttp)
   implementation(libs.playservices.base)
   implementation(libs.playservices.basement)
+  implementation(libs.reactive.streams)
+
   api(libs.playservices.tasks)
 
   kapt(libs.autovalue)
@@ -131,6 +133,7 @@ dependencies {
   androidTestImplementation(libs.truth)
   androidTestImplementation(libs.androidx.test.runner)
   androidTestImplementation(libs.androidx.test.junit)
+  androidTestImplementation(libs.kotlinx.coroutines.reactive)
   androidTestImplementation(libs.mockito.core)
   androidTestImplementation(libs.mockito.dexmaker)
   kapt("com.google.dagger:dagger-android-processor:2.43.2")

Diff for: firebase-functions/src/androidTest/backend/functions/index.js

@@ -14,6 +14,16 @@
 
 const assert = require('assert');
 const functions = require('firebase-functions');
+const functionsV2 = require('firebase-functions/v2');
+
+/**
+ * Pauses the execution for a specified amount of time.
+ * @param {number} ms - The number of milliseconds to sleep.
+ * @return {Promise<void>} A promise that resolves after the specified time.
+ */
+function sleep(ms) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
 
 exports.dataTest = functions.https.onRequest((request, response) => {
   assert.deepEqual(request.body, {
@@ -122,3 +132,100 @@ exports.timeoutTest = functions.https.onRequest((request, response) => {
   // Wait for longer than 500ms.
   setTimeout(() => response.send({data: true}), 500);
 });
+
+const streamData = ['hello', 'world', 'this', 'is', 'cool'];
+
+/**
+ * Generates chunks of text asynchronously, yielding one chunk at a time.
+ * @async
+ * @generator
+ * @yields {string} A chunk of text from the data array.
+ */
+async function* generateText() {
+  for (const chunk of streamData) {
+    yield chunk;
+    await sleep(100);
+  }
+}
+
+exports.genStream = functionsV2.https.onCall(async (request, response) => {
+  if (request.acceptsStreaming) {
+    for await (const chunk of generateText()) {
+      response.sendChunk(chunk);
+    }
+  }
+  return streamData.join(' ');
+});
+
+exports.genStreamError = functionsV2.https.onCall(
+    async (request, response) => {
+      // Note: The functions backend does not pass the error message to the
+      // client at this time.
+      throw Error("BOOM")
+    });
+
+const weatherForecasts = {
+  Toronto: { conditions: 'snowy', temperature: 25 },
+  London: { conditions: 'rainy', temperature: 50 },
+  Dubai: { conditions: 'sunny', temperature: 75 }
+};
+
+/**
+ * Generates weather forecasts asynchronously for the given locations.
+ * @async
+ * @generator
+ * @param {Array<{name: string}>} locations - An array of location objects.
+ */
+async function* generateForecast(locations) {
+  for (const location of locations) {
+    yield { 'location': location,  ...weatherForecasts[location.name] };
+    await sleep(100);
+  }
+};
+
+exports.genStreamWeather = functionsV2.https.onCall(
+    async (request, response) => {
+      const locations = request.data && request.data.data?
+      request.data.data: [];
+      const forecasts = [];
+      if (request.acceptsStreaming) {
+        for await (const chunk of generateForecast(locations)) {
+          forecasts.push(chunk);
+          response.sendChunk(chunk);
+        }
+      }
+      return {forecasts};
+    });
+
+exports.genStreamEmpty = functionsV2.https.onCall(
+  async (request, response) => {
+    if (request.acceptsStreaming) {
+      // Send no chunks
+    }
+    // Implicitly return null.
+  }
+);
+
+exports.genStreamResultOnly = functionsV2.https.onCall(
+  async (request, response) => {
+    if (request.acceptsStreaming) {
+      // Do not send any chunks.
+    }
+    return "Only a result";
+  }
+);
+
+exports.genStreamLargeData = functionsV2.https.onCall(
+  async (request, response) => {
+    if (request.acceptsStreaming) {
+      const largeString = 'A'.repeat(10000);
+      const chunkSize = 1024;
+      for (let i = 0; i < largeString.length; i += chunkSize) {
+        const chunk = largeString.substring(i, i + chunkSize);
+        response.sendChunk(chunk);
+        await sleep(100);
+      }
+    }
+    return "Stream Completed";
+  }
+);

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

@@ -0,0 +1,218 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.firebase.functions
+
+import androidx.test.core.app.ApplicationProvider
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.google.common.truth.Truth.assertThat
+import com.google.firebase.Firebase
+import com.google.firebase.initialize
+import java.util.concurrent.TimeUnit
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.reactive.asFlow
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.reactivestreams.Subscriber
+import org.reactivestreams.Subscription
+
+@RunWith(AndroidJUnit4::class)
+class StreamTests {
+
+  private lateinit var functions: FirebaseFunctions
+
+  @Before
+  fun setup() {
+    Firebase.initialize(ApplicationProvider.getApplicationContext())
+    functions = Firebase.functions
+  }
+
+  internal class StreamSubscriber : Subscriber<StreamResponse> {
+    internal val messages = mutableListOf<StreamResponse.Message>()
+    internal var result: StreamResponse.Result? = null
+    internal var throwable: Throwable? = null
+    internal var isComplete = false
+    internal lateinit var subscription: Subscription
+
+    override fun onSubscribe(subscription: Subscription) {
+      this.subscription = subscription
+      subscription.request(Long.MAX_VALUE)
+    }
+
+    override fun onNext(streamResponse: StreamResponse) {
+      if (streamResponse is StreamResponse.Message) {
+        messages.add(streamResponse)
+      } else {
+        result = streamResponse as StreamResponse.Result
+      }
+    }
+
+    override fun onError(t: Throwable?) {
+      throwable = t
+    }
+
+    override fun onComplete() {
+      isComplete = true
+    }
+  }
+
+  @Test
+  fun genStream_withPublisher_receivesMessagesAndFinalResult() = runBlocking {
+    val input = mapOf("data" to "Why is the sky blue")
+    val function = functions.getHttpsCallable("genStream")
+    val subscriber = StreamSubscriber()
+
+    function.stream(input).subscribe(subscriber)
+
+    while (!subscriber.isComplete) {
+      delay(100)
+    }
+    assertThat(subscriber.messages.map { it.message.data.toString() })
+      .containsExactly("hello", "world", "this", "is", "cool")
+    assertThat(subscriber.result).isNotNull()
+    assertThat(subscriber.result!!.result.data.toString()).isEqualTo("hello world this is cool")
+    assertThat(subscriber.throwable).isNull()
+    assertThat(subscriber.isComplete).isTrue()
+  }
+
+  @Test
+  fun genStream_withFlow_receivesMessagesAndFinalResult() = runBlocking {
+    val input = mapOf("data" to "Why is the sky blue")
+    val function = functions.getHttpsCallable("genStream")
+    var isComplete = false
+    var throwable: Throwable? = null
+    val messages = mutableListOf<StreamResponse.Message>()
+    var result: StreamResponse.Result? = null
+
+    val flow = function.stream(input).asFlow()
+    try {
+      withTimeout(1000) {
+        flow.collect { response ->
+          if (response is StreamResponse.Message) {
+            messages.add(response)
+          } else {
+            result = response as StreamResponse.Result
+          }
+        }
+      }
+      isComplete = true
+    } catch (e: Throwable) {
+      throwable = e
+    }
+
+    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()
+  }
+
+  @Test
+  fun genStreamError_receivesError() = runBlocking {
+    val input = mapOf("data" to "test error")
+    val function =
+      functions.getHttpsCallable("genStreamError").withTimeout(2000, TimeUnit.MILLISECONDS)
+    val subscriber = StreamSubscriber()
+
+    function.stream(input).subscribe(subscriber)
+
+    withTimeout(2000) {
+      while (subscriber.throwable == null) {
+        delay(100)
+      }
+    }
+
+    assertThat(subscriber.throwable).isNotNull()
+    assertThat(subscriber.throwable).isInstanceOf(FirebaseFunctionsException::class.java)
+  }
+
+  @Test
+  fun genStreamWeather_receivesWeatherForecasts() = runBlocking {
+    val inputData = listOf(mapOf("name" to "Toronto"), mapOf("name" to "London"))
+    val input = mapOf("data" to inputData)
+
+    val function = functions.getHttpsCallable("genStreamWeather")
+    val subscriber = StreamSubscriber()
+
+    function.stream(input).subscribe(subscriber)
+
+    while (!subscriber.isComplete) {
+      delay(100)
+    }
+
+    assertThat(subscriber.messages.map { it.message.data.toString() })
+      .containsExactly(
+        "{temperature=25, location={name=Toronto}, conditions=snowy}",
+        "{temperature=50, location={name=London}, conditions=rainy}"
+      )
+    assertThat(subscriber.result).isNotNull()
+    assertThat(subscriber.result!!.result.data.toString()).contains("forecasts")
+    assertThat(subscriber.throwable).isNull()
+    assertThat(subscriber.isComplete).isTrue()
+  }
+
+  @Test
+  fun genStreamEmpty_receivesNoMessages() = runBlocking {
+    val function = functions.getHttpsCallable("genStreamEmpty")
+    val subscriber = StreamSubscriber()
+
+    function.stream(mapOf("data" to "test")).subscribe(subscriber)
+
+    withTimeout(2000) { delay(500) }
+    assertThat(subscriber.messages).isEmpty()
+    assertThat(subscriber.result).isNull()
+  }
+
+  @Test
+  fun genStreamResultOnly_receivesOnlyResult() = runBlocking {
+    val function = functions.getHttpsCallable("genStreamResultOnly")
+    val subscriber = StreamSubscriber()
+
+    function.stream(mapOf("data" to "test")).subscribe(subscriber)
+
+    while (!subscriber.isComplete) {
+      delay(100)
+    }
+    assertThat(subscriber.messages).isEmpty()
+    assertThat(subscriber.result).isNotNull()
+    assertThat(subscriber.result!!.result.data.toString()).isEqualTo("Only a result")
+  }
+
+  @Test
+  fun genStreamLargeData_receivesMultipleChunks() = runBlocking {
+    val function = functions.getHttpsCallable("genStreamLargeData")
+    val subscriber = StreamSubscriber()
+
+    function.stream(mapOf("data" to "test large data")).subscribe(subscriber)
+
+    while (!subscriber.isComplete) {
+      delay(100)
+    }
+    assertThat(subscriber.messages).isNotEmpty()
+    assertThat(subscriber.messages.size).isEqualTo(10)
+    val receivedString =
+      subscriber.messages.joinToString(separator = "") { it.message.data.toString() }
+    val expectedString = "A".repeat(10000)
+    assertThat(receivedString.length).isEqualTo(10000)
+    assertThat(receivedString).isEqualTo(expectedString)
+    assertThat(subscriber.result).isNotNull()
+    assertThat(subscriber.result!!.result.data.toString()).isEqualTo("Stream Completed")
+  }
+}