feat(functions): add streamAsFlow()

Author: thatfiredev

firebase-functions/api.txt

@@ -86,6 +86,8 @@ package com.google.firebase.functions {
     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 kotlinx.coroutines.flow.Flow<com.google.firebase.functions.StreamResponse> streamAsFlow();
+    method public kotlinx.coroutines.flow.Flow<com.google.firebase.functions.StreamResponse> streamAsFlow(Object? data = null);
     method public com.google.firebase.functions.HttpsCallableReference withTimeout(long timeout, java.util.concurrent.TimeUnit units);
     property public final long timeout;
   }

firebase-functions/firebase-functions.gradle.kts

@@ -113,6 +113,7 @@ dependencies {
   implementation(libs.playservices.base)
   implementation(libs.playservices.basement)
   implementation(libs.reactive.streams)
+  implementation(libs.kotlinx.coroutines.reactive)
 
   api(libs.playservices.tasks)
 
@@ -133,7 +134,6 @@ 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")

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

@@ -23,7 +23,6 @@ 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
@@ -100,7 +99,7 @@ class StreamTests {
     val messages = mutableListOf<StreamResponse.Message>()
     var result: StreamResponse.Result? = null
 
-    val flow = function.stream(input).asFlow()
+    val flow = function.streamAsFlow(input)
     try {
       withTimeout(1000) {
         flow.collect { response ->

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

@@ -15,6 +15,8 @@ package com.google.firebase.functions
 
 import androidx.annotation.VisibleForTesting
 import com.google.android.gms.tasks.Task
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.reactive.asFlow
 import java.net.URL
 import java.util.concurrent.TimeUnit
 import org.reactivestreams.Publisher
@@ -173,6 +175,50 @@ public class HttpsCallableReference {
     }
   }
 
+  /**
+   * Streams data to the specified HTTPS endpoint.
+   *
+   * The data passed into the trigger can be any of the following types:
+   *
+   * * Any primitive type, including null, int, long, float, and boolean.
+   * * [String]
+   * * [List<?>][java.util.List], where the contained objects are also one of these types.
+   * * [Map<String, ?>][java.util.Map], where the values are also one of these types.
+   * * [org.json.JSONArray]
+   * * [org.json.JSONObject]
+   * * [org.json.JSONObject.NULL]
+   *
+   * If the returned streamResponse fails, the exception will be one of the following types:
+   *
+   * * [java.io.IOException]
+   * - if the HTTPS request failed to connect.
+   * * [FirebaseFunctionsException]
+   * - if the request connected, but the function returned an error.
+   *
+   * The request to the Cloud Functions backend made by this method automatically includes a
+   * Firebase Instance ID token to identify the app instance. If a user is logged in with Firebase
+   * Auth, an auth token for the user will also be automatically included.
+   *
+   * Firebase Instance ID sends data to the Firebase backend periodically to collect information
+   * regarding the app instance. To stop this, see
+   * [com.google.firebase.iid.FirebaseInstanceId.deleteInstanceId]. It will resume with a new
+   * Instance ID the next time you call this method.
+   *
+   * @param data Parameters to pass to the endpoint. Defaults to `null` if not provided.
+   * @return [Flow] that will emit intermediate data, and the final result, as it is generated
+   * by the function.
+   * @see org.json.JSONArray
+   *
+   * @see org.json.JSONObject
+   *
+   * @see java.io.IOException
+   *
+   * @see FirebaseFunctionsException
+   */
+  @JvmOverloads
+  public fun streamAsFlow(data: Any? = null): Flow<StreamResponse> =
+    stream(data).asFlow()
+
   /**
    * Changes the timeout for calls from this instance of Functions. The default is 60 seconds.
    *