feat!: Removes EventStream, replacing with AsyncThrowingStream

Author: NeedleInAJayStack

MIGRATION.md

@@ -16,6 +16,10 @@ The documentation here will be very helpful in the conversion: https://www.swift
 
 This was changed to `ConcurrentFieldExecutionStrategy`, and takes no parameters.
 
+### EventStream removal
+
+The `EventStream` abstraction used to provide pre-concurrency subscription support has been removed. This means that `graphqlSubscribe(...).stream` will now be an `AsyncThrowingStream<GraphQLResult, Error>` type, instead of an `EventStream` type, and that downcasting to `ConcurrentEventStream` is no longer necessary.
+
 ## 2 to 3
 
 ### TypeReference removal

README.md

@@ -75,15 +75,14 @@ let schema = try GraphQLSchema(
                     return eventResult
                 },
                 subscribe: { _, _, _, _, _ in               // Defines how to construct the event stream
-                    let asyncStream = AsyncThrowingStream<String, Error> { continuation in
+                    return AsyncThrowingStream<String, Error> { continuation in
                         let timer = Timer.scheduledTimer(
                             withTimeInterval: 3,
                             repeats: true,
                         ) {
                             continuation.yield("world")     // Emits "world" every 3 seconds
                         }
                     }
-                    return ConcurrentEventStream<String>(asyncStream)
                 }
             )
         ]
@@ -97,8 +96,6 @@ To execute a subscription use the `graphqlSubscribe` function:
 let subscriptionResult = try await graphqlSubscribe(
     schema: schema,
 )
-// Must downcast from EventStream to concrete type to use in 'for await' loop below
-let concurrentStream = subscriptionResult.stream! as! ConcurrentEventStream
 for try await result in concurrentStream.stream {
     print(result)
 }
@@ -110,9 +107,6 @@ The code above will print the following JSON every 3 seconds:
 { "hello": "world" }
 ```
 
-The example above assumes that your environment has access to Swift Concurrency. If that is not the case, try using
-[GraphQLRxSwift](https://github.com/GraphQLSwift/GraphQLRxSwift)
-
 ## Encoding Results
 
 If you encode a `GraphQLResult` with an ordinary `JSONEncoder`, there are no guarantees that the field order will match the query,

Sources/GraphQL/GraphQL.swift

@@ -44,19 +44,18 @@ public struct GraphQLResult: Equatable, Codable, Sendable, CustomStringConvertib
 
 /// SubscriptionResult wraps the observable and error data returned by the subscribe request.
 public struct SubscriptionResult {
-    public let stream: SubscriptionEventStream?
+    public let stream: AsyncThrowingStream<GraphQLResult, Error>?
     public let errors: [GraphQLError]
 
-    public init(stream: SubscriptionEventStream? = nil, errors: [GraphQLError] = []) {
+    public init(
+        stream: AsyncThrowingStream<GraphQLResult, Error>? = nil,
+        errors: [GraphQLError] = []
+    ) {
         self.stream = stream
         self.errors = errors
     }
 }
 
-/// SubscriptionObservable represents an event stream of fully resolved GraphQL subscription
-/// results. Subscribers can be added to this stream.
-public typealias SubscriptionEventStream = EventStream<GraphQLResult>
-
 /// This is the primary entry point function for fulfilling GraphQL operations
 /// by parsing, validating, and executing a GraphQL document along side a
 /// GraphQL schema.

Sources/GraphQL/Subscription/EventStream.swift

@@ -46,25 +46,42 @@ func subscribe(
     )
 
     if let sourceStream = sourceResult.stream {
-        let subscriptionStream = sourceStream.map { eventPayload -> GraphQLResult in
-            // For each payload yielded from a subscription, map it over the normal
-            // GraphQL `execute` function, with `payload` as the rootValue.
-            // This implements the "MapSourceToResponseEvent" algorithm described in
-            // the GraphQL specification. The `execute` function provides the
-            // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
-            // "ExecuteQuery" algorithm, for which `execute` is also used.
-            try await execute(
-                queryStrategy: queryStrategy,
-                mutationStrategy: mutationStrategy,
-                subscriptionStrategy: subscriptionStrategy,
-                instrumentation: instrumentation,
-                schema: schema,
-                documentAST: documentAST,
-                rootValue: eventPayload,
-                context: context,
-                variableValues: variableValues,
-                operationName: operationName
-            )
+        // We must create a new AsyncSequence because AsyncSequence.map requires a concrete type
+        // (which we cannot know),
+        // and we need the result to be a concrete type.
+        let subscriptionStream = AsyncThrowingStream<GraphQLResult, Error> { continuation in
+            let task = Task {
+                do {
+                    for try await eventPayload in sourceStream {
+                        // For each payload yielded from a subscription, map it over the normal
+                        // GraphQL `execute` function, with `payload` as the rootValue.
+                        // This implements the "MapSourceToResponseEvent" algorithm described in
+                        // the GraphQL specification. The `execute` function provides the
+                        // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
+                        // "ExecuteQuery" algorithm, for which `execute` is also used.
+                        let newEvent = try await execute(
+                            queryStrategy: queryStrategy,
+                            mutationStrategy: mutationStrategy,
+                            subscriptionStrategy: subscriptionStrategy,
+                            instrumentation: instrumentation,
+                            schema: schema,
+                            documentAST: documentAST,
+                            rootValue: eventPayload,
+                            context: context,
+                            variableValues: variableValues,
+                            operationName: operationName
+                        )
+                        continuation.yield(newEvent)
+                    }
+                    continuation.finish()
+                } catch {
+                    continuation.finish(throwing: error)
+                }
+            }
+
+            continuation.onTermination = { @Sendable reason in
+                task.cancel()
+            }
         }
         return SubscriptionResult(stream: subscriptionStream, errors: sourceResult.errors)
     } else {
@@ -151,7 +168,7 @@ func createSourceEventStream(
 }
 
 func executeSubscription(
-    context: ExecutionContext,
+    context: ExecutionContext
 ) async throws -> SourceEventStreamResult {
     // Get the first node
     let type = try getOperationRootType(schema: context.schema, operation: context.operation)
@@ -245,7 +262,7 @@ func executeSubscription(
         return SourceEventStreamResult(errors: context.errors)
     } else if let error = resolved as? GraphQLError {
         return SourceEventStreamResult(errors: [error])
-    } else if let stream = resolved as? EventStream<Any> {
+    } else if let stream = resolved as? any AsyncSequence {
         return SourceEventStreamResult(stream: stream)
     } else if resolved == nil {
         return SourceEventStreamResult(errors: [
@@ -255,7 +272,7 @@ func executeSubscription(
         let resolvedObj = resolved as AnyObject
         return SourceEventStreamResult(errors: [
             GraphQLError(
-                message: "Subscription field resolver must return EventStream<Any>. Received: '\(resolvedObj)'"
+                message: "Subscription field resolver must return an AsyncSequence. Received: '\(resolvedObj)'"
             ),
         ])
     }
@@ -266,10 +283,10 @@ func executeSubscription(
 // checking. Normal resolvers for subscription fields should handle type casting, same as resolvers
 // for query fields.
 struct SourceEventStreamResult {
-    public let stream: EventStream<Any>?
+    public let stream: (any AsyncSequence)?
     public let errors: [GraphQLError]
 
-    public init(stream: EventStream<Any>? = nil, errors: [GraphQLError] = []) {
+    public init(stream: (any AsyncSequence)? = nil, errors: [GraphQLError] = []) {
         self.stream = stream
         self.errors = errors
     }

Sources/GraphQL/Subscription/Subscribe.swift

@@ -20,8 +20,8 @@ class SimplePubSub<T> {
         }
     }
 
-    func subscribe() -> ConcurrentEventStream<T> {
-        let asyncStream = AsyncThrowingStream<T, Error> { continuation in
+    func subscribe() -> AsyncThrowingStream<T, Error> {
+        return AsyncThrowingStream<T, Error> { continuation in
             let subscriber = Subscriber<T>(
                 callback: { newValue in
                     continuation.yield(newValue)
@@ -32,7 +32,6 @@ class SimplePubSub<T> {
             )
             subscribers.append(subscriber)
         }
-        return ConcurrentEventStream<T>(asyncStream)
     }
 }
 

Tests/GraphQLTests/SubscriptionTests/SimplePubSub.swift

@@ -129,15 +129,14 @@ class EmailDb {
             },
             subscribe: { _, args, _, _ throws -> Any? in
                 let priority = args["priority"].int ?? 0
-                let filtered = self.publisher.subscribe().stream
-                    .filterStream { emailAny throws in
-                        if let email = emailAny as? Email {
-                            return email.priority >= priority
-                        } else {
-                            return true
-                        }
+                let filtered = self.publisher.subscribe().filter { emailAny throws in
+                    if let email = emailAny as? Email {
+                        return email.priority >= priority
+                    } else {
+                        return true
                     }
-                return ConcurrentEventStream<Any>(filtered)
+                }
+                return filtered
             }
         )
     }
@@ -146,7 +145,7 @@ class EmailDb {
     func subscription(
         query: String,
         variableValues: [String: Map] = [:]
-    ) async throws -> SubscriptionEventStream {
+    ) async throws -> AsyncThrowingStream<GraphQLResult, Error> {
         return try await createSubscription(
             schema: defaultSchema(),
             query: query,
@@ -186,7 +185,7 @@ func createSubscription(
     schema: GraphQLSchema,
     query: String,
     variableValues: [String: Map] = [:]
-) async throws -> SubscriptionEventStream {
+) async throws -> AsyncThrowingStream<GraphQLResult, Error> {
     let result = try await graphqlSubscribe(
         queryStrategy: SerialFieldExecutionStrategy(),
         mutationStrategy: SerialFieldExecutionStrategy(),