feat!: allow any AsyncSequence type in subscription

This allows intermediate `map`/`filter` requests to occur before binding the GraphQL subscription to the messenger.

Author: NeedleInAJayStack

Package.resolved

@@ -13,7 +13,7 @@ let package = Package(
     ],
     dependencies: [
         .package(url: "https://github.com/GraphQLSwift/Graphiti.git", from: "3.0.0"),
-        .package(url: "https://github.com/GraphQLSwift/GraphQL.git", from: "4.0.0"),
+        .package(url: "https://github.com/GraphQLSwift/GraphQL.git", from: "4.0.1"),
     ],
     targets: [
         .target(

Package.swift

@@ -1,10 +1,10 @@
 // Contains convenient `connection_init` payloads for users of this package
 
 /// `connection_init` `payload` that is empty
-public struct EmptyInitPayload: Equatable & Codable {}
+public struct EmptyInitPayload: Equatable & Codable & Sendable {}
 
 /// `connection_init` `payload` that includes an `authToken` field
-public struct TokenInitPayload: Equatable & Codable {
+public struct TokenInitPayload: Equatable & Codable & Sendable {
     public let authToken: String
 
     public init(authToken: String) {

Sources/GraphQLWS/InitPayloads.swift

@@ -4,12 +4,17 @@ import GraphQL
 /// Server implements the server-side portion of the protocol, allowing a few callbacks for customization.
 ///
 /// By default, there are no authorization checks
-public class Server<InitPayload: Equatable & Codable>: @unchecked Sendable {
+public class Server<
+    InitPayload: Equatable & Codable & Sendable,
+    SubscriptionSequenceType: AsyncSequence & Sendable
+>: @unchecked Sendable where
+    SubscriptionSequenceType.Element == GraphQLResult
+{
     // We keep this weak because we strongly inject this object into the messenger callback
     weak var messenger: Messenger?
 
     let onExecute: (GraphQLRequest) async throws -> GraphQLResult
-    let onSubscribe: (GraphQLRequest) async throws -> Result<AsyncThrowingStream<GraphQLResult, Error>, GraphQLErrors>
+    let onSubscribe: (GraphQLRequest) async throws -> SubscriptionSequenceType
     var auth: (InitPayload) async throws -> Void
 
     var onExit: () async throws -> Void = {}
@@ -33,7 +38,7 @@ public class Server<InitPayload: Equatable & Codable>: @unchecked Sendable {
     public init(
         messenger: Messenger,
         onExecute: @escaping (GraphQLRequest) async throws -> GraphQLResult,
-        onSubscribe: @escaping (GraphQLRequest) async throws -> Result<AsyncThrowingStream<GraphQLResult, Error>, GraphQLErrors>
+        onSubscribe: @escaping (GraphQLRequest) async throws -> SubscriptionSequenceType
     ) {
         self.messenger = messenger
         self.onExecute = onExecute
@@ -166,16 +171,9 @@ public class Server<InitPayload: Equatable & Codable>: @unchecked Sendable {
         }
 
         if isStreaming {
-            do {
-                let result = try await onSubscribe(graphQLRequest)
-                let stream: AsyncThrowingStream<GraphQLResult, Error>
+            subscriptionTasks[id] = Task {
                 do {
-                    stream = try result.get()
-                } catch {
-                    try await sendError(error, id: id)
-                    return
-                }
-                subscriptionTasks[id] = Task {
+                    let stream = try await onSubscribe(graphQLRequest)
                     for try await event in stream {
                         try Task.checkCancellation()
                         do {
@@ -185,10 +183,11 @@ public class Server<InitPayload: Equatable & Codable>: @unchecked Sendable {
                             throw error
                         }
                     }
-                    try await self.sendComplete(id: id)
+                } catch {
+                    try await sendError(error, id: id)
+                    throw error
                 }
-            } catch {
-                try await sendError(error, id: id)
+                try await self.sendComplete(id: id)
             }
         } else {
             do {

Sources/GraphQLWS/Server.swift

@@ -8,7 +8,7 @@ import XCTest
 class GraphqlWsTests: XCTestCase {
     var clientMessenger: TestMessenger!
     var serverMessenger: TestMessenger!
-    var server: Server<TokenInitPayload>!
+    var server: Server<TokenInitPayload, AsyncThrowingStream<GraphQLResult, Error>>!
     var context: TestContext!
 
     override func setUp() {
@@ -21,7 +21,7 @@ class GraphqlWsTests: XCTestCase {
         let api = TestAPI()
         let context = TestContext()
 
-        server = Server<TokenInitPayload>(
+        server = .init(
             messenger: serverMessenger,
             onExecute: { graphQLRequest in
                 try await api.execute(
@@ -33,7 +33,7 @@ class GraphqlWsTests: XCTestCase {
                 try await api.subscribe(
                     request: graphQLRequest.query,
                     context: context
-                )
+                ).get()
             }
         )
         self.context = context