feat: GraphQL executions are handled in parallel

Subscriptions were already parallelized, but this parallelizes normal queries as well.

Author: NeedleInAJayStack

Sources/GraphQLTransportWS/Server.swift

@@ -25,7 +25,7 @@ where
 
     private var initialized = false
     private var initResult: InitPayloadResult?
-    private var subscriptionTasks = [String: Task<Void, any Error>]()
+    private var executionTasks = [String: Task<Void, any Error>]()
 
     /// Create a new server
     ///
@@ -53,7 +53,7 @@ where
     }
 
     deinit {
-        subscriptionTasks.values.forEach { $0.cancel() }
+        executionTasks.values.forEach { $0.cancel() }
     }
 
     /// Listen and react to the provided async sequence of client messages. This function will block until the stream is completed.
@@ -133,22 +133,26 @@ where
         }
 
         let id = subscribeRequest.id
-        if subscriptionTasks[id] != nil {
-            try await error(.subscriberAlreadyExists(id: id))
-        }
-
         let graphQLRequest = subscribeRequest.payload
 
-        var isStreaming = false
+        let isStreaming: Bool
         do {
             isStreaming = try graphQLRequest.isSubscription()
         } catch {
             try await sendError(error, id: id)
             return
         }
 
-        if isStreaming {
-            subscriptionTasks[id] = Task {
+        guard executionTasks[id] == nil else {
+            try await self.error(.subscriberAlreadyExists(id: id))
+            return
+        }
+        executionTasks[id] = Task {
+            defer {
+                executionTasks.removeValue(forKey: id)
+            }
+
+            if isStreaming {
                 do {
                     let stream = try await onSubscribe(graphQLRequest, initResult)
                     for try await event in stream {
@@ -157,19 +161,16 @@ where
                     }
                 } catch {
                     try await sendError(error, id: id)
-                    subscriptionTasks.removeValue(forKey: id)
-                    throw error
                 }
                 try await self.sendComplete(id: id)
-                subscriptionTasks.removeValue(forKey: id)
-            }
-        } else {
-            do {
-                let result = try await onExecute(graphQLRequest, initResult)
-                try await sendNext(result, id: id)
-                try await sendComplete(id: id)
-            } catch {
-                try await sendError(error, id: id)
+            } else {
+                do {
+                    let result = try await onExecute(graphQLRequest, initResult)
+                    try await sendNext(result, id: id)
+                    try await sendComplete(id: id)
+                } catch {
+                    try await sendError(error, id: id)
+                }
             }
         }
     }
@@ -181,9 +182,9 @@ where
         }
 
         let id = completeRequest.id
-        if let task = subscriptionTasks[id] {
+        if let task = executionTasks[id] {
             task.cancel()
-            subscriptionTasks.removeValue(forKey: id)
+            executionTasks.removeValue(forKey: id)
         }
         try await onOperationComplete(id)
     }