feat(cloud_functions): add support for cloud functions stream

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

@@ -33,21 +33,30 @@ export const testFunctionCustomRegion = functions.https.onCall(
 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) {
-      setTimeout(
-        () => resolve({ timeLimit: 'exceeded' }),
-        parseInt(data.testTimeout, 10)
-      );
-    } else {
-      reject(
-        new functions.https.HttpsError(
-          'invalid-argument',
-          'testTimeout must be provided.'
-        )
-      );
-    }
+
+  const timeoutMs = parseInt(data?.testTimeout, 10);
+
+  if (isNaN(timeoutMs)) {
+    throw new functions.https.HttpsError(
+      'invalid-argument',
+      'testTimeout must be provided.'
+    );
+  }
+
+  if (req.acceptsStreaming) {
+    setTimeout(() => {
+      res?.sendChunk({ timeLimit: 'exceeded' });
+    }, timeoutMs);
+
+    return new Promise((resolve) => {
+      setTimeout(resolve, timeoutMs + 100);
+    });
+  }
+
+  return new Promise((resolve) => {
+    setTimeout(() => resolve({ timeLimit: 'exceeded' }), timeoutMs);
   });
+
 });
 
 // For e2e testing errors & return values.

packages/cloud_functions/cloud_functions/lib/cloud_functions.dart

@@ -15,7 +15,13 @@ import 'package:firebase_core_platform_interface/firebase_core_platform_interfac
 import 'package:flutter/foundation.dart';
 
 export 'package:cloud_functions_platform_interface/cloud_functions_platform_interface.dart'
-    show HttpsCallableOptions, FirebaseFunctionsException;
+    show
+        HttpsCallableOptions,
+        FirebaseFunctionsException,
+        AbortSignal,
+        TimeLimit,
+        Abort,
+        Any;
 
 part 'src/firebase_functions.dart';
 part 'src/https_callable.dart';

packages/cloud_functions/cloud_functions_platform_interface/lib/src/https_callable_options.dart

@@ -10,11 +10,72 @@ class HttpsCallableOptions {
   /// Defaults [limitedUseAppCheckToken] to `false`
   HttpsCallableOptions(
       {this.timeout = const Duration(seconds: 60),
-      this.limitedUseAppCheckToken = false});
+      this.limitedUseAppCheckToken = false,
+      this.webAbortSignal});
 
   /// Returns the timeout for this instance
   Duration timeout;
 
   /// Sets whether or not to use limited-use App Check tokens when invoking the associated function.
   bool limitedUseAppCheckToken;
+
+  /// An AbortSignal that can be used to cancel the streaming response.
+  /// When the signal is aborted, the underlying HTTP connection will be terminated.
+  AbortSignal? webAbortSignal;
+}
+
+/// Represents a base class for encapsulating abort signals.
+sealed class AbortSignal {}
+
+/// Creates an [AbortSignal] that will automatically abort after a specified [time].
+///
+/// This is equivalent to calling `AbortSignal.timeout(ms)` in the Web SDK.
+///
+/// Typically used to cancel long-running operations after a timeout duration.
+///
+/// Example:
+/// ```dart
+/// final signal = HttpsCallableOptions(webAbortSignal: TimeLimit(Duration(seconds: 10)));
+/// ```
+class TimeLimit extends AbortSignal {
+  final Duration time;
+  TimeLimit(this.time);
+}
+
+/// Creates an [AbortSignal] that is immediately aborted with an optional [reason].
+///
+/// This is equivalent to calling `AbortSignal.abort(reason)` in the Web SDK.
+///
+/// Useful when you want to explicitly cancel a callable before it begins, or to provide
+/// a specific reason for cancellation.
+///
+/// Example:
+/// ```dart
+/// final signal = HttpsCallableOptions(webAbortSignal: Abort('User exited'));
+/// ```
+
+class Abort extends AbortSignal {
+  final Object? reason;
+  Abort([this.reason]);
+}
+
+/// Creates an [AbortSignal] that is aborted when **any** of the provided [signals] is aborted.
+///
+/// This is equivalent to calling `AbortSignal.any([...])` in the Web SDK.
+///
+/// Useful for combining multiple abort conditions.
+///
+/// Example:
+/// ```dart
+/// final signal = HttpsCallableOptions(
+///   webAbortSignal: Any([
+///     TimeLimit(Duration(seconds: 10)),
+///     Abort('User cancelled'),
+///   ]),
+/// );
+/// ```
+
+class Any extends AbortSignal {
+  final List<AbortSignal> signals;
+  Any(this.signals);
 }

packages/cloud_functions/cloud_functions_web/lib/https_callable_web.dart

@@ -11,6 +11,7 @@ import 'package:cloud_functions_web/interop/functions_interop.dart' as interop;
 
 import 'interop/functions.dart' as functions_interop;
 import 'utils.dart';
+import 'package:web/web.dart' as web;
 
 /// A web specific implementation of [HttpsCallable].
 class HttpsCallableWeb extends HttpsCallablePlatform {
@@ -77,9 +78,14 @@ class HttpsCallableWeb extends HttpsCallablePlatform {
     }
 
     final JSAny? parametersJS = parameters?.jsify();
+    web.AbortSignal? signal;
+    if (options.webAbortSignal != null) {
+      signal = _createJsAbortSignal(options.webAbortSignal!);
+    }
     interop.HttpsCallableStreamOptions callableStreamOptions =
         interop.HttpsCallableStreamOptions(
-            limitedUseAppCheckTokens: options.limitedUseAppCheckToken.toJS);
+            limitedUseAppCheckTokens: options.limitedUseAppCheckToken.toJS,
+            signal: signal);
     try {
       await for (final value
           in callable.stream(parametersJS, callableStreamOptions)) {
@@ -89,4 +95,20 @@ class HttpsCallableWeb extends HttpsCallablePlatform {
       throw convertFirebaseFunctionsException(e as JSObject, s);
     }
   }
+
+  web.AbortSignal _createJsAbortSignal(AbortSignal signal) {
+    try {
+      switch (signal) {
+        case TimeLimit(:final time):
+          return web.AbortSignal.timeout(time.inMilliseconds);
+        case Abort(:final reason):
+          return web.AbortSignal.abort(reason.jsify());
+        case Any(:final signals):
+          final jsSignals = signals.map(_createJsAbortSignal).toList().toJS;
+          return web.AbortSignal.any(jsSignals);
+      }
+    } catch (e, s) {
+      throw convertFirebaseFunctionsException(e as JSObject, s);
+    }
+  }
 }

packages/cloud_functions/cloud_functions_web/lib/interop/functions_interop.dart

@@ -102,12 +102,14 @@ extension HttpsCallableStreamResultJsImplExtension
 @anonymous
 abstract class HttpsCallableStreamOptions {
   external factory HttpsCallableStreamOptions(
-      {JSBoolean? limitedUseAppCheckTokens});
+      {JSBoolean? limitedUseAppCheckTokens, web.AbortSignal? signal});
 }
 
 extension HttpsCallableStreamOptionsExtension on HttpsCallableStreamOptions {
   external JSBoolean? get limitedUseAppCheckTokens;
   external set limitedUseAppCheckTokens(JSBoolean? t);
+  external web.AbortSignal? signal;
+  external set siganl(web.AbortSignal? s);
 }
 
 extension type JsAsyncIterator<T extends JSAny>._(JSObject _)

tests/integration_test/cloud_functions/cloud_functions_e2e_test.dart

@@ -2,6 +2,7 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'dart:async';
 import 'dart:io';
 
 import 'package:cloud_functions/cloud_functions.dart';
@@ -354,7 +355,8 @@ void main() {
       });
 
       test('accepts a [List]', () async {
-        final stream = callable.stream(data.list).where((event) => event is Chunk);
+        final stream =
+            callable.stream(data.list).where((event) => event is Chunk);
         await expectLater(
           stream,
           emits(
@@ -366,17 +368,54 @@ void main() {
 
       test('accepts a deeply nested [Map]', () async {
         final stream = callable.stream({
-            'type': 'deepMap',
-            'inputData': data.deepMap,
-          }).where((event) => event is Chunk);
+          'type': 'deepMap',
+          'inputData': data.deepMap,
+        }).where((event) => event is Chunk);
         await expectLater(
           stream,
           emits(
-            isA<Chunk>()
-                .having((e) => e.partialData, 'partialData', equals(data.deepMap)),
+            isA<Chunk>().having(
+              (e) => e.partialData,
+              'partialData',
+              equals(data.deepMap),
+            ),
           ),
         );
       });
+
+      test(
+        'times out when aborted with TimeLimit signal',
+        () async {
+          final instance = FirebaseFunctions.instance;
+          instance.useFunctionsEmulator('localhost', 5001);
+
+          final completer = Completer<void>();
+
+          final timeoutCallable = FirebaseFunctions.instance.httpsCallable(
+            kTestFunctionTimeout,
+            options: HttpsCallableOptions(
+              webAbortSignal: TimeLimit(const Duration(seconds: 3)),
+            ),
+          );
+
+          timeoutCallable.stream().listen(
+            (data) {
+              completer.completeError('Should have thrown');
+            },
+            onError: (error) {
+              if (error is FirebaseFunctionsException) {
+                expect(error.code, equals('deadline-exceeded'));
+                completer.complete();
+              } else {
+                completer.completeError('Unexpected error type: $error');
+              }
+            },
+          );
+
+          await completer.future;
+        },
+        skip: !kIsWeb,
+      );
     });
   });
 }