[Functions] Add support for streamable cloud functions

.github/workflows/functions.yml

@@ -31,8 +31,6 @@ jobs:
       matrix:
         target: [ios, tvos, macos, watchos]
         build-env:
-          - os: macos-14
-            xcode: Xcode_15.2
           - os: macos-15
             xcode: Xcode_16.2
     runs-on: ${{ matrix.build-env.os }}
@@ -43,14 +41,12 @@ jobs:
       run: sudo xcode-select -s /Applications/${{ matrix.build-env.xcode }}.app/Contents/Developer
     - name: Setup Bundler
       run: scripts/setup_bundler.sh
-    # The integration tests are flaky on Xcode 15 so only run the unit tests. The integration tests still run with SPM.
-    # - name: Integration Test Server
-    #   run: FirebaseFunctions/Backend/start.sh synchronous
+    - name: Integration Test Server
+      run: FirebaseFunctions/Backend/start.sh synchronous
     - name: Build and test
       run: |
         scripts/third_party/travis/retry.sh scripts/pod_lib_lint.rb FirebaseFunctions.podspec \
-          --test-specs=unit --platforms=${{ matrix.target }}
-
+          --platforms=${{ matrix.target }}
 
   spm-package-resolved:
     runs-on: macos-14
@@ -145,6 +141,9 @@ jobs:
         key: ${{needs.spm-package-resolved.outputs.cache_key}}
     - name: Xcode
       run: sudo xcode-select -s /Applications/${{ matrix.xcode }}.app/Contents/Developer
+    - name: Install visionOS, if needed.
+      if: matrix.target == 'visionOS'
+      run: xcodebuild -downloadPlatform visionOS
     - name: Initialize xcodebuild
       run: scripts/setup_spm_tests.sh
     - name: Unit Tests

FirebaseFunctions/Backend/index.js

@@ -16,6 +16,14 @@ const assert = require('assert');
 const functionsV1 = require('firebase-functions/v1');
 const functionsV2 = require('firebase-functions/v2');
 
+// MARK: - Utilities
+
+function sleep(ms) {
+  return new Promise(resolve => setTimeout(resolve, ms));
+};
+
+// MARK: - Callable Functions
+
 exports.dataTest = functionsV1.https.onRequest((request, response) => {
   assert.deepEqual(request.body, {
     data: {
@@ -121,22 +129,18 @@ exports.timeoutTest = functionsV1.https.onRequest((request, response) => {
 
 const streamData = ["hello", "world", "this", "is", "cool"]
 
-function sleep(ms) {
-  return new Promise(resolve => setTimeout(resolve, ms));
-};
-
 async function* generateText() {
   for (const chunk of streamData) {
     yield chunk;
-    await sleep(1000);
+    await sleep(100);
   }
 };
 
 exports.genStream = functionsV2.https.onCall(
   async (request, response) => {
     if (request.acceptsStreaming) {
       for await (const chunk of generateText()) {
-        response.sendChunk({ chunk });
+        response.sendChunk(chunk);
       }
     }
     return streamData.join(" ");
@@ -145,11 +149,81 @@ exports.genStream = functionsV2.https.onCall(
 
 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 }
+};
+
+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 forecasts = [];
     if (request.acceptsStreaming) {
-      for await (const chunk of generateText()) {
-        response.write({ chunk });
+      for await (const chunk of generateForecast(request.data)) {
+        forecasts.push(chunk)
+        response.sendChunk(chunk);
+      }
+    }
+    return { forecasts };
+  }
+);
+
+exports.genStreamWeatherError = functionsV2.https.onCall(
+  async (request, response) => {
+    if (request.acceptsStreaming) {
+      for await (const chunk of generateForecast(request.data)) {
+        // Remove the location field, since the SDK cannot decode the message
+        // if it's there.
+        delete chunk.location;
+        response.sendChunk(chunk);
+      }
+    }
+    return "Number of forecasts generated: " + request.data.length;
+  }
+);
+
+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);
       }
-      throw Error("BOOM")
     }
+    return "Stream Completed";
   }
 );

FirebaseFunctions/Backend/start.sh

@@ -57,6 +57,11 @@ FUNCTIONS_BIN="./node_modules/.bin/functions"
 "${FUNCTIONS_BIN}" deploy timeoutTest --trigger-http
 "${FUNCTIONS_BIN}" deploy genStream --trigger-http
 "${FUNCTIONS_BIN}" deploy genStreamError --trigger-http
+"${FUNCTIONS_BIN}" deploy genStreamWeather --trigger-http
+"${FUNCTIONS_BIN}" deploy genStreamWeatherError --trigger-http
+"${FUNCTIONS_BIN}" deploy genStreamEmpty --trigger-http
+"${FUNCTIONS_BIN}" deploy genStreamResultOnly --trigger-http
+"${FUNCTIONS_BIN}" deploy genStreamLargeData --trigger-http
 
 if [ "$1" != "synchronous" ]; then
   # Wait for the user to tell us to stop the server.

FirebaseFunctions/CHANGELOG.md

@@ -1,3 +1,6 @@
+# Unreleased
+- [added] Streaming callable functions are now supported.
+
 # 11.9.0
 - [fixed] Fixed App Check token reporting to enable differentiating outdated
   (`MISSING`) and inauthentic (`INVALID`) clients; see [Monitor App Check

FirebaseFunctions/Sources/Callable+Codable.swift

@@ -15,7 +15,11 @@
 import FirebaseSharedSwift
 import Foundation
 
-/// A `Callable` is reference to a particular Callable HTTPS trigger in Cloud Functions.
+/// A `Callable` is a reference to a particular Callable HTTPS trigger in Cloud Functions.
+///
+/// - Note: If the Callable HTTPS trigger accepts no parameters, ``Never`` can be used for
+///   iOS 17.0+. Otherwise, a simple encodable placeholder type (e.g.,
+///   `struct EmptyRequest: Encodable {}`) can be used.
 public struct Callable<Request: Encodable, Response: Decodable> {
   /// The timeout to use when calling the function. Defaults to 70 seconds.
   public var timeoutInterval: TimeInterval {
@@ -160,3 +164,175 @@ public struct Callable<Request: Encodable, Response: Decodable> {
     return try await call(data)
   }
 }
+
+/// Used to determine when a `StreamResponse<_, _>` is being decoded.
+private protocol StreamResponseProtocol {}
+
+/// A convenience type used to receive both the streaming callable function's yielded messages and
+/// its return value.
+///
+/// This can be used as the generic `Response` parameter to ``Callable`` to receive both the
+/// yielded messages and final return value of the streaming callable function.
+@available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
+public enum StreamResponse<Message: Decodable, Result: Decodable>: Decodable,
+  StreamResponseProtocol {
+  /// The message yielded by the callable function.
+  case message(Message)
+  /// The final result returned by the callable function.
+  case result(Result)
+
+  private enum CodingKeys: String, CodingKey {
+    case message
+    case result
+  }
+
+  public init(from decoder: any Decoder) throws {
+    do {
+      let container = try decoder
+        .container(keyedBy: Self<Message, Result>.CodingKeys.self)
+      guard let onlyKey = container.allKeys.first, container.allKeys.count == 1 else {
+        throw DecodingError
+          .typeMismatch(
+            Self<Message,
+              Result>.self,
+            DecodingError.Context(
+              codingPath: container.codingPath,
+              debugDescription: "Invalid number of keys found, expected one.",
+              underlyingError: nil
+            )
+          )
+      }
+
+      switch onlyKey {
+      case .message:
+        self = try Self
+          .message(container.decode(Message.self, forKey: .message))
+      case .result:
+        self = try Self
+          .result(container.decode(Result.self, forKey: .result))
+      }
+    } catch {
+      throw FunctionsError(.dataLoss, userInfo: [NSUnderlyingErrorKey: error])
+    }
+  }
+}
+
+@available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
+public extension Callable where Request: Sendable, Response: Sendable {
+  /// Creates a stream that yields responses from the streaming callable function.
+  ///
+  /// The request to the Cloud Functions backend made by this method automatically includes a FCM
+  /// token to identify the app instance. If a user is logged in with Firebase Auth, an auth ID
+  /// token for the user is included. If App Check is integrated, an app check token is included.
+  ///
+  /// Firebase Cloud Messaging sends data to the Firebase backend periodically to collect
+  /// information regarding the app instance. To stop this, see `Messaging.deleteData()`. It
+  /// resumes with a new FCM Token the next time you call this method.
+  ///
+  /// - Important: The final result returned by the callable function is only accessible when
+  ///   using `StreamResponse` as the `Response` generic type.
+  ///
+  /// Example of using `stream` _without_ `StreamResponse`:
+  /// ```swift
+  /// let callable: Callable<MyRequest, MyResponse> = // ...
+  /// let request: MyRequest = // ...
+  /// let stream = try callable.stream(request)
+  /// for try await response in stream {
+  ///   // Process each `MyResponse` message
+  ///   print(response)
+  /// }
+  /// ```
+  ///
+  /// Example of using `stream` _with_ `StreamResponse`:
+  /// ```swift
+  /// let callable: Callable<MyRequest, StreamResponse<MyMessage, MyResult>> = // ...
+  /// let request: MyRequest = // ...
+  /// let stream = try callable.stream(request)
+  /// for try await response in stream {
+  ///   switch response {
+  ///   case .message(let message):
+  ///     // Process each `MyMessage`
+  ///     print(message)
+  ///   case .result(let result):
+  ///     // Process the final `MyResult`
+  ///     print(result)
+  ///   }
+  /// }
+  /// ```
+  ///
+  /// - Parameter data: The `Request` data to pass to the callable function.
+  /// - Throws: A ``FunctionsError`` if the parameter `data` cannot be encoded.
+  /// - Returns: A stream wrapping responses yielded by the streaming callable function or
+  ///   a ``FunctionsError`` if an error occurred.
+  func stream(_ data: Request? = nil) throws -> AsyncThrowingStream<Response, Error> {
+    let encoded: Any
+    do {
+      encoded = try encoder.encode(data)
+    } catch {
+      throw FunctionsError(.invalidArgument, userInfo: [NSUnderlyingErrorKey: error])
+    }
+
+    return AsyncThrowingStream { continuation in
+      Task {
+        do {
+          for try await response in callable.stream(encoded) {
+            do {
+              // This response JSON should only be able to be decoded to an `StreamResponse<_, _>`
+              // instance. If the decoding succeeds and the decoded response conforms to
+              // `StreamResponseProtocol`, we know the `Response` generic argument
+              // is `StreamResponse<_, _>`.
+              let responseJSON = switch response {
+              case .message(let json), .result(let json): json
+              }
+              let response = try decoder.decode(Response.self, from: responseJSON)
+              if response is StreamResponseProtocol {
+                continuation.yield(response)
+              } else {
+                // `Response` is a custom type that matched the decoding logic as the
+                // `StreamResponse<_, _>` type. Only the `StreamResponse<_, _>` type should decode
+                // successfully here to avoid exposing the `result` value in a custom type.
+                throw FunctionsError(.internal)
+              }
+            } catch let error as FunctionsError where error.code == .dataLoss {
+              // `Response` is of type `StreamResponse<_, _>`, but failed to decode. Rethrow.
+              throw error
+            } catch {
+              // `Response` is *not* of type `StreamResponse<_, _>`, and needs to be unboxed and
+              // decoded.
+              guard case let .message(messageJSON) = response else {
+                // Since `Response` is not a `StreamResponse<_, _>`, only messages should be
+                // decoded.
+                continue
+              }
+
+              do {
+                let boxedMessage = try decoder.decode(
+                  StreamResponseMessage.self,
+                  from: messageJSON
+                )
+                continuation.yield(boxedMessage.message)
+              } catch {
+                throw FunctionsError(.dataLoss, userInfo: [NSUnderlyingErrorKey: error])
+              }
+            }
+          }
+        } catch {
+          continuation.finish(throwing: error)
+        }
+        continuation.finish()
+      }
+    }
+  }
+
+  /// A container type for the type-safe decoding of the message object from the generic `Response`
+  /// type.
+  private struct StreamResponseMessage: Decodable {
+    let message: Response
+  }
+}
+
+/// A container type for differentiating between message and result responses.
+enum JSONStreamResponse {
+  case message([String: Any])
+  case result([String: Any])
+}