Swift Concurrency Support

Sources/GraphQL/GraphQL.swift

@@ -226,3 +226,115 @@ public func graphqlSubscribe(
         operationName: operationName
     )
 }
+
+// MARK: Async/Await
+
+#if compiler(>=5.5) && canImport(_Concurrency)
+
+@available(macOS 12, iOS 15, watchOS 8, tvOS 15, *)
+/// This is the primary entry point function for fulfilling GraphQL operations
+/// by parsing, validating, and executing a GraphQL document along side a
+/// GraphQL schema.
+///
+/// More sophisticated GraphQL servers, such as those which persist queries,
+/// may wish to separate the validation and execution phases to a static time
+/// tooling step, and a server runtime step.
+///
+/// - parameter queryStrategy:        The field execution strategy to use for query requests
+/// - parameter mutationStrategy:     The field execution strategy to use for mutation requests
+/// - parameter subscriptionStrategy: The field execution strategy to use for subscription requests
+/// - parameter instrumentation:      The instrumentation implementation to call during the parsing, validating, execution, and field resolution stages.
+/// - parameter schema:               The GraphQL type system to use when validating and executing a query.
+/// - parameter request:              A GraphQL language formatted string representing the requested operation.
+/// - parameter rootValue:            The value provided as the first argument to resolver functions on the top level type (e.g. the query object type).
+/// - parameter contextValue:         A context value provided to all resolver functions functions
+/// - parameter variableValues:       A mapping of variable name to runtime value to use for all variables defined in the `request`.
+/// - parameter operationName:        The name of the operation to use if `request` contains multiple possible operations. Can be omitted if `request` contains only one operation.
+///
+/// - throws: throws GraphQLError if an error occurs while parsing the `request`.
+///
+/// - returns: returns a `Map` dictionary containing the result of the query inside the key `data` and any validation or execution errors inside the key `errors`. The value of `data` might be `null` if, for example, the query is invalid. It's possible to have both `data` and `errors` if an error occurs only in a specific field. If that happens the value of that field will be `null` and there will be an error inside `errors` specifying the reason for the failure and the path of the failed field.
+public func graphql(
+    queryStrategy: QueryFieldExecutionStrategy = SerialFieldExecutionStrategy(),
+    mutationStrategy: MutationFieldExecutionStrategy = SerialFieldExecutionStrategy(),
+    subscriptionStrategy: SubscriptionFieldExecutionStrategy = SerialFieldExecutionStrategy(),
+    instrumentation: Instrumentation = NoOpInstrumentation,
+    schema: GraphQLSchema,
+    request: String,
+    rootValue: Any = (),
+    context: Any = (),
+    eventLoopGroup: EventLoopGroup,
+    variableValues: [String: Map] = [:],
+    operationName: String? = nil
+) async throws -> GraphQLResult {
+    return try await graphql(
+        queryStrategy: queryStrategy,
+        mutationStrategy: mutationStrategy,
+        subscriptionStrategy: subscriptionStrategy,
+        instrumentation: instrumentation,
+        schema: schema,
+        request: request,
+        rootValue: rootValue,
+        context: context,
+        eventLoopGroup: eventLoopGroup,
+        variableValues: variableValues,
+        operationName: operationName
+    ).get()
+}
+
+@available(macOS 12, iOS 15, watchOS 8, tvOS 15, *)
+/// This is the primary entry point function for fulfilling GraphQL subscription
+/// operations by parsing, validating, and executing a GraphQL subscription
+/// document along side a GraphQL schema.
+///
+/// More sophisticated GraphQL servers, such as those which persist queries,
+/// may wish to separate the validation and execution phases to a static time
+/// tooling step, and a server runtime step.
+///
+/// - parameter queryStrategy:        The field execution strategy to use for query requests
+/// - parameter mutationStrategy:     The field execution strategy to use for mutation requests
+/// - parameter subscriptionStrategy: The field execution strategy to use for subscription requests
+/// - parameter instrumentation:      The instrumentation implementation to call during the parsing, validating, execution, and field resolution stages.
+/// - parameter schema:               The GraphQL type system to use when validating and executing a query.
+/// - parameter request:              A GraphQL language formatted string representing the requested operation.
+/// - parameter rootValue:            The value provided as the first argument to resolver functions on the top level type (e.g. the query object type).
+/// - parameter contextValue:         A context value provided to all resolver functions
+/// - parameter variableValues:       A mapping of variable name to runtime value to use for all variables defined in the `request`.
+/// - parameter operationName:        The name of the operation to use if `request` contains multiple possible operations. Can be omitted if `request` contains only one operation.
+///
+/// - throws: throws GraphQLError if an error occurs while parsing the `request`.
+///
+/// - returns: returns a SubscriptionResult containing the subscription observable inside the key `observable` and any validation or execution errors inside the key `errors`. The
+/// value of `observable` might be `null` if, for example, the query is invalid. It's not possible to have both `observable` and `errors`. The observable payloads are
+/// GraphQLResults which contain the result of the query inside the key `data` and any validation or execution errors inside the key `errors`. The value of `data` might be `null`.
+/// It's possible to have both `data` and `errors` if an error occurs only in a specific field. If that happens the value of that field will be `null` and there
+/// will be an error inside `errors` specifying the reason for the failure and the path of the failed field.
+public func graphqlSubscribe(
+    queryStrategy: QueryFieldExecutionStrategy = SerialFieldExecutionStrategy(),
+    mutationStrategy: MutationFieldExecutionStrategy = SerialFieldExecutionStrategy(),
+    subscriptionStrategy: SubscriptionFieldExecutionStrategy = SerialFieldExecutionStrategy(),
+    instrumentation: Instrumentation = NoOpInstrumentation,
+    schema: GraphQLSchema,
+    request: String,
+    rootValue: Any = (),
+    context: Any = (),
+    eventLoopGroup: EventLoopGroup,
+    variableValues: [String: Map] = [:],
+    operationName: String? = nil
+) async throws -> SubscriptionResult {
+    return try await graphqlSubscribe(
+        queryStrategy: queryStrategy,
+        mutationStrategy: mutationStrategy,
+        subscriptionStrategy: subscriptionStrategy,
+        instrumentation: instrumentation,
+        schema: schema,
+        request: request,
+        rootValue: rootValue,
+        context: context,
+        eventLoopGroup: eventLoopGroup,
+        variableValues: variableValues,
+        operationName: operationName
+    ).get()
+}
+
+#endif

Sources/GraphQL/Subscription/EventStream.swift

@@ -6,3 +6,51 @@ open class EventStream<Element> {
         fatalError("This function should be overridden by implementing classes")
     }
 }
+
+#if compiler(>=5.5) && canImport(_Concurrency)
+
+@available(macOS 12, iOS 15, watchOS 8, tvOS 15, *)
+/// Event stream that wraps an `AsyncThrowingStream` from Swift's standard concurrency system.
+public class ConcurrentEventStream<Element>: EventStream<Element> {
+    public let stream: AsyncThrowingStream<Element, Error>
+
+    public init(_ stream: AsyncThrowingStream<Element, Error>) {
+        self.stream = stream
+    }
+
+    /// Performs the closure on each event in the current stream and returns a stream of the results.
+    /// - Parameter closure: The closure to apply to each event in the stream
+    /// - Returns: A stream of the results
+    override open func map<To>(_ closure: @escaping (Element) throws -> To) -> ConcurrentEventStream<To> {
+        let newStream = self.stream.mapStream(closure)
+        return ConcurrentEventStream<To>.init(newStream)
+    }
+}
+
+@available(macOS 12, iOS 15, watchOS 8, tvOS 15, *)
+extension AsyncThrowingStream {
+    func mapStream<To>(_ closure: @escaping (Element) throws -> To) -> AsyncThrowingStream<To, Error> {
+        return AsyncThrowingStream<To, Error> { continuation in
+            Task {
+                for try await event in self {
+                    let newEvent = try closure(event)
+                    continuation.yield(newEvent)
+                }
+            }
+        }
+    }
+
+    func filterStream(_ isIncluded: @escaping (Element) throws -> Bool) -> AsyncThrowingStream<Element, Error> {
+        return AsyncThrowingStream<Element, Error> { continuation in
+            Task {
+                for try await event in self {
+                    if try isIncluded(event) {
+                        continuation.yield(event)
+                    }
+                }
+            }
+        }
+    }
+}
+
+#endif

Tests/GraphQLTests/HelloWorldTests/HelloWorldTests.swift

@@ -62,4 +62,28 @@ class HelloWorldTests : XCTestCase {
 
         XCTAssertEqual(result, expected)
     }
+
+    #if compiler(>=5.5) && canImport(_Concurrency)
+
+    @available(macOS 12, iOS 15, watchOS 8, tvOS 15, *)
+    func testHelloAsync() async throws {
+        let group = MultiThreadedEventLoopGroup(numberOfThreads: System.coreCount)
+
+        defer {
+            XCTAssertNoThrow(try group.syncShutdownGracefully())
+        }
+
+        let query = "{ hello }"
+        let expected = GraphQLResult(data: ["hello": "world"])
+
+        let result = try await graphql(
+            schema: schema,
+            request: query,
+            eventLoopGroup: group
+        )
+
+        XCTAssertEqual(result, expected)
+    }
+
+    #endif
 }

Tests/GraphQLTests/SubscriptionTests/SimplePubSub.swift

@@ -0,0 +1,49 @@
+import GraphQL
+
+
+#if compiler(>=5.5) && canImport(_Concurrency)
+
+@available(macOS 12, iOS 15, watchOS 8, tvOS 15, *)
+/// A very simple publish/subscriber used for testing
+class SimplePubSub<T> {
+    private var subscribers: [Subscriber<T>]
+
+    init() {
+        subscribers = []
+    }
+
+    func emit(event: T) {
+        for subscriber in subscribers {
+            subscriber.callback(event)
+        }
+    }
+
+    func cancel() {
+        for subscriber in subscribers {
+            subscriber.cancel()
+        }
+    }
+
+    func subscribe() -> ConcurrentEventStream<T> {
+        let asyncStream = AsyncThrowingStream<T, Error> { continuation in
+            let subscriber = Subscriber<T>(
+                callback: { newValue in
+                    continuation.yield(newValue)
+                },
+                cancel: {
+                    continuation.finish()
+                }
+            )
+            subscribers.append(subscriber)
+            return
+        }
+        return ConcurrentEventStream<T>.init(asyncStream)
+    }
+}
+
+struct Subscriber<T> {
+    let callback: (T) -> Void
+    let cancel: () -> Void
+}
+
+#endif