feat(cloud_functions): add support for cloud functions stream

.github/workflows/scripts/functions/package.json

@@ -14,8 +14,8 @@
   },
   "main": "lib/index.js",
   "dependencies": {
-    "firebase-admin": "^11.5.0",
-    "firebase-functions": "^4.5.0"
+    "firebase-admin": "^13.2.0",
+    "firebase-functions": "^6.3.2"
   },
   "devDependencies": {
     "firebase-functions-test": "^0.2.0",

.github/workflows/scripts/functions/src/index.ts

@@ -2,6 +2,7 @@ import * as assert from 'assert';
 import * as functions from 'firebase-functions';
 import * as functionsv2 from 'firebase-functions/v2';
 
+
 // For example app.
 // noinspection JSUnusedGlobalSymbols
 export const listFruit = functions.https.onCall(() => {
@@ -14,12 +15,17 @@ export const listfruits2ndgen = functionsv2.https.onCall(() => {
 
 // For e2e testing a custom region.
 // noinspection JSUnusedGlobalSymbols
-export const testFunctionCustomRegion = functions
-  .region('europe-west1')
-  .https.onCall(() => 'europe-west1');
+
+export const testFunctionCustomRegion = functions.https.onCall(
+  {
+    region: 'europe-west1'
+  },
+  () => 'europe-west1'
+);
 
 // For e2e testing timeouts.
-export const testFunctionTimeout = functions.https.onCall((data) => {
+export const testFunctionTimeout = functions.https.onCall((req, res) => {
+  const data = req.data
   console.log(JSON.stringify({ data }));
   return new Promise((resolve, reject) => {
     if (data && data.testTimeout) {
@@ -40,7 +46,8 @@ export const testFunctionTimeout = functions.https.onCall((data) => {
 
 // For e2e testing errors & return values.
 // noinspection JSUnusedGlobalSymbols
-export const testFunctionDefaultRegion = functions.https.onCall((data) => {
+export const testFunctionDefaultRegion = functions.https.onCall((req, res) => {
+  const data = req.data;
   console.log(JSON.stringify({ data }));
   if (typeof data === 'undefined') {
     return 'undefined';
@@ -66,7 +73,7 @@ export const testFunctionDefaultRegion = functions.https.onCall((data) => {
     return 'array';
   }
 
-  if(data.type === 'rawData') {
+  if (data.type === 'rawData') {
     return data;
   }
 
@@ -159,3 +166,18 @@ export const testFunctionDefaultRegion = functions.https.onCall((data) => {
 export const testMapConvertType = functions.https.onCall((data) => ({
   foo: 'bar',
 }));
+
+exports.getFruits = functions.https.onCall(async (request, response) => {
+  const fruits = ['Apple', 'Mango', 'Banana']
+
+  const allFruits = fruits.map(async (fruit) => {
+    // Stream each fruit as it resolves!
+    if (request.acceptsStreaming) {
+      response?.sendChunk(fruit);
+    }
+    return fruit;
+  });
+
+  // Fallback for non-streaming clients
+  return Promise.all(allFruits);
+});

packages/cloud_functions/cloud_functions/android/build.gradle

@@ -64,6 +64,7 @@ android {
     implementation platform("com.google.firebase:firebase-bom:${getRootProjectExtOrCoreProperty("FirebaseSDKVersion", firebaseCoreProject)}")
     implementation 'com.google.firebase:firebase-functions'
     implementation 'androidx.annotation:annotation:1.7.0'
+    implementation 'org.reactivestreams:reactive-streams:1.0.4'
   }
 }
 

packages/cloud_functions/cloud_functions/android/src/main/java/io/flutter/plugins/firebase/functions/FirebaseFunctionsStreamHandler.java

@@ -0,0 +1,85 @@
+// Copyright 2018 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package io.flutter.plugins.firebase.functions;
+
+import android.net.Uri;
+import com.google.firebase.functions.FirebaseFunctions;
+import com.google.firebase.functions.HttpsCallableOptions;
+import com.google.firebase.functions.HttpsCallableReference;
+import com.google.firebase.functions.StreamResponse;
+import io.flutter.plugin.common.EventChannel;
+import io.flutter.plugin.common.EventChannel.StreamHandler;
+import java.net.URL;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import org.reactivestreams.Publisher;
+
+public class FirebaseFunctionsStreamHandler implements StreamHandler {
+
+  private final FirebaseFunctions firebaseFunctions;
+
+  private StreamResponseSubscriber subscriber;
+
+  public FirebaseFunctionsStreamHandler(FirebaseFunctions functions) {
+    this.firebaseFunctions = functions;
+  }
+
+  @Override
+  public void onListen(Object arguments, EventChannel.EventSink events) {
+    @SuppressWarnings("unchecked")
+    Map<String, Object> argumentsMap = (Map<String, Object>) arguments;
+    httpsStreamCall(argumentsMap, events);
+  }
+
+  @Override
+  public void onCancel(Object arguments) {
+    subscriber.cancel();
+  }
+
+  private void httpsStreamCall(Map<String, Object> arguments, EventChannel.EventSink events) {
+    try {
+
+      String functionName = (String) arguments.get("functionName");
+      String functionUri = (String) arguments.get("functionUri");
+      String origin = (String) arguments.get("origin");
+      Integer timeout = (Integer) arguments.get("timeout");
+      Object parameters = arguments.get("parameters");
+      boolean limitedUseAppCheckToken =
+          (boolean) Objects.requireNonNull(arguments.get("limitedUseAppCheckToken"));
+
+      if (origin != null) {
+        Uri originUri = Uri.parse(origin);
+        firebaseFunctions.useEmulator(originUri.getHost(), originUri.getPort());
+      }
+
+      HttpsCallableReference httpsCallableReference;
+      HttpsCallableOptions options =
+          new HttpsCallableOptions.Builder()
+              .setLimitedUseAppCheckTokens(limitedUseAppCheckToken)
+              .build();
+
+      Publisher<StreamResponse> publisher;
+      if (functionName != null) {
+        httpsCallableReference = firebaseFunctions.getHttpsCallable(functionName, options);
+        publisher = httpsCallableReference.stream(parameters);
+      } else if (functionUri != null) {
+        httpsCallableReference =
+            firebaseFunctions.getHttpsCallableFromUrl(new URL(functionUri), options);
+        publisher = httpsCallableReference.stream();
+      } else {
+        throw new IllegalArgumentException("Either functionName or functionUri must be set");
+      }
+
+      if (timeout != null) {
+        httpsCallableReference.setTimeout(timeout.longValue(), TimeUnit.MILLISECONDS);
+      }
+      subscriber = new StreamResponseSubscriber(events);
+      publisher.subscribe(subscriber);
+    } catch (Exception e) {
+      events.error("firebase_functions", e.getMessage(), null);
+    }
+  }
+}

packages/cloud_functions/cloud_functions/android/src/main/java/io/flutter/plugins/firebase/functions/FlutterFirebaseFunctionsPlugin.java

@@ -17,6 +17,7 @@
 import com.google.firebase.functions.HttpsCallableReference;
 import com.google.firebase.functions.HttpsCallableResult;
 import io.flutter.embedding.engine.plugins.FlutterPlugin;
+import io.flutter.plugin.common.EventChannel;
 import io.flutter.plugin.common.MethodCall;
 import io.flutter.plugin.common.MethodChannel;
 import io.flutter.plugin.common.MethodChannel.MethodCallHandler;
@@ -35,6 +36,7 @@ public class FlutterFirebaseFunctionsPlugin
 
   private static final String METHOD_CHANNEL_NAME = "plugins.flutter.io/firebase_functions";
   private MethodChannel channel;
+  private FlutterPluginBinding pluginBinding;
 
   /**
    * Default Constructor.
@@ -45,6 +47,7 @@ public FlutterFirebaseFunctionsPlugin() {}
 
   @Override
   public void onAttachedToEngine(@NonNull FlutterPluginBinding binding) {
+    pluginBinding = binding;
     channel = new MethodChannel(binding.getBinaryMessenger(), METHOD_CHANNEL_NAME);
     channel.setMethodCallHandler(this);
   }
@@ -55,6 +58,16 @@ public void onDetachedFromEngine(@NonNull FlutterPluginBinding binding) {
     channel = null;
   }
 
+  private void registerEventChannel(Map<String, Object> arguments) {
+    final String eventId = (String) Objects.requireNonNull(arguments.get("eventChannelId"));
+    final String eventChannelName = METHOD_CHANNEL_NAME + "/" + eventId;
+    final EventChannel eventChannel =
+        new EventChannel(pluginBinding.getBinaryMessenger(), eventChannelName);
+    FirebaseFunctions functions = getFunctions(arguments);
+    FirebaseFunctionsStreamHandler streamHandler = new FirebaseFunctionsStreamHandler(functions);
+    eventChannel.setStreamHandler(streamHandler);
+  }
+
   private FirebaseFunctions getFunctions(Map<String, Object> arguments) {
     String appName = (String) Objects.requireNonNull(arguments.get("appName"));
     String region = (String) Objects.requireNonNull(arguments.get("region"));
@@ -116,24 +129,26 @@ private Task<Object> httpsFunctionCall(Map<String, Object> arguments) {
 
   @Override
   public void onMethodCall(MethodCall call, @NonNull final Result result) {
-    if (!call.method.equals("FirebaseFunctions#call")) {
+    if (call.method.equals("FirebaseFunctions#registerEventChannel")) {
+      registerEventChannel(call.arguments());
+      result.success(null);
+    } else if (call.method.equals("FirebaseFunctions#call")) {
+      httpsFunctionCall(call.arguments())
+          .addOnCompleteListener(
+              task -> {
+                if (task.isSuccessful()) {
+                  result.success(task.getResult());
+                } else {
+                  Exception exception = task.getException();
+                  result.error(
+                      "firebase_functions",
+                      exception != null ? exception.getMessage() : null,
+                      getExceptionDetails(exception));
+                }
+              });
+    } else {
       result.notImplemented();
-      return;
     }
-
-    httpsFunctionCall(call.arguments())
-        .addOnCompleteListener(
-            task -> {
-              if (task.isSuccessful()) {
-                result.success(task.getResult());
-              } else {
-                Exception exception = task.getException();
-                result.error(
-                    "firebase_functions",
-                    exception != null ? exception.getMessage() : null,
-                    getExceptionDetails(exception));
-              }
-            });
   }
 
   private Map<String, Object> getExceptionDetails(@Nullable Exception exception) {

packages/cloud_functions/cloud_functions/android/src/main/java/io/flutter/plugins/firebase/functions/StreamResponseSubscriber.java

@@ -0,0 +1,65 @@
+// Copyright 2018 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package io.flutter.plugins.firebase.functions;
+
+import android.os.Handler;
+import android.os.Looper;
+import com.google.firebase.functions.StreamResponse;
+import io.flutter.plugin.common.EventChannel;
+import java.util.HashMap;
+import java.util.Map;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+public class StreamResponseSubscriber implements Subscriber<StreamResponse> {
+  private Subscription subscription;
+  private final EventChannel.EventSink eventSink;
+
+  private final Handler mainThreadHandler = new Handler(Looper.getMainLooper());
+
+  public StreamResponseSubscriber(EventChannel.EventSink eventSink) {
+    this.eventSink = eventSink;
+  }
+
+  @Override
+  public void onSubscribe(Subscription s) {
+    this.subscription = s;
+    subscription.request(Long.MAX_VALUE);
+  }
+
+  @Override
+  public void onNext(StreamResponse streamResponse) {
+    Map<String, Object> responseMap = new HashMap<>();
+    if (streamResponse instanceof StreamResponse.Message) {
+      Object message = ((StreamResponse.Message) streamResponse).getMessage().getData();
+      responseMap.put("message", message);
+      mainThreadHandler.post(() -> eventSink.success(responseMap));
+    } else {
+      Object result = ((StreamResponse.Result) streamResponse).getResult().getData();
+      responseMap.put("result", result);
+      mainThreadHandler.post(() -> eventSink.success(responseMap));
+    }
+  }
+
+  @Override
+  public void onError(Throwable t) {
+    if (eventSink != null) {
+      eventSink.error("firebase_functions", t.getMessage(), null);
+    }
+  }
+
+  @Override
+  public void onComplete() {
+    if (eventSink != null) {
+      eventSink.endOfStream();
+    }
+  }
+
+  public void cancel() {
+    if (subscription != null) {
+      subscription.cancel();
+    }
+  }
+}

packages/cloud_functions/cloud_functions/example/android/app/src/main/AndroidManifest.xml

@@ -2,6 +2,7 @@
     <application
         android:label="example"
         android:name="${applicationName}"
+        android:usesCleartextTraffic="true"
         android:icon="@mipmap/ic_launcher">
         <activity
             android:name=".MainActivity"

packages/cloud_functions/cloud_functions/example/android/settings.gradle

@@ -18,7 +18,7 @@ pluginManagement {
 
 plugins {
     id "dev.flutter.flutter-plugin-loader" version "1.0.0"
-    id "com.android.application" version "7.3.0" apply false
+    id "com.android.application" version "8.1.0" apply false
     // START: FlutterFire Configuration
     id "com.google.gms.google-services" version "4.3.15" apply false
     // END: FlutterFire Configuration