[WIP] Streamable functions

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,10 +129,6 @@ 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;
@@ -153,3 +157,29 @@ exports.genStreamError = functionsV2.https.onCall(
     }
   }
 );
+
+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(500);
+  }
+};
+
+exports.genStreamWeather = functionsV2.https.onCall(
+  async (request, response) => {
+    if (request.acceptsStreaming) {
+      for await (const chunk of generateForecast(request.data)) {
+        response.sendChunk({ chunk });
+      }
+    }
+    return "Number of forecasts generated: " + request.data.length;
+  }
+);
+
+// TODO: Maybe a function that returns Void?

FirebaseFunctions/Backend/start.sh

@@ -57,6 +57,7 @@ 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
 
 if [ "$1" != "synchronous" ]; then
   # Wait for the user to tell us to stop the server.

FirebaseFunctions/Sources/Callable+Codable.swift

@@ -159,4 +159,27 @@ public struct Callable<Request: Encodable, Response: Decodable> {
   public func callAsFunction(_ data: Request) async throws -> Response {
     return try await call(data)
   }
+
+  // TODO: Look into handling parameter-less functions.
+  @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
+  public func stream(_ data: Request) -> AsyncThrowingStream<Response, Error> {
+    return AsyncThrowingStream { continuation in
+      Task {
+        do {
+          let encoded = try encoder.encode(data)
+          for try await result in callable.stream(encoded) {
+            if let response = try? decoder.decode([String: Response].self, from: result.data) {
+              continuation.yield(response["chunk"]!)
+            } else if let response = try? decoder.decode(Response.self, from: result.data) {
+              continuation.yield(response)
+            }
+            // TODO: Silently failing. The response cannot be decoded to the given type.
+          }
+        } catch {
+          continuation.finish(throwing: error)
+        }
+        continuation.finish()
+      }
+    }
+  }
 }

FirebaseFunctions/Sources/Functions.swift

@@ -471,6 +471,167 @@ enum FunctionsConstants {
     }
   }
 
+  ///  Function to initialize a streamaing event for an HTTPCallable
+  /// - Parameters:
+  ///   - url: The url of the Callable HTTPS trigger.
+  ///   - data:   Object to be sent in the request.
+  ///   - options: The options with which to customize the Callable HTTPS trigger.
+  ///   - timeout: timeout for  the HTTPSCallableResult request.
+  /// - Returns: HTTPSCallableResult Streaming.
+  @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
+  func stream(at url: URL,
+              withObject data: Any?,
+              options: HTTPSCallableOptions?,
+              timeout: TimeInterval)
+    -> AsyncThrowingStream<HTTPSCallableResult, Error> {
+    AsyncThrowingStream { continuation in
+      // TODO: Vertex prints the curl command. Should this?
+
+      Task {
+        // TODO: This API does not throw. Should the throwing request
+        // setup be in the stream or one level up?
+        let urlRequest: URLRequest
+        do {
+          let context = try await contextProvider.context(options: options)
+          urlRequest = try makeRequestForStreamableContent(
+            url: url,
+            data: data,
+            options: options,
+            timeout: timeout,
+            context: context
+          )
+        } catch {
+          continuation.finish(throwing: error)
+          return
+        }
+
+        let stream: URLSession.AsyncBytes
+        let rawResponse: URLResponse
+        do {
+          // TODO: Look into injecting URLSession for unit tests.
+          (stream, rawResponse) = try await URLSession.shared.bytes(for: urlRequest)
+        } catch {
+          continuation.finish(throwing: error)
+          return
+        }
+
+        // Verify the status code is an HTTP response.
+        guard let response = rawResponse as? HTTPURLResponse else {
+          continuation.finish(
+            throwing: FunctionsError(
+              .internal,
+              userInfo: [NSLocalizedDescriptionKey: "Response was not an HTTP response."]
+            )
+          )
+          return
+        }
+
+        // Verify the status code is a 200.
+        guard response.statusCode == 200 else {
+          continuation.finish(
+            throwing: FunctionsError(
+              .internal,
+              userInfo: [NSLocalizedDescriptionKey: "Response is not a successful 200."]
+            )
+          )
+          return
+        }
+
+        for try await line in stream.lines {
+          if line.hasPrefix("data:") {
+            // We can assume 5 characters since it's utf-8 encoded, removing `data:`.
+            let jsonText = String(line.dropFirst(5))
+            let data: Data
+            do {
+              // TODO: The error potentially thrown here is not a Functions error.
+              data = try jsonData(jsonText: jsonText)
+            } catch {
+              continuation.finish(throwing: error)
+              return
+            }
+
+            // Handle the content and parse it.
+            do {
+              let content = try callableResult(fromResponseData: data)
+              continuation.yield(content)
+            } catch {
+              continuation.finish(throwing: error)
+              return
+            }
+          } else {
+            continuation.finish(
+              throwing: FunctionsError(
+                .internal,
+                userInfo: [NSLocalizedDescriptionKey: "Unexpected format for streamed response."]
+              )
+            )
+          }
+        }
+
+        continuation.finish(throwing: nil)
+      }
+    }
+  }
+
+  private func jsonData(jsonText: String) throws -> Data {
+    guard let data = jsonText.data(using: .utf8) else {
+      throw DecodingError.dataCorrupted(DecodingError.Context(
+        codingPath: [],
+        debugDescription: "Could not parse response as UTF8."
+      ))
+    }
+    return data
+  }
+
+  private func makeRequestForStreamableContent(url: URL,
+                                               data: Any?,
+                                               options: HTTPSCallableOptions?,
+                                               timeout: TimeInterval,
+                                               context: FunctionsContext) throws
+    -> URLRequest {
+    var urlRequest = URLRequest(
+      url: url,
+      cachePolicy: .useProtocolCachePolicy,
+      timeoutInterval: timeout
+    )
+
+    let data = data ?? NSNull()
+    let encoded = try serializer.encode(data)
+    let body = ["data": encoded]
+    let payload = try JSONSerialization.data(withJSONObject: body, options: [.fragmentsAllowed])
+    urlRequest.httpBody = payload
+
+    // Set the headers for starting a streaming session.
+    urlRequest.setValue("application/json", forHTTPHeaderField: "Content-Type")
+    urlRequest.setValue("text/event-stream", forHTTPHeaderField: "Accept")
+    urlRequest.httpMethod = "POST"
+
+    if let authToken = context.authToken {
+      let value = "Bearer \(authToken)"
+      urlRequest.setValue(value, forHTTPHeaderField: "Authorization")
+    }
+
+    if let fcmToken = context.fcmToken {
+      urlRequest.setValue(fcmToken, forHTTPHeaderField: Constants.fcmTokenHeader)
+    }
+
+    if options?.requireLimitedUseAppCheckTokens == true {
+      if let appCheckToken = context.limitedUseAppCheckToken {
+        urlRequest.setValue(
+          appCheckToken,
+          forHTTPHeaderField: Constants.appCheckTokenHeader
+        )
+      }
+    } else if let appCheckToken = context.appCheckToken {
+      urlRequest.setValue(
+        appCheckToken,
+        forHTTPHeaderField: Constants.appCheckTokenHeader
+      )
+    }
+
+    return urlRequest
+  }
+
   private func makeFetcher(url: URL,
                            data: Any?,
                            options: HTTPSCallableOptions?,
@@ -564,8 +725,10 @@ enum FunctionsConstants {
       throw FunctionsError(.internal, userInfo: userInfo)
     }
 
-    // `result` is checked for backwards compatibility:
-    guard let dataJSON = responseJSON["data"] ?? responseJSON["result"] else {
+    // `result` is checked for backwards compatibility,
+    // `message` is checked for StreamableContent:
+    guard let dataJSON = responseJSON["data"] ?? responseJSON["result"] ?? responseJSON["message"]
+    else {
       let userInfo = [NSLocalizedDescriptionKey: "Response is missing data field."]
       throw FunctionsError(.internal, userInfo: userInfo)
     }

FirebaseFunctions/Sources/HTTPSCallable.swift

@@ -39,7 +39,7 @@ open class HTTPSCallable: NSObject {
   // The functions client to use for making calls.
   private let functions: Functions
 
-  private let url: URL
+  let url: URL
 
   private let options: HTTPSCallableOptions?
 
@@ -143,4 +143,9 @@ open class HTTPSCallable: NSObject {
     try await functions
       .callFunction(at: url, withObject: data, options: options, timeout: timeoutInterval)
   }
+
+  @available(macOS 12.0, iOS 15.0, watchOS 8.0, tvOS 15.0, *)
+  func stream(_ data: Any? = nil) -> AsyncThrowingStream<HTTPSCallableResult, Error> {
+    functions.stream(at: url, withObject: data, options: options, timeout: timeoutInterval)
+  }
 }