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

firebase-functions/api.txt

@@ -86,6 +86,8 @@ package com.google.firebase.functions {
     method @NonNull public com.google.android.gms.tasks.Task<com.google.firebase.functions.HttpsCallableResult> call();
     method public long getTimeout();
     method public void setTimeout(long timeout, @NonNull java.util.concurrent.TimeUnit units);
+    method @NonNull public org.reactivestreams.Publisher stream(@Nullable Object data = null);
+    method @NonNull public org.reactivestreams.Publisher stream();
     method @NonNull public com.google.firebase.functions.HttpsCallableReference withTimeout(long timeout, @NonNull java.util.concurrent.TimeUnit units);
     property public final long timeout;
   }

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)

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

@@ -122,3 +122,58 @@ exports.timeoutTest = functions.https.onRequest((request, response) => {
   // Wait for longer than 500ms.
   setTimeout(() => response.send({data: true}), 500);
 });
+
+const data = ["hello", "world", "this", "is", "cool"];
+
+/**
+ * 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));
+}
+
+/**
+ * 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 data) {
+    yield chunk;
+    await sleep(1000);
+  }
+}
+
+exports.genStream = functions.https.onCall(async (request, response) => {
+  if (response && response.acceptsStreaming) {
+    for await (const chunk of generateText()) {
+      console.log("got chunk", chunk);
+      response.write({chunk});
+    }
+  }
+  return data.join(" ");
+});
+
+exports.genStreamError = functions.https.onCall(async (request, response) => {
+  if (response && response.acceptsStreaming) {
+    for await (const chunk of generateText()) {
+      console.log("got chunk", chunk);
+      response.write({chunk});
+    }
+    throw new Error("BOOM");
+  }
+});
+
+exports.genStreamNoReturn = functions.https.onCall(
+    async (request, response) => {
+      if (response && response.acceptsStreaming) {
+        for await (const chunk of generateText()) {
+          console.log("got chunk", chunk);
+          response.write({chunk});
+        }
+      }
+    },
+);

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

@@ -0,0 +1,160 @@
+package com.google.firebase.functions.ktx
+
+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.functions.FirebaseFunctions
+import com.google.firebase.functions.functions
+import com.google.firebase.initialize
+import java.util.concurrent.TimeUnit
+import org.junit.After
+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
+  var onNextList = mutableListOf<Any>()
+  private lateinit var subscriber: Subscriber<Any>
+  private var throwable: Throwable? = null
+  private var isComplete = false
+
+  @Before
+  fun setup() {
+    Firebase.initialize(ApplicationProvider.getApplicationContext())
+    functions = Firebase.functions
+    subscriber =
+      object : Subscriber<Any> {
+        override fun onSubscribe(subscription: Subscription?) {
+          subscription?.request(1)
+        }
+
+        override fun onNext(t: Any) {
+          onNextList.add(t)
+        }
+
+        override fun onError(t: Throwable?) {
+          throwable = t
+        }
+
+        override fun onComplete() {
+          isComplete = true
+        }
+      }
+  }
+
+  @After
+  fun clear() {
+    onNextList.clear()
+    throwable = null
+    isComplete = false
+  }
+
+  @Test
+  fun testGenStream() {
+    val input = hashMapOf("data" to "Why is the sky blue")
+    val function = functions.getHttpsCallable("genStream")
+
+    function.stream(input).subscribe(subscriber)
+
+    Thread.sleep(8000)
+    val onNextStringList = onNextList.map { it.toString() }
+    assertThat(onNextStringList)
+      .containsExactly(
+        "{chunk=hello}",
+        "{chunk=world}",
+        "{chunk=this}",
+        "{chunk=is}",
+        "{chunk=cool}",
+        "hello world this is cool"
+      )
+    assertThat(throwable).isNull()
+    assertThat(isComplete).isTrue()
+  }
+
+  @Test
+  fun testGenStreamError() {
+    val input = hashMapOf("data" to "Why is the sky blue")
+    val function =
+      functions.getHttpsCallable("genStreamError").withTimeout(800, TimeUnit.MILLISECONDS)
+
+    function.stream(input).subscribe(subscriber)
+    Thread.sleep(2000)
+
+    val onNextStringList = onNextList.map { it.toString() }
+    assertThat(onNextStringList)
+      .containsExactly(
+        "{chunk=hello}",
+      )
+    assertThat(throwable).isNotNull()
+    assertThat(isComplete).isFalse()
+  }
+
+  @Test
+  fun testGenStreamNoReturn() {
+    val input = hashMapOf("data" to "Why is the sky blue")
+    val function = functions.getHttpsCallable("genStreamNoReturn")
+
+    function.stream(input).subscribe(subscriber)
+    Thread.sleep(8000)
+
+    val onNextStringList = onNextList.map { it.toString() }
+    assertThat(onNextStringList)
+      .containsExactly(
+        "{chunk=hello}",
+        "{chunk=world}",
+        "{chunk=this}",
+        "{chunk=is}",
+        "{chunk=cool}"
+      )
+    assertThat(throwable).isNull()
+    assertThat(isComplete).isFalse()
+  }
+
+  @Test
+  fun testGenStream_cancelStream() {
+    val input = hashMapOf("data" to "Why is the sky blue")
+    val function = functions.getHttpsCallable("genStreamNoReturn")
+    val publisher = function.stream(input)
+    var subscription: Subscription? = null
+    val cancelableSubscriber =
+      object : Subscriber<Any> {
+        override fun onSubscribe(s: Subscription?) {
+          subscription = s
+          s?.request(1)
+        }
+
+        override fun onNext(message: Any) {
+          onNextList.add(message)
+        }
+
+        override fun onError(t: Throwable?) {
+          throwable = t
+        }
+
+        override fun onComplete() {
+          isComplete = true
+        }
+      }
+
+    publisher.subscribe(cancelableSubscriber)
+    Thread.sleep(2000)
+    subscription?.cancel()
+    Thread.sleep(6000)
+
+    val onNextStringList = onNextList.map { it.toString() }
+    assertThat(onNextStringList)
+      .containsExactly(
+        "{chunk=hello}",
+        "{chunk=world}",
+      )
+    assertThat(throwable).isNotNull()
+    assertThat(requireNotNull(throwable).message).isEqualTo("Stream was canceled")
+    assertThat(isComplete).isFalse()
+  }
+}