GraphQL executions are handled in parallel

Sources/GraphQLTransportWS/Client.swift

@@ -50,43 +50,52 @@ public actor Client<InitPayload: Equatable & Codable> {
  do {
  response = try decoder.decode(Response.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  switch response.type {
  case .connectionAck:
- guard
- let connectionAckResponse = try? decoder.decode(
+ let connectionAckResponse: ConnectionAckResponse
+ do {
+ connectionAckResponse = try decoder.decode(
  ConnectionAckResponse.self,
  from: message
  )
-else {
- try await error(.invalidResponseFormat(messageType: .connectionAck))
+} catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .connectionAck, error: error))
  return
  }
  try await onConnectionAck(connectionAckResponse, self)
  case .next:
- guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .next))
+ let nextResponse: NextResponse
+ do {
+ nextResponse = try decoder.decode(NextResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .next, error: error))
  return
  }
  try await onNext(nextResponse, self)
  case .error:
- guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .error))
+ let errorResponse: ErrorResponse
+ do {
+ errorResponse = try decoder.decode(ErrorResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .error, error: error))
  return
  }
  try await onError(errorResponse, self)
  case .complete:
- guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
- else {
- try await error(.invalidResponseFormat(messageType: .complete))
+ let completeResponse: CompleteResponse
+ do {
+ completeResponse = try decoder.decode(CompleteResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .complete, error: error))
  return
  }
  try await onComplete(completeResponse, self)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -123,9 +132,4 @@ public actor Client<InitPayload: Equatable & Codable> {
  )
  )
  }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Sources/GraphQLTransportWS/GraphqlTransportWSError.swift

@@ -58,16 +58,16 @@ struct GraphQLTransportWSError: Error {
  )
  }
 
- static func invalidRequestFormat(messageType: RequestMessageType) -> Self {
+ static func invalidRequestFormat(messageType: RequestMessageType, error: Error) -> Self {
  return self.init(
- "Request message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Request message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }
 
- static func invalidResponseFormat(messageType: ResponseMessageType) -> Self {
+ static func invalidResponseFormat(messageType: ResponseMessageType, error: Error) -> Self {
  return self.init(
- "Response message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Response message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }

Sources/GraphQLTransportWS/Messenger.swift

@@ -15,3 +15,10 @@ public protocol Messenger: Sendable {
  /// - code: An error code
  func error(_ message: String, code: Int) async throws
 }
+
+extension Messenger {
+ /// Send an error through the messenger and close the connection
+ func error(_ error: GraphQLTransportWSError) async throws {
+ try await self.error(error.message, code: error.code.rawValue)
+ }
+}

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.
@@ -70,39 +70,44 @@ where
  do {
  request = try decoder.decode(Request.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  // handle incoming message
  switch request.type {
  case .connectionInit:
- guard
- let connectionInitRequest = try? decoder.decode(
+ let connectionInitRequest: ConnectionInitRequest<InitPayload>
+ do {
+ connectionInitRequest = try decoder.decode(
  ConnectionInitRequest<InitPayload>.self,
  from: message
  )
-else {
- try await error(.invalidRequestFormat(messageType: .connectionInit))
+} catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .connectionInit, error: error))
  return
  }
  try await onConnectionInit(connectionInitRequest, messenger)
  case .subscribe:
- guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .subscribe))
+ let subscribeRequest: SubscribeRequest
+ do {
+ subscribeRequest = try decoder.decode(SubscribeRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .subscribe, error: error))
  return
  }
  try await onSubscribe(subscribeRequest)
  case .complete:
- guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .complete))
+ let completeRequest: CompleteRequest
+ do {
+ completeRequest = try decoder.decode(CompleteRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .complete, error: error))
  return
  }
  try await onOperationComplete(completeRequest)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -111,14 +116,14 @@ where
  _: Messenger
  ) async throws {
  guard !initialized else {
- try await error(.tooManyInitializations())
+ try await messenger.error(.tooManyInitializations())
  return
  }
 
  do {
  initResult = try await onInit(connectionInitRequest.payload)
  } catch {
- try await self.error(.forbidden())
+ try await messenger.error(.forbidden())
  return
  }
  initialized = true
@@ -128,62 +133,66 @@ where
 
  private func onSubscribe(_ subscribeRequest: SubscribeRequest) async throws {
  guard initialized, let initResult else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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 messenger.error(.subscriberAlreadyExists(id: id))
+ return
+ }
+ executionTasks[id] = Task {
+ defer {
+ executionTasks.removeValue(forKey: id)
+ }
+
+ if isStreaming {
+ let stream: SubscriptionSequenceType
  do {
- let stream = try await onSubscribe(graphQLRequest, initResult)
- for try await event in stream {
- try Task.checkCancellation()
- try await self.sendNext(event, id: id)
- }
+ stream = try await onSubscribe(graphQLRequest, initResult)
  } catch {
  try await sendError(error, id: id)
- subscriptionTasks.removeValue(forKey: id)
- throw error
+ return
+ }
+ for try await event in stream {
+ try await self.sendNext(event, id: id)
+ }
+ executionTasks.removeValue(forKey: id)
+ } else {
+ let result: GraphQLResult
+ do {
+ result = try await onExecute(graphQLRequest, initResult)
+ } catch {
+ try await sendError(error, id: id)
+ return
  }
- 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)
  }
+ try await sendComplete(id: id)
  }
  }
 
  private func onOperationComplete(_ completeRequest: CompleteRequest) async throws {
  guard initialized else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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)
  }
@@ -238,14 +247,4 @@ where
  private func sendError(_ error: Error, id: String) async throws {
  try await sendError([error], id: id)
  }
-
- /// Send an `error` response through the messenger
- private func sendError(_ errorMessage: String, id: String) async throws {
- try await sendError(GraphQLError(message: errorMessage), id: id)
- }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift

@@ -268,6 +268,62 @@ struct GraphqlTransportWSTests {
  )
  }
 
+ /// Tests malformed requests include decoder details in the transport error
+ @Test func malformedRequestIncludesDecodingDetails() async throws {
+ let api = TestAPI()
+ let context = TestContext()
+ let server = Server<TokenInitPayload, Void, AsyncThrowingStream<GraphQLResult, Error>>(
+ messenger: serverMessenger,
+ onInit: { _ in },
+ onExecute: { graphQLRequest, _ in
+ try await api.execute(
+ request: graphQLRequest.query,
+ context: context
+ )
+ },
+ onSubscribe: { graphQLRequest, _ in
+ try await api.subscribe(
+ request: graphQLRequest.query,
+ context: context
+ ).get()
+ }
+ )
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"complete"}"#.utf8))
+ continuation.finish()
+
+ try await server.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in serverMessenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Request message doesn't match 'complete' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
+ /// Tests malformed responses include decoder details in the transport error
+ @Test func malformedResponseIncludesDecodingDetails() async throws {
+ let messenger = TestMessenger()
+ let client = Client<TokenInitPayload>(messenger: messenger)
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"next"}"#.utf8))
+ continuation.finish()
+
+ try await client.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in messenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Response message doesn't match 'next' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
  enum TestError: Error {
  case couldBeAnything
  }

Sources/GraphQLTransportWS/Client.swift

@@ -50,43 +50,52 @@ public actor Client<InitPayload: Equatable & Codable> {
  do {
  response = try decoder.decode(Response.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  switch response.type {
  case .connectionAck:
- guard
- let connectionAckResponse = try? decoder.decode(
+ let connectionAckResponse: ConnectionAckResponse
+ do {
+ connectionAckResponse = try decoder.decode(
  ConnectionAckResponse.self,
  from: message
  )
-else {
- try await error(.invalidResponseFormat(messageType: .connectionAck))
+} catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .connectionAck, error: error))
  return
  }
  try await onConnectionAck(connectionAckResponse, self)
  case .next:
- guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .next))
+ let nextResponse: NextResponse
+ do {
+ nextResponse = try decoder.decode(NextResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .next, error: error))
  return
  }
  try await onNext(nextResponse, self)
  case .error:
- guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .error))
+ let errorResponse: ErrorResponse
+ do {
+ errorResponse = try decoder.decode(ErrorResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .error, error: error))
  return
  }
  try await onError(errorResponse, self)
  case .complete:
- guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
- else {
- try await error(.invalidResponseFormat(messageType: .complete))
+ let completeResponse: CompleteResponse
+ do {
+ completeResponse = try decoder.decode(CompleteResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .complete, error: error))
  return
  }
  try await onComplete(completeResponse, self)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -123,9 +132,4 @@ public actor Client<InitPayload: Equatable & Codable> {
  )
  )
  }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Sources/GraphQLTransportWS/GraphqlTransportWSError.swift

@@ -58,16 +58,16 @@ struct GraphQLTransportWSError: Error {
  )
  }
 
- static func invalidRequestFormat(messageType: RequestMessageType) -> Self {
+ static func invalidRequestFormat(messageType: RequestMessageType, error: Error) -> Self {
  return self.init(
- "Request message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Request message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }
 
- static func invalidResponseFormat(messageType: ResponseMessageType) -> Self {
+ static func invalidResponseFormat(messageType: ResponseMessageType, error: Error) -> Self {
  return self.init(
- "Response message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Response message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }

Sources/GraphQLTransportWS/Messenger.swift

@@ -15,3 +15,10 @@ public protocol Messenger: Sendable {
  /// - code: An error code
  func error(_ message: String, code: Int) async throws
 }
+
+extension Messenger {
+ /// Send an error through the messenger and close the connection
+ func error(_ error: GraphQLTransportWSError) async throws {
+ try await self.error(error.message, code: error.code.rawValue)
+ }
+}

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.
@@ -70,39 +70,44 @@ where
  do {
  request = try decoder.decode(Request.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  // handle incoming message
  switch request.type {
  case .connectionInit:
- guard
- let connectionInitRequest = try? decoder.decode(
+ let connectionInitRequest: ConnectionInitRequest<InitPayload>
+ do {
+ connectionInitRequest = try decoder.decode(
  ConnectionInitRequest<InitPayload>.self,
  from: message
  )
-else {
- try await error(.invalidRequestFormat(messageType: .connectionInit))
+} catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .connectionInit, error: error))
  return
  }
  try await onConnectionInit(connectionInitRequest, messenger)
  case .subscribe:
- guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .subscribe))
+ let subscribeRequest: SubscribeRequest
+ do {
+ subscribeRequest = try decoder.decode(SubscribeRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .subscribe, error: error))
  return
  }
  try await onSubscribe(subscribeRequest)
  case .complete:
- guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .complete))
+ let completeRequest: CompleteRequest
+ do {
+ completeRequest = try decoder.decode(CompleteRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .complete, error: error))
  return
  }
  try await onOperationComplete(completeRequest)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -111,14 +116,14 @@ where
  _: Messenger
  ) async throws {
  guard !initialized else {
- try await error(.tooManyInitializations())
+ try await messenger.error(.tooManyInitializations())
  return
  }
 
  do {
  initResult = try await onInit(connectionInitRequest.payload)
  } catch {
- try await self.error(.forbidden())
+ try await messenger.error(.forbidden())
  return
  }
  initialized = true
@@ -128,62 +133,66 @@ where
 
  private func onSubscribe(_ subscribeRequest: SubscribeRequest) async throws {
  guard initialized, let initResult else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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 messenger.error(.subscriberAlreadyExists(id: id))
+ return
+ }
+ executionTasks[id] = Task {
+ defer {
+ executionTasks.removeValue(forKey: id)
+ }
+
+ if isStreaming {
+ let stream: SubscriptionSequenceType
  do {
- let stream = try await onSubscribe(graphQLRequest, initResult)
- for try await event in stream {
- try Task.checkCancellation()
- try await self.sendNext(event, id: id)
- }
+ stream = try await onSubscribe(graphQLRequest, initResult)
  } catch {
  try await sendError(error, id: id)
- subscriptionTasks.removeValue(forKey: id)
- throw error
+ return
+ }
+ for try await event in stream {
+ try await self.sendNext(event, id: id)
+ }
+ executionTasks.removeValue(forKey: id)
+ } else {
+ let result: GraphQLResult
+ do {
+ result = try await onExecute(graphQLRequest, initResult)
+ } catch {
+ try await sendError(error, id: id)
+ return
  }
- 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)
  }
+ try await sendComplete(id: id)
  }
  }
 
  private func onOperationComplete(_ completeRequest: CompleteRequest) async throws {
  guard initialized else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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)
  }
@@ -238,14 +247,4 @@ where
  private func sendError(_ error: Error, id: String) async throws {
  try await sendError([error], id: id)
  }
-
- /// Send an `error` response through the messenger
- private func sendError(_ errorMessage: String, id: String) async throws {
- try await sendError(GraphQLError(message: errorMessage), id: id)
- }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift

@@ -268,6 +268,62 @@ struct GraphqlTransportWSTests {
  )
  }
 
+ /// Tests malformed requests include decoder details in the transport error
+ @Test func malformedRequestIncludesDecodingDetails() async throws {
+ let api = TestAPI()
+ let context = TestContext()
+ let server = Server<TokenInitPayload, Void, AsyncThrowingStream<GraphQLResult, Error>>(
+ messenger: serverMessenger,
+ onInit: { _ in },
+ onExecute: { graphQLRequest, _ in
+ try await api.execute(
+ request: graphQLRequest.query,
+ context: context
+ )
+ },
+ onSubscribe: { graphQLRequest, _ in
+ try await api.subscribe(
+ request: graphQLRequest.query,
+ context: context
+ ).get()
+ }
+ )
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"complete"}"#.utf8))
+ continuation.finish()
+
+ try await server.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in serverMessenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Request message doesn't match 'complete' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
+ /// Tests malformed responses include decoder details in the transport error
+ @Test func malformedResponseIncludesDecodingDetails() async throws {
+ let messenger = TestMessenger()
+ let client = Client<TokenInitPayload>(messenger: messenger)
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"next"}"#.utf8))
+ continuation.finish()
+
+ try await client.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in messenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Response message doesn't match 'next' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
  enum TestError: Error {
  case couldBeAnything
  }

Sources/GraphQLTransportWS/Client.swift

@@ -50,43 +50,52 @@ public actor Client<InitPayload: Equatable & Codable> {
  do {
  response = try decoder.decode(Response.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  switch response.type {
  case .connectionAck:
- guard
- let connectionAckResponse = try? decoder.decode(
+ let connectionAckResponse: ConnectionAckResponse
+ do {
+ connectionAckResponse = try decoder.decode(
  ConnectionAckResponse.self,
  from: message
  )
-else {
- try await error(.invalidResponseFormat(messageType: .connectionAck))
+} catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .connectionAck, error: error))
  return
  }
  try await onConnectionAck(connectionAckResponse, self)
  case .next:
- guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .next))
+ let nextResponse: NextResponse
+ do {
+ nextResponse = try decoder.decode(NextResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .next, error: error))
  return
  }
  try await onNext(nextResponse, self)
  case .error:
- guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .error))
+ let errorResponse: ErrorResponse
+ do {
+ errorResponse = try decoder.decode(ErrorResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .error, error: error))
  return
  }
  try await onError(errorResponse, self)
  case .complete:
- guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
- else {
- try await error(.invalidResponseFormat(messageType: .complete))
+ let completeResponse: CompleteResponse
+ do {
+ completeResponse = try decoder.decode(CompleteResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .complete, error: error))
  return
  }
  try await onComplete(completeResponse, self)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -123,9 +132,4 @@ public actor Client<InitPayload: Equatable & Codable> {
  )
  )
  }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Sources/GraphQLTransportWS/GraphqlTransportWSError.swift

@@ -58,16 +58,16 @@ struct GraphQLTransportWSError: Error {
  )
  }
 
- static func invalidRequestFormat(messageType: RequestMessageType) -> Self {
+ static func invalidRequestFormat(messageType: RequestMessageType, error: Error) -> Self {
  return self.init(
- "Request message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Request message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }
 
- static func invalidResponseFormat(messageType: ResponseMessageType) -> Self {
+ static func invalidResponseFormat(messageType: ResponseMessageType, error: Error) -> Self {
  return self.init(
- "Response message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Response message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }

Sources/GraphQLTransportWS/Messenger.swift

@@ -15,3 +15,10 @@ public protocol Messenger: Sendable {
  /// - code: An error code
  func error(_ message: String, code: Int) async throws
 }
+
+extension Messenger {
+ /// Send an error through the messenger and close the connection
+ func error(_ error: GraphQLTransportWSError) async throws {
+ try await self.error(error.message, code: error.code.rawValue)
+ }
+}

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.
@@ -70,39 +70,44 @@ where
  do {
  request = try decoder.decode(Request.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  // handle incoming message
  switch request.type {
  case .connectionInit:
- guard
- let connectionInitRequest = try? decoder.decode(
+ let connectionInitRequest: ConnectionInitRequest<InitPayload>
+ do {
+ connectionInitRequest = try decoder.decode(
  ConnectionInitRequest<InitPayload>.self,
  from: message
  )
-else {
- try await error(.invalidRequestFormat(messageType: .connectionInit))
+} catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .connectionInit, error: error))
  return
  }
  try await onConnectionInit(connectionInitRequest, messenger)
  case .subscribe:
- guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .subscribe))
+ let subscribeRequest: SubscribeRequest
+ do {
+ subscribeRequest = try decoder.decode(SubscribeRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .subscribe, error: error))
  return
  }
  try await onSubscribe(subscribeRequest)
  case .complete:
- guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .complete))
+ let completeRequest: CompleteRequest
+ do {
+ completeRequest = try decoder.decode(CompleteRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .complete, error: error))
  return
  }
  try await onOperationComplete(completeRequest)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -111,14 +116,14 @@ where
  _: Messenger
  ) async throws {
  guard !initialized else {
- try await error(.tooManyInitializations())
+ try await messenger.error(.tooManyInitializations())
  return
  }
 
  do {
  initResult = try await onInit(connectionInitRequest.payload)
  } catch {
- try await self.error(.forbidden())
+ try await messenger.error(.forbidden())
  return
  }
  initialized = true
@@ -128,62 +133,66 @@ where
 
  private func onSubscribe(_ subscribeRequest: SubscribeRequest) async throws {
  guard initialized, let initResult else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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 messenger.error(.subscriberAlreadyExists(id: id))
+ return
+ }
+ executionTasks[id] = Task {
+ defer {
+ executionTasks.removeValue(forKey: id)
+ }
+
+ if isStreaming {
+ let stream: SubscriptionSequenceType
  do {
- let stream = try await onSubscribe(graphQLRequest, initResult)
- for try await event in stream {
- try Task.checkCancellation()
- try await self.sendNext(event, id: id)
- }
+ stream = try await onSubscribe(graphQLRequest, initResult)
  } catch {
  try await sendError(error, id: id)
- subscriptionTasks.removeValue(forKey: id)
- throw error
+ return
+ }
+ for try await event in stream {
+ try await self.sendNext(event, id: id)
+ }
+ executionTasks.removeValue(forKey: id)
+ } else {
+ let result: GraphQLResult
+ do {
+ result = try await onExecute(graphQLRequest, initResult)
+ } catch {
+ try await sendError(error, id: id)
+ return
  }
- 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)
  }
+ try await sendComplete(id: id)
  }
  }
 
  private func onOperationComplete(_ completeRequest: CompleteRequest) async throws {
  guard initialized else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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)
  }
@@ -238,14 +247,4 @@ where
  private func sendError(_ error: Error, id: String) async throws {
  try await sendError([error], id: id)
  }
-
- /// Send an `error` response through the messenger
- private func sendError(_ errorMessage: String, id: String) async throws {
- try await sendError(GraphQLError(message: errorMessage), id: id)
- }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift

@@ -268,6 +268,62 @@ struct GraphqlTransportWSTests {
  )
  }
 
+ /// Tests malformed requests include decoder details in the transport error
+ @Test func malformedRequestIncludesDecodingDetails() async throws {
+ let api = TestAPI()
+ let context = TestContext()
+ let server = Server<TokenInitPayload, Void, AsyncThrowingStream<GraphQLResult, Error>>(
+ messenger: serverMessenger,
+ onInit: { _ in },
+ onExecute: { graphQLRequest, _ in
+ try await api.execute(
+ request: graphQLRequest.query,
+ context: context
+ )
+ },
+ onSubscribe: { graphQLRequest, _ in
+ try await api.subscribe(
+ request: graphQLRequest.query,
+ context: context
+ ).get()
+ }
+ )
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"complete"}"#.utf8))
+ continuation.finish()
+
+ try await server.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in serverMessenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Request message doesn't match 'complete' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
+ /// Tests malformed responses include decoder details in the transport error
+ @Test func malformedResponseIncludesDecodingDetails() async throws {
+ let messenger = TestMessenger()
+ let client = Client<TokenInitPayload>(messenger: messenger)
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"next"}"#.utf8))
+ continuation.finish()
+
+ try await client.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in messenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Response message doesn't match 'next' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
  enum TestError: Error {
  case couldBeAnything
  }

Sources/GraphQLTransportWS/Client.swift

@@ -50,43 +50,52 @@ public actor Client<InitPayload: Equatable & Codable> {
  do {
  response = try decoder.decode(Response.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  switch response.type {
  case .connectionAck:
- guard
- let connectionAckResponse = try? decoder.decode(
+ let connectionAckResponse: ConnectionAckResponse
+ do {
+ connectionAckResponse = try decoder.decode(
  ConnectionAckResponse.self,
  from: message
  )
-else {
- try await error(.invalidResponseFormat(messageType: .connectionAck))
+} catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .connectionAck, error: error))
  return
  }
  try await onConnectionAck(connectionAckResponse, self)
  case .next:
- guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .next))
+ let nextResponse: NextResponse
+ do {
+ nextResponse = try decoder.decode(NextResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .next, error: error))
  return
  }
  try await onNext(nextResponse, self)
  case .error:
- guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .error))
+ let errorResponse: ErrorResponse
+ do {
+ errorResponse = try decoder.decode(ErrorResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .error, error: error))
  return
  }
  try await onError(errorResponse, self)
  case .complete:
- guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
- else {
- try await error(.invalidResponseFormat(messageType: .complete))
+ let completeResponse: CompleteResponse
+ do {
+ completeResponse = try decoder.decode(CompleteResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .complete, error: error))
  return
  }
  try await onComplete(completeResponse, self)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -123,9 +132,4 @@ public actor Client<InitPayload: Equatable & Codable> {
  )
  )
  }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Sources/GraphQLTransportWS/GraphqlTransportWSError.swift

@@ -58,16 +58,16 @@ struct GraphQLTransportWSError: Error {
  )
  }
 
- static func invalidRequestFormat(messageType: RequestMessageType) -> Self {
+ static func invalidRequestFormat(messageType: RequestMessageType, error: Error) -> Self {
  return self.init(
- "Request message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Request message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }
 
- static func invalidResponseFormat(messageType: ResponseMessageType) -> Self {
+ static func invalidResponseFormat(messageType: ResponseMessageType, error: Error) -> Self {
  return self.init(
- "Response message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Response message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }

Sources/GraphQLTransportWS/Messenger.swift

@@ -15,3 +15,10 @@ public protocol Messenger: Sendable {
  /// - code: An error code
  func error(_ message: String, code: Int) async throws
 }
+
+extension Messenger {
+ /// Send an error through the messenger and close the connection
+ func error(_ error: GraphQLTransportWSError) async throws {
+ try await self.error(error.message, code: error.code.rawValue)
+ }
+}

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.
@@ -70,39 +70,44 @@ where
  do {
  request = try decoder.decode(Request.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  // handle incoming message
  switch request.type {
  case .connectionInit:
- guard
- let connectionInitRequest = try? decoder.decode(
+ let connectionInitRequest: ConnectionInitRequest<InitPayload>
+ do {
+ connectionInitRequest = try decoder.decode(
  ConnectionInitRequest<InitPayload>.self,
  from: message
  )
-else {
- try await error(.invalidRequestFormat(messageType: .connectionInit))
+} catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .connectionInit, error: error))
  return
  }
  try await onConnectionInit(connectionInitRequest, messenger)
  case .subscribe:
- guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .subscribe))
+ let subscribeRequest: SubscribeRequest
+ do {
+ subscribeRequest = try decoder.decode(SubscribeRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .subscribe, error: error))
  return
  }
  try await onSubscribe(subscribeRequest)
  case .complete:
- guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .complete))
+ let completeRequest: CompleteRequest
+ do {
+ completeRequest = try decoder.decode(CompleteRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .complete, error: error))
  return
  }
  try await onOperationComplete(completeRequest)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -111,14 +116,14 @@ where
  _: Messenger
  ) async throws {
  guard !initialized else {
- try await error(.tooManyInitializations())
+ try await messenger.error(.tooManyInitializations())
  return
  }
 
  do {
  initResult = try await onInit(connectionInitRequest.payload)
  } catch {
- try await self.error(.forbidden())
+ try await messenger.error(.forbidden())
  return
  }
  initialized = true
@@ -128,62 +133,66 @@ where
 
  private func onSubscribe(_ subscribeRequest: SubscribeRequest) async throws {
  guard initialized, let initResult else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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 messenger.error(.subscriberAlreadyExists(id: id))
+ return
+ }
+ executionTasks[id] = Task {
+ defer {
+ executionTasks.removeValue(forKey: id)
+ }
+
+ if isStreaming {
+ let stream: SubscriptionSequenceType
  do {
- let stream = try await onSubscribe(graphQLRequest, initResult)
- for try await event in stream {
- try Task.checkCancellation()
- try await self.sendNext(event, id: id)
- }
+ stream = try await onSubscribe(graphQLRequest, initResult)
  } catch {
  try await sendError(error, id: id)
- subscriptionTasks.removeValue(forKey: id)
- throw error
+ return
+ }
+ for try await event in stream {
+ try await self.sendNext(event, id: id)
+ }
+ executionTasks.removeValue(forKey: id)
+ } else {
+ let result: GraphQLResult
+ do {
+ result = try await onExecute(graphQLRequest, initResult)
+ } catch {
+ try await sendError(error, id: id)
+ return
  }
- 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)
  }
+ try await sendComplete(id: id)
  }
  }
 
  private func onOperationComplete(_ completeRequest: CompleteRequest) async throws {
  guard initialized else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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)
  }
@@ -238,14 +247,4 @@ where
  private func sendError(_ error: Error, id: String) async throws {
  try await sendError([error], id: id)
  }
-
- /// Send an `error` response through the messenger
- private func sendError(_ errorMessage: String, id: String) async throws {
- try await sendError(GraphQLError(message: errorMessage), id: id)
- }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift

@@ -268,6 +268,62 @@ struct GraphqlTransportWSTests {
  )
  }
 
+ /// Tests malformed requests include decoder details in the transport error
+ @Test func malformedRequestIncludesDecodingDetails() async throws {
+ let api = TestAPI()
+ let context = TestContext()
+ let server = Server<TokenInitPayload, Void, AsyncThrowingStream<GraphQLResult, Error>>(
+ messenger: serverMessenger,
+ onInit: { _ in },
+ onExecute: { graphQLRequest, _ in
+ try await api.execute(
+ request: graphQLRequest.query,
+ context: context
+ )
+ },
+ onSubscribe: { graphQLRequest, _ in
+ try await api.subscribe(
+ request: graphQLRequest.query,
+ context: context
+ ).get()
+ }
+ )
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"complete"}"#.utf8))
+ continuation.finish()
+
+ try await server.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in serverMessenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Request message doesn't match 'complete' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
+ /// Tests malformed responses include decoder details in the transport error
+ @Test func malformedResponseIncludesDecodingDetails() async throws {
+ let messenger = TestMessenger()
+ let client = Client<TokenInitPayload>(messenger: messenger)
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"next"}"#.utf8))
+ continuation.finish()
+
+ try await client.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in messenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Response message doesn't match 'next' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
  enum TestError: Error {
  case couldBeAnything
  }

Sources/GraphQLTransportWS/Client.swift

@@ -50,43 +50,52 @@ public actor Client<InitPayload: Equatable & Codable> {
  do {
  response = try decoder.decode(Response.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  switch response.type {
  case .connectionAck:
- guard
- let connectionAckResponse = try? decoder.decode(
+ let connectionAckResponse: ConnectionAckResponse
+ do {
+ connectionAckResponse = try decoder.decode(
  ConnectionAckResponse.self,
  from: message
  )
-else {
- try await error(.invalidResponseFormat(messageType: .connectionAck))
+} catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .connectionAck, error: error))
  return
  }
  try await onConnectionAck(connectionAckResponse, self)
  case .next:
- guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .next))
+ let nextResponse: NextResponse
+ do {
+ nextResponse = try decoder.decode(NextResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .next, error: error))
  return
  }
  try await onNext(nextResponse, self)
  case .error:
- guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .error))
+ let errorResponse: ErrorResponse
+ do {
+ errorResponse = try decoder.decode(ErrorResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .error, error: error))
  return
  }
  try await onError(errorResponse, self)
  case .complete:
- guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
- else {
- try await error(.invalidResponseFormat(messageType: .complete))
+ let completeResponse: CompleteResponse
+ do {
+ completeResponse = try decoder.decode(CompleteResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .complete, error: error))
  return
  }
  try await onComplete(completeResponse, self)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -123,9 +132,4 @@ public actor Client<InitPayload: Equatable & Codable> {
  )
  )
  }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Sources/GraphQLTransportWS/GraphqlTransportWSError.swift

@@ -58,16 +58,16 @@ struct GraphQLTransportWSError: Error {
  )
  }
 
- static func invalidRequestFormat(messageType: RequestMessageType) -> Self {
+ static func invalidRequestFormat(messageType: RequestMessageType, error: Error) -> Self {
  return self.init(
- "Request message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Request message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }
 
- static func invalidResponseFormat(messageType: ResponseMessageType) -> Self {
+ static func invalidResponseFormat(messageType: ResponseMessageType, error: Error) -> Self {
  return self.init(
- "Response message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Response message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }

Sources/GraphQLTransportWS/Messenger.swift

@@ -15,3 +15,10 @@ public protocol Messenger: Sendable {
  /// - code: An error code
  func error(_ message: String, code: Int) async throws
 }
+
+extension Messenger {
+ /// Send an error through the messenger and close the connection
+ func error(_ error: GraphQLTransportWSError) async throws {
+ try await self.error(error.message, code: error.code.rawValue)
+ }
+}

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.
@@ -70,39 +70,44 @@ where
  do {
  request = try decoder.decode(Request.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  // handle incoming message
  switch request.type {
  case .connectionInit:
- guard
- let connectionInitRequest = try? decoder.decode(
+ let connectionInitRequest: ConnectionInitRequest<InitPayload>
+ do {
+ connectionInitRequest = try decoder.decode(
  ConnectionInitRequest<InitPayload>.self,
  from: message
  )
-else {
- try await error(.invalidRequestFormat(messageType: .connectionInit))
+} catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .connectionInit, error: error))
  return
  }
  try await onConnectionInit(connectionInitRequest, messenger)
  case .subscribe:
- guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .subscribe))
+ let subscribeRequest: SubscribeRequest
+ do {
+ subscribeRequest = try decoder.decode(SubscribeRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .subscribe, error: error))
  return
  }
  try await onSubscribe(subscribeRequest)
  case .complete:
- guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .complete))
+ let completeRequest: CompleteRequest
+ do {
+ completeRequest = try decoder.decode(CompleteRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .complete, error: error))
  return
  }
  try await onOperationComplete(completeRequest)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -111,14 +116,14 @@ where
  _: Messenger
  ) async throws {
  guard !initialized else {
- try await error(.tooManyInitializations())
+ try await messenger.error(.tooManyInitializations())
  return
  }
 
  do {
  initResult = try await onInit(connectionInitRequest.payload)
  } catch {
- try await self.error(.forbidden())
+ try await messenger.error(.forbidden())
  return
  }
  initialized = true
@@ -128,62 +133,66 @@ where
 
  private func onSubscribe(_ subscribeRequest: SubscribeRequest) async throws {
  guard initialized, let initResult else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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 messenger.error(.subscriberAlreadyExists(id: id))
+ return
+ }
+ executionTasks[id] = Task {
+ defer {
+ executionTasks.removeValue(forKey: id)
+ }
+
+ if isStreaming {
+ let stream: SubscriptionSequenceType
  do {
- let stream = try await onSubscribe(graphQLRequest, initResult)
- for try await event in stream {
- try Task.checkCancellation()
- try await self.sendNext(event, id: id)
- }
+ stream = try await onSubscribe(graphQLRequest, initResult)
  } catch {
  try await sendError(error, id: id)
- subscriptionTasks.removeValue(forKey: id)
- throw error
+ return
+ }
+ for try await event in stream {
+ try await self.sendNext(event, id: id)
+ }
+ executionTasks.removeValue(forKey: id)
+ } else {
+ let result: GraphQLResult
+ do {
+ result = try await onExecute(graphQLRequest, initResult)
+ } catch {
+ try await sendError(error, id: id)
+ return
  }
- 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)
  }
+ try await sendComplete(id: id)
  }
  }
 
  private func onOperationComplete(_ completeRequest: CompleteRequest) async throws {
  guard initialized else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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)
  }
@@ -238,14 +247,4 @@ where
  private func sendError(_ error: Error, id: String) async throws {
  try await sendError([error], id: id)
  }
-
- /// Send an `error` response through the messenger
- private func sendError(_ errorMessage: String, id: String) async throws {
- try await sendError(GraphQLError(message: errorMessage), id: id)
- }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift

@@ -268,6 +268,62 @@ struct GraphqlTransportWSTests {
  )
  }
 
+ /// Tests malformed requests include decoder details in the transport error
+ @Test func malformedRequestIncludesDecodingDetails() async throws {
+ let api = TestAPI()
+ let context = TestContext()
+ let server = Server<TokenInitPayload, Void, AsyncThrowingStream<GraphQLResult, Error>>(
+ messenger: serverMessenger,
+ onInit: { _ in },
+ onExecute: { graphQLRequest, _ in
+ try await api.execute(
+ request: graphQLRequest.query,
+ context: context
+ )
+ },
+ onSubscribe: { graphQLRequest, _ in
+ try await api.subscribe(
+ request: graphQLRequest.query,
+ context: context
+ ).get()
+ }
+ )
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"complete"}"#.utf8))
+ continuation.finish()
+
+ try await server.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in serverMessenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Request message doesn't match 'complete' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
+ /// Tests malformed responses include decoder details in the transport error
+ @Test func malformedResponseIncludesDecodingDetails() async throws {
+ let messenger = TestMessenger()
+ let client = Client<TokenInitPayload>(messenger: messenger)
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"next"}"#.utf8))
+ continuation.finish()
+
+ try await client.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in messenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Response message doesn't match 'next' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
  enum TestError: Error {
  case couldBeAnything
  }

Sources/GraphQLTransportWS/Client.swift

@@ -50,43 +50,52 @@ public actor Client<InitPayload: Equatable & Codable> {
  do {
  response = try decoder.decode(Response.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  switch response.type {
  case .connectionAck:
- guard
- let connectionAckResponse = try? decoder.decode(
+ let connectionAckResponse: ConnectionAckResponse
+ do {
+ connectionAckResponse = try decoder.decode(
  ConnectionAckResponse.self,
  from: message
  )
-else {
- try await error(.invalidResponseFormat(messageType: .connectionAck))
+} catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .connectionAck, error: error))
  return
  }
  try await onConnectionAck(connectionAckResponse, self)
  case .next:
- guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .next))
+ let nextResponse: NextResponse
+ do {
+ nextResponse = try decoder.decode(NextResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .next, error: error))
  return
  }
  try await onNext(nextResponse, self)
  case .error:
- guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .error))
+ let errorResponse: ErrorResponse
+ do {
+ errorResponse = try decoder.decode(ErrorResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .error, error: error))
  return
  }
  try await onError(errorResponse, self)
  case .complete:
- guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
- else {
- try await error(.invalidResponseFormat(messageType: .complete))
+ let completeResponse: CompleteResponse
+ do {
+ completeResponse = try decoder.decode(CompleteResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .complete, error: error))
  return
  }
  try await onComplete(completeResponse, self)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -123,9 +132,4 @@ public actor Client<InitPayload: Equatable & Codable> {
  )
  )
  }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Sources/GraphQLTransportWS/GraphqlTransportWSError.swift

@@ -58,16 +58,16 @@ struct GraphQLTransportWSError: Error {
  )
  }
 
- static func invalidRequestFormat(messageType: RequestMessageType) -> Self {
+ static func invalidRequestFormat(messageType: RequestMessageType, error: Error) -> Self {
  return self.init(
- "Request message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Request message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }
 
- static func invalidResponseFormat(messageType: ResponseMessageType) -> Self {
+ static func invalidResponseFormat(messageType: ResponseMessageType, error: Error) -> Self {
  return self.init(
- "Response message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Response message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }

Sources/GraphQLTransportWS/Messenger.swift

@@ -15,3 +15,10 @@ public protocol Messenger: Sendable {
  /// - code: An error code
  func error(_ message: String, code: Int) async throws
 }
+
+extension Messenger {
+ /// Send an error through the messenger and close the connection
+ func error(_ error: GraphQLTransportWSError) async throws {
+ try await self.error(error.message, code: error.code.rawValue)
+ }
+}

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.
@@ -70,39 +70,44 @@ where
  do {
  request = try decoder.decode(Request.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  // handle incoming message
  switch request.type {
  case .connectionInit:
- guard
- let connectionInitRequest = try? decoder.decode(
+ let connectionInitRequest: ConnectionInitRequest<InitPayload>
+ do {
+ connectionInitRequest = try decoder.decode(
  ConnectionInitRequest<InitPayload>.self,
  from: message
  )
-else {
- try await error(.invalidRequestFormat(messageType: .connectionInit))
+} catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .connectionInit, error: error))
  return
  }
  try await onConnectionInit(connectionInitRequest, messenger)
  case .subscribe:
- guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .subscribe))
+ let subscribeRequest: SubscribeRequest
+ do {
+ subscribeRequest = try decoder.decode(SubscribeRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .subscribe, error: error))
  return
  }
  try await onSubscribe(subscribeRequest)
  case .complete:
- guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .complete))
+ let completeRequest: CompleteRequest
+ do {
+ completeRequest = try decoder.decode(CompleteRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .complete, error: error))
  return
  }
  try await onOperationComplete(completeRequest)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -111,14 +116,14 @@ where
  _: Messenger
  ) async throws {
  guard !initialized else {
- try await error(.tooManyInitializations())
+ try await messenger.error(.tooManyInitializations())
  return
  }
 
  do {
  initResult = try await onInit(connectionInitRequest.payload)
  } catch {
- try await self.error(.forbidden())
+ try await messenger.error(.forbidden())
  return
  }
  initialized = true
@@ -128,62 +133,66 @@ where
 
  private func onSubscribe(_ subscribeRequest: SubscribeRequest) async throws {
  guard initialized, let initResult else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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 messenger.error(.subscriberAlreadyExists(id: id))
+ return
+ }
+ executionTasks[id] = Task {
+ defer {
+ executionTasks.removeValue(forKey: id)
+ }
+
+ if isStreaming {
+ let stream: SubscriptionSequenceType
  do {
- let stream = try await onSubscribe(graphQLRequest, initResult)
- for try await event in stream {
- try Task.checkCancellation()
- try await self.sendNext(event, id: id)
- }
+ stream = try await onSubscribe(graphQLRequest, initResult)
  } catch {
  try await sendError(error, id: id)
- subscriptionTasks.removeValue(forKey: id)
- throw error
+ return
+ }
+ for try await event in stream {
+ try await self.sendNext(event, id: id)
+ }
+ executionTasks.removeValue(forKey: id)
+ } else {
+ let result: GraphQLResult
+ do {
+ result = try await onExecute(graphQLRequest, initResult)
+ } catch {
+ try await sendError(error, id: id)
+ return
  }
- 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)
  }
+ try await sendComplete(id: id)
  }
  }
 
  private func onOperationComplete(_ completeRequest: CompleteRequest) async throws {
  guard initialized else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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)
  }
@@ -238,14 +247,4 @@ where
  private func sendError(_ error: Error, id: String) async throws {
  try await sendError([error], id: id)
  }
-
- /// Send an `error` response through the messenger
- private func sendError(_ errorMessage: String, id: String) async throws {
- try await sendError(GraphQLError(message: errorMessage), id: id)
- }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift

@@ -268,6 +268,62 @@ struct GraphqlTransportWSTests {
  )
  }
 
+ /// Tests malformed requests include decoder details in the transport error
+ @Test func malformedRequestIncludesDecodingDetails() async throws {
+ let api = TestAPI()
+ let context = TestContext()
+ let server = Server<TokenInitPayload, Void, AsyncThrowingStream<GraphQLResult, Error>>(
+ messenger: serverMessenger,
+ onInit: { _ in },
+ onExecute: { graphQLRequest, _ in
+ try await api.execute(
+ request: graphQLRequest.query,
+ context: context
+ )
+ },
+ onSubscribe: { graphQLRequest, _ in
+ try await api.subscribe(
+ request: graphQLRequest.query,
+ context: context
+ ).get()
+ }
+ )
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"complete"}"#.utf8))
+ continuation.finish()
+
+ try await server.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in serverMessenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Request message doesn't match 'complete' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
+ /// Tests malformed responses include decoder details in the transport error
+ @Test func malformedResponseIncludesDecodingDetails() async throws {
+ let messenger = TestMessenger()
+ let client = Client<TokenInitPayload>(messenger: messenger)
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"next"}"#.utf8))
+ continuation.finish()
+
+ try await client.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in messenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Response message doesn't match 'next' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
  enum TestError: Error {
  case couldBeAnything
  }

Sources/GraphQLTransportWS/Client.swift

@@ -50,43 +50,52 @@ public actor Client<InitPayload: Equatable & Codable> {
  do {
  response = try decoder.decode(Response.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  switch response.type {
  case .connectionAck:
- guard
- let connectionAckResponse = try? decoder.decode(
+ let connectionAckResponse: ConnectionAckResponse
+ do {
+ connectionAckResponse = try decoder.decode(
  ConnectionAckResponse.self,
  from: message
  )
-else {
- try await error(.invalidResponseFormat(messageType: .connectionAck))
+} catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .connectionAck, error: error))
  return
  }
  try await onConnectionAck(connectionAckResponse, self)
  case .next:
- guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .next))
+ let nextResponse: NextResponse
+ do {
+ nextResponse = try decoder.decode(NextResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .next, error: error))
  return
  }
  try await onNext(nextResponse, self)
  case .error:
- guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .error))
+ let errorResponse: ErrorResponse
+ do {
+ errorResponse = try decoder.decode(ErrorResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .error, error: error))
  return
  }
  try await onError(errorResponse, self)
  case .complete:
- guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
- else {
- try await error(.invalidResponseFormat(messageType: .complete))
+ let completeResponse: CompleteResponse
+ do {
+ completeResponse = try decoder.decode(CompleteResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .complete, error: error))
  return
  }
  try await onComplete(completeResponse, self)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -123,9 +132,4 @@ public actor Client<InitPayload: Equatable & Codable> {
  )
  )
  }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Sources/GraphQLTransportWS/GraphqlTransportWSError.swift

@@ -58,16 +58,16 @@ struct GraphQLTransportWSError: Error {
  )
  }
 
- static func invalidRequestFormat(messageType: RequestMessageType) -> Self {
+ static func invalidRequestFormat(messageType: RequestMessageType, error: Error) -> Self {
  return self.init(
- "Request message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Request message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }
 
- static func invalidResponseFormat(messageType: ResponseMessageType) -> Self {
+ static func invalidResponseFormat(messageType: ResponseMessageType, error: Error) -> Self {
  return self.init(
- "Response message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Response message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }

Sources/GraphQLTransportWS/Messenger.swift

@@ -15,3 +15,10 @@ public protocol Messenger: Sendable {
  /// - code: An error code
  func error(_ message: String, code: Int) async throws
 }
+
+extension Messenger {
+ /// Send an error through the messenger and close the connection
+ func error(_ error: GraphQLTransportWSError) async throws {
+ try await self.error(error.message, code: error.code.rawValue)
+ }
+}

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.
@@ -70,39 +70,44 @@ where
  do {
  request = try decoder.decode(Request.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  // handle incoming message
  switch request.type {
  case .connectionInit:
- guard
- let connectionInitRequest = try? decoder.decode(
+ let connectionInitRequest: ConnectionInitRequest<InitPayload>
+ do {
+ connectionInitRequest = try decoder.decode(
  ConnectionInitRequest<InitPayload>.self,
  from: message
  )
-else {
- try await error(.invalidRequestFormat(messageType: .connectionInit))
+} catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .connectionInit, error: error))
  return
  }
  try await onConnectionInit(connectionInitRequest, messenger)
  case .subscribe:
- guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .subscribe))
+ let subscribeRequest: SubscribeRequest
+ do {
+ subscribeRequest = try decoder.decode(SubscribeRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .subscribe, error: error))
  return
  }
  try await onSubscribe(subscribeRequest)
  case .complete:
- guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .complete))
+ let completeRequest: CompleteRequest
+ do {
+ completeRequest = try decoder.decode(CompleteRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .complete, error: error))
  return
  }
  try await onOperationComplete(completeRequest)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -111,14 +116,14 @@ where
  _: Messenger
  ) async throws {
  guard !initialized else {
- try await error(.tooManyInitializations())
+ try await messenger.error(.tooManyInitializations())
  return
  }
 
  do {
  initResult = try await onInit(connectionInitRequest.payload)
  } catch {
- try await self.error(.forbidden())
+ try await messenger.error(.forbidden())
  return
  }
  initialized = true
@@ -128,62 +133,66 @@ where
 
  private func onSubscribe(_ subscribeRequest: SubscribeRequest) async throws {
  guard initialized, let initResult else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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 messenger.error(.subscriberAlreadyExists(id: id))
+ return
+ }
+ executionTasks[id] = Task {
+ defer {
+ executionTasks.removeValue(forKey: id)
+ }
+
+ if isStreaming {
+ let stream: SubscriptionSequenceType
  do {
- let stream = try await onSubscribe(graphQLRequest, initResult)
- for try await event in stream {
- try Task.checkCancellation()
- try await self.sendNext(event, id: id)
- }
+ stream = try await onSubscribe(graphQLRequest, initResult)
  } catch {
  try await sendError(error, id: id)
- subscriptionTasks.removeValue(forKey: id)
- throw error
+ return
+ }
+ for try await event in stream {
+ try await self.sendNext(event, id: id)
+ }
+ executionTasks.removeValue(forKey: id)
+ } else {
+ let result: GraphQLResult
+ do {
+ result = try await onExecute(graphQLRequest, initResult)
+ } catch {
+ try await sendError(error, id: id)
+ return
  }
- 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)
  }
+ try await sendComplete(id: id)
  }
  }
 
  private func onOperationComplete(_ completeRequest: CompleteRequest) async throws {
  guard initialized else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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)
  }
@@ -238,14 +247,4 @@ where
  private func sendError(_ error: Error, id: String) async throws {
  try await sendError([error], id: id)
  }
-
- /// Send an `error` response through the messenger
- private func sendError(_ errorMessage: String, id: String) async throws {
- try await sendError(GraphQLError(message: errorMessage), id: id)
- }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift

@@ -268,6 +268,62 @@ struct GraphqlTransportWSTests {
  )
  }
 
+ /// Tests malformed requests include decoder details in the transport error
+ @Test func malformedRequestIncludesDecodingDetails() async throws {
+ let api = TestAPI()
+ let context = TestContext()
+ let server = Server<TokenInitPayload, Void, AsyncThrowingStream<GraphQLResult, Error>>(
+ messenger: serverMessenger,
+ onInit: { _ in },
+ onExecute: { graphQLRequest, _ in
+ try await api.execute(
+ request: graphQLRequest.query,
+ context: context
+ )
+ },
+ onSubscribe: { graphQLRequest, _ in
+ try await api.subscribe(
+ request: graphQLRequest.query,
+ context: context
+ ).get()
+ }
+ )
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"complete"}"#.utf8))
+ continuation.finish()
+
+ try await server.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in serverMessenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Request message doesn't match 'complete' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
+ /// Tests malformed responses include decoder details in the transport error
+ @Test func malformedResponseIncludesDecodingDetails() async throws {
+ let messenger = TestMessenger()
+ let client = Client<TokenInitPayload>(messenger: messenger)
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"next"}"#.utf8))
+ continuation.finish()
+
+ try await client.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in messenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Response message doesn't match 'next' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
  enum TestError: Error {
  case couldBeAnything
  }

Sources/GraphQLTransportWS/Client.swift

@@ -50,43 +50,52 @@ public actor Client<InitPayload: Equatable & Codable> {
  do {
  response = try decoder.decode(Response.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  switch response.type {
  case .connectionAck:
- guard
- let connectionAckResponse = try? decoder.decode(
+ let connectionAckResponse: ConnectionAckResponse
+ do {
+ connectionAckResponse = try decoder.decode(
  ConnectionAckResponse.self,
  from: message
  )
-else {
- try await error(.invalidResponseFormat(messageType: .connectionAck))
+} catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .connectionAck, error: error))
  return
  }
  try await onConnectionAck(connectionAckResponse, self)
  case .next:
- guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .next))
+ let nextResponse: NextResponse
+ do {
+ nextResponse = try decoder.decode(NextResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .next, error: error))
  return
  }
  try await onNext(nextResponse, self)
  case .error:
- guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
- try await error(.invalidResponseFormat(messageType: .error))
+ let errorResponse: ErrorResponse
+ do {
+ errorResponse = try decoder.decode(ErrorResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .error, error: error))
  return
  }
  try await onError(errorResponse, self)
  case .complete:
- guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
- else {
- try await error(.invalidResponseFormat(messageType: .complete))
+ let completeResponse: CompleteResponse
+ do {
+ completeResponse = try decoder.decode(CompleteResponse.self, from: message)
+ } catch {
+ try await messenger.error(.invalidResponseFormat(messageType: .complete, error: error))
  return
  }
  try await onComplete(completeResponse, self)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -123,9 +132,4 @@ public actor Client<InitPayload: Equatable & Codable> {
  )
  )
  }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Sources/GraphQLTransportWS/GraphqlTransportWSError.swift

@@ -58,16 +58,16 @@ struct GraphQLTransportWSError: Error {
  )
  }
 
- static func invalidRequestFormat(messageType: RequestMessageType) -> Self {
+ static func invalidRequestFormat(messageType: RequestMessageType, error: Error) -> Self {
  return self.init(
- "Request message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Request message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }
 
- static func invalidResponseFormat(messageType: ResponseMessageType) -> Self {
+ static func invalidResponseFormat(messageType: ResponseMessageType, error: Error) -> Self {
  return self.init(
- "Response message doesn't match '\(messageType.type.rawValue)' JSON format",
+ "Response message doesn't match '\(messageType.type.rawValue)' JSON format: \(error)",
  code: .miscellaneous
  )
  }

Sources/GraphQLTransportWS/Messenger.swift

@@ -15,3 +15,10 @@ public protocol Messenger: Sendable {
  /// - code: An error code
  func error(_ message: String, code: Int) async throws
 }
+
+extension Messenger {
+ /// Send an error through the messenger and close the connection
+ func error(_ error: GraphQLTransportWSError) async throws {
+ try await self.error(error.message, code: error.code.rawValue)
+ }
+}

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.
@@ -70,39 +70,44 @@ where
  do {
  request = try decoder.decode(Request.self, from: message)
  } catch {
- try await self.error(.noType())
+ try await messenger.error(.noType())
  return
  }
 
  // handle incoming message
  switch request.type {
  case .connectionInit:
- guard
- let connectionInitRequest = try? decoder.decode(
+ let connectionInitRequest: ConnectionInitRequest<InitPayload>
+ do {
+ connectionInitRequest = try decoder.decode(
  ConnectionInitRequest<InitPayload>.self,
  from: message
  )
-else {
- try await error(.invalidRequestFormat(messageType: .connectionInit))
+} catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .connectionInit, error: error))
  return
  }
  try await onConnectionInit(connectionInitRequest, messenger)
  case .subscribe:
- guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .subscribe))
+ let subscribeRequest: SubscribeRequest
+ do {
+ subscribeRequest = try decoder.decode(SubscribeRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .subscribe, error: error))
  return
  }
  try await onSubscribe(subscribeRequest)
  case .complete:
- guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
- else {
- try await error(.invalidRequestFormat(messageType: .complete))
+ let completeRequest: CompleteRequest
+ do {
+ completeRequest = try decoder.decode(CompleteRequest.self, from: message)
+ } catch {
+ try await messenger.error(.invalidRequestFormat(messageType: .complete, error: error))
  return
  }
  try await onOperationComplete(completeRequest)
  default:
- try await error(.invalidType())
+ try await messenger.error(.invalidType())
  }
  }
 
@@ -111,14 +116,14 @@ where
  _: Messenger
  ) async throws {
  guard !initialized else {
- try await error(.tooManyInitializations())
+ try await messenger.error(.tooManyInitializations())
  return
  }
 
  do {
  initResult = try await onInit(connectionInitRequest.payload)
  } catch {
- try await self.error(.forbidden())
+ try await messenger.error(.forbidden())
  return
  }
  initialized = true
@@ -128,62 +133,66 @@ where
 
  private func onSubscribe(_ subscribeRequest: SubscribeRequest) async throws {
  guard initialized, let initResult else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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 messenger.error(.subscriberAlreadyExists(id: id))
+ return
+ }
+ executionTasks[id] = Task {
+ defer {
+ executionTasks.removeValue(forKey: id)
+ }
+
+ if isStreaming {
+ let stream: SubscriptionSequenceType
  do {
- let stream = try await onSubscribe(graphQLRequest, initResult)
- for try await event in stream {
- try Task.checkCancellation()
- try await self.sendNext(event, id: id)
- }
+ stream = try await onSubscribe(graphQLRequest, initResult)
  } catch {
  try await sendError(error, id: id)
- subscriptionTasks.removeValue(forKey: id)
- throw error
+ return
+ }
+ for try await event in stream {
+ try await self.sendNext(event, id: id)
+ }
+ executionTasks.removeValue(forKey: id)
+ } else {
+ let result: GraphQLResult
+ do {
+ result = try await onExecute(graphQLRequest, initResult)
+ } catch {
+ try await sendError(error, id: id)
+ return
  }
- 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)
  }
+ try await sendComplete(id: id)
  }
  }
 
  private func onOperationComplete(_ completeRequest: CompleteRequest) async throws {
  guard initialized else {
- try await error(.notInitialized())
+ try await messenger.error(.notInitialized())
  return
  }
 
  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)
  }
@@ -238,14 +247,4 @@ where
  private func sendError(_ error: Error, id: String) async throws {
  try await sendError([error], id: id)
  }
-
- /// Send an `error` response through the messenger
- private func sendError(_ errorMessage: String, id: String) async throws {
- try await sendError(GraphQLError(message: errorMessage), id: id)
- }
-
- /// Send an error through the messenger and close the connection
- private func error(_ error: GraphQLTransportWSError) async throws {
- try await messenger.error(error.message, code: error.code.rawValue)
- }
 }

Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift

@@ -268,6 +268,62 @@ struct GraphqlTransportWSTests {
  )
  }
 
+ /// Tests malformed requests include decoder details in the transport error
+ @Test func malformedRequestIncludesDecodingDetails() async throws {
+ let api = TestAPI()
+ let context = TestContext()
+ let server = Server<TokenInitPayload, Void, AsyncThrowingStream<GraphQLResult, Error>>(
+ messenger: serverMessenger,
+ onInit: { _ in },
+ onExecute: { graphQLRequest, _ in
+ try await api.execute(
+ request: graphQLRequest.query,
+ context: context
+ )
+ },
+ onSubscribe: { graphQLRequest, _ in
+ try await api.subscribe(
+ request: graphQLRequest.query,
+ context: context
+ ).get()
+ }
+ )
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"complete"}"#.utf8))
+ continuation.finish()
+
+ try await server.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in serverMessenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Request message doesn't match 'complete' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
+ /// Tests malformed responses include decoder details in the transport error
+ @Test func malformedResponseIncludesDecodingDetails() async throws {
+ let messenger = TestMessenger()
+ let client = Client<TokenInitPayload>(messenger: messenger)
+ let (incoming, continuation) = AsyncThrowingStream<Data, any Error>.makeStream()
+
+ continuation.yield(Data(#"{"type":"next"}"#.utf8))
+ continuation.finish()
+
+ try await client.listen(to: incoming)
+
+ let error = await #expect(throws: TestMessengerError.self) {
+ for try await _ in messenger.stream {}
+ }
+ #expect(error?.code == 4400)
+ #expect(error?.message.contains("Response message doesn't match 'next' JSON format") == true)
+ #expect(error?.message.contains("keyNotFound") == true)
+ #expect(error?.message.contains(#""id""#) == true)
+ }
+
  enum TestError: Error {
  case couldBeAnything
  }