Skip to content

Commit d0b7113

Browse files
NeedleInAJayStackNeedleInAJayStack
committed
feat!: Adds Data sequence support to Server and Client
It also changes the error pathway to close immediately instead of sending a formatted message.
1 parent e381e88 commit d0b7113

4 files changed

Lines changed: 139 additions & 140 deletions

File tree

‎Sources/GraphQLTransportWS/Client.swift‎

Lines changed: 53 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -35,65 +35,72 @@ public actor Client<InitPayload: Equatable & Codable> {
3535
self.onError = onError
3636
self.onComplete = onComplete
3737
}
38+
39+
/// Listen and react to the provided async sequence of server messages. This function will block until the stream is completed.
40+
/// - Parameter incoming: The server message sequence that the client should react to.
41+
public func listen<A: AsyncSequence & Sendable>(to incoming: A) async throws
42+
where A.Element == Data {
43+
for try await message in incoming {
44+
try await respond(to: message)
45+
}
46+
}
3847

3948
/// Listen and react to the provided async sequence of server messages. This function will block until the stream is completed.
4049
/// - Parameter incoming: The server message sequence that the client should react to.
50+
@available(*, deprecated, message: "Use `Data` sequence instead.")
4151
public func listen<A: AsyncSequence & Sendable>(to incoming: A) async throws
4252
where A.Element == String {
43-
for try await message in incoming {
44-
// Detect and ignore error responses.
45-
if message.starts(with: "44") {
46-
// TODO: Determine what to do with returned error messages
53+
for try await stringMessage in incoming {
54+
guard let message = stringMessage.data(using: .utf8) else {
55+
try await self.error(.invalidEncoding())
4756
return
4857
}
58+
try await respond(to: message)
59+
}
60+
}
61+
62+
private func respond(to message: Data) async throws {
63+
let response: Response
64+
do {
65+
response = try decoder.decode(Response.self, from: message)
66+
} catch {
67+
try await self.error(.noType())
68+
return
69+
}
4970

50-
guard let json = message.data(using: .utf8) else {
51-
try await error(.invalidEncoding())
71+
switch response.type {
72+
case .connectionAck:
73+
guard
74+
let connectionAckResponse = try? decoder.decode(
75+
ConnectionAckResponse.self,
76+
from: message
77+
)
78+
else {
79+
try await error(.invalidResponseFormat(messageType: .connectionAck))
5280
return
5381
}
54-
55-
let response: Response
56-
do {
57-
response = try decoder.decode(Response.self, from: json)
58-
} catch {
59-
try await self.error(.noType())
82+
try await onConnectionAck(connectionAckResponse, self)
83+
case .next:
84+
guard let nextResponse = try? decoder.decode(NextResponse.self, from: message) else {
85+
try await error(.invalidResponseFormat(messageType: .next))
6086
return
6187
}
62-
63-
switch response.type {
64-
case .connectionAck:
65-
guard
66-
let connectionAckResponse = try? decoder.decode(
67-
ConnectionAckResponse.self,
68-
from: json
69-
)
70-
else {
71-
try await error(.invalidResponseFormat(messageType: .connectionAck))
72-
return
73-
}
74-
try await onConnectionAck(connectionAckResponse, self)
75-
case .next:
76-
guard let nextResponse = try? decoder.decode(NextResponse.self, from: json) else {
77-
try await error(.invalidResponseFormat(messageType: .next))
78-
return
79-
}
80-
try await onNext(nextResponse, self)
81-
case .error:
82-
guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: json) else {
83-
try await error(.invalidResponseFormat(messageType: .error))
84-
return
85-
}
86-
try await onError(errorResponse, self)
87-
case .complete:
88-
guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: json)
89-
else {
90-
try await error(.invalidResponseFormat(messageType: .complete))
91-
return
92-
}
93-
try await onComplete(completeResponse, self)
94-
default:
95-
try await error(.invalidType())
88+
try await onNext(nextResponse, self)
89+
case .error:
90+
guard let errorResponse = try? decoder.decode(ErrorResponse.self, from: message) else {
91+
try await error(.invalidResponseFormat(messageType: .error))
92+
return
93+
}
94+
try await onError(errorResponse, self)
95+
case .complete:
96+
guard let completeResponse = try? decoder.decode(CompleteResponse.self, from: message)
97+
else {
98+
try await error(.invalidResponseFormat(messageType: .complete))
99+
return
96100
}
101+
try await onComplete(completeResponse, self)
102+
default:
103+
try await error(.invalidType())
97104
}
98105
}
99106

‎Sources/GraphQLTransportWS/Server.swift‎

Lines changed: 54 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -51,68 +51,76 @@ where
5151
self.onOperationComplete = onOperationComplete
5252
self.onOperationError = onOperationError
5353
}
54+
55+
deinit {
56+
subscriptionTasks.values.forEach { $0.cancel() }
57+
}
58+
59+
/// Listen and react to the provided async sequence of client messages. This function will block until the stream is completed.
60+
/// - Parameter incoming: The client message sequence that the server should react to.
61+
public func listen<A: AsyncSequence & Sendable>(to incoming: A) async throws
62+
where A.Element == Data {
63+
for try await message in incoming {
64+
try await respond(to: message)
65+
}
66+
}
5467

5568
/// Listen and react to the provided async sequence of client messages. This function will block until the stream is completed.
5669
/// - Parameter incoming: The client message sequence that the server should react to.
70+
@available(*, deprecated, message: "Use `Data` sequence instead.")
5771
public func listen<A: AsyncSequence & Sendable>(to incoming: A) async throws
5872
where A.Element == String {
59-
for try await message in incoming {
60-
// Detect and ignore error responses.
61-
if message.starts(with: "44") {
62-
// TODO: Determine what to do with returned error messages
73+
for try await stringMessage in incoming {
74+
guard let message = stringMessage.data(using: .utf8) else {
75+
try await error(.invalidEncoding())
6376
return
6477
}
6578

66-
guard let json = message.data(using: .utf8) else {
67-
try await error(.invalidEncoding())
79+
try await respond(to: message)
80+
}
81+
}
82+
83+
private func respond(to message: Data) async throws {
84+
let request: Request
85+
do {
86+
request = try decoder.decode(Request.self, from: message)
87+
} catch {
88+
try await self.error(.noType())
89+
return
90+
}
91+
92+
// handle incoming message
93+
switch request.type {
94+
case .connectionInit:
95+
guard
96+
let connectionInitRequest = try? decoder.decode(
97+
ConnectionInitRequest<InitPayload>.self,
98+
from: message
99+
)
100+
else {
101+
try await error(.invalidRequestFormat(messageType: .connectionInit))
68102
return
69103
}
70-
71-
let request: Request
72-
do {
73-
request = try decoder.decode(Request.self, from: json)
74-
} catch {
75-
try await self.error(.noType())
104+
try await onConnectionInit(connectionInitRequest, messenger)
105+
case .subscribe:
106+
guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: message)
107+
else {
108+
try await error(.invalidRequestFormat(messageType: .subscribe))
76109
return
77110
}
78-
79-
// handle incoming message
80-
switch request.type {
81-
case .connectionInit:
82-
guard
83-
let connectionInitRequest = try? decoder.decode(
84-
ConnectionInitRequest<InitPayload>.self,
85-
from: json
86-
)
87-
else {
88-
try await error(.invalidRequestFormat(messageType: .connectionInit))
89-
return
90-
}
91-
try await onConnectionInit(connectionInitRequest, messenger)
92-
case .subscribe:
93-
guard let subscribeRequest = try? decoder.decode(SubscribeRequest.self, from: json)
94-
else {
95-
try await error(.invalidRequestFormat(messageType: .subscribe))
96-
return
97-
}
98-
try await onSubscribe(subscribeRequest)
99-
case .complete:
100-
guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: json)
101-
else {
102-
try await error(.invalidRequestFormat(messageType: .complete))
103-
return
104-
}
105-
try await onOperationComplete(completeRequest)
106-
default:
107-
try await error(.invalidType())
111+
try await onSubscribe(subscribeRequest)
112+
case .complete:
113+
guard let completeRequest = try? decoder.decode(CompleteRequest.self, from: message)
114+
else {
115+
try await error(.invalidRequestFormat(messageType: .complete))
116+
return
108117
}
118+
try await onOperationComplete(completeRequest)
119+
default:
120+
try await error(.invalidType())
109121
}
110122
}
111123

112-
deinit {
113-
subscriptionTasks.values.forEach { $0.cancel() }
114-
}
115-
116124
private func onConnectionInit(
117125
_ connectionInitRequest: ConnectionInitRequest<InitPayload>,
118126
_: Messenger

‎Tests/GraphQLTransportWSTests/GraphQLTransportWSTests.swift‎

Lines changed: 20 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -28,21 +28,7 @@ struct GraphqlTransportWSTests {
2828
).get()
2929
}
3030
)
31-
let (messageStream, messageContinuation) = AsyncThrowingStream<String, any Error>
32-
.makeStream()
33-
let serverMessageStream = serverMessenger.stream.map { message in
34-
messageContinuation.yield(message)
35-
// Expect only one message
36-
messageContinuation.finish()
37-
return message
38-
}
39-
let client = Client<TokenInitPayload>(
40-
messenger: clientMessenger,
41-
onError: { message, _ in
42-
messageContinuation.finish(throwing: message.payload[0])
43-
await clientMessenger.close()
44-
}
45-
)
31+
let client = Client<TokenInitPayload>(messenger: clientMessenger)
4632
let clientStream = clientMessenger.stream
4733
Task {
4834
try await server.listen(to: clientStream)
@@ -59,13 +45,16 @@ struct GraphqlTransportWSTests {
5945
),
6046
id: UUID().uuidString
6147
)
62-
try await client.listen(to: serverMessageStream)
6348

64-
let messages = try await messageStream.reduce(into: [String]()) { result, message in
65-
result.append(message)
49+
let error = await #expect(throws: TestMessengerError.self) {
50+
try await client.listen(to: serverMessenger.stream)
6651
}
6752
#expect(
68-
messages == ["\(ErrorCode.notInitialized): Connection not initialized"]
53+
error
54+
== TestMessengerError(
55+
code: ErrorCode.notInitialized.rawValue,
56+
message: "Connection not initialized"
57+
)
6958
)
7059
}
7160

@@ -91,21 +80,7 @@ struct GraphqlTransportWSTests {
9180
).get()
9281
}
9382
)
94-
let (messageStream, messageContinuation) = AsyncThrowingStream<String, any Error>
95-
.makeStream()
96-
let serverMessageStream = serverMessenger.stream.map { message in
97-
messageContinuation.yield(message)
98-
// Expect only one message
99-
messageContinuation.finish()
100-
return message
101-
}
102-
let client = Client<TokenInitPayload>(
103-
messenger: clientMessenger,
104-
onError: { message, _ in
105-
messageContinuation.finish(throwing: message.payload[0])
106-
await clientMessenger.close()
107-
}
108-
)
83+
let client = Client<TokenInitPayload>(messenger: clientMessenger)
10984
let clientStream = clientMessenger.stream
11085
Task {
11186
try await server.listen(to: clientStream)
@@ -117,13 +92,16 @@ struct GraphqlTransportWSTests {
11792
authToken: ""
11893
)
11994
)
120-
try await client.listen(to: serverMessageStream)
12195

122-
let messages = try await messageStream.reduce(into: [String]()) { result, message in
123-
result.append(message)
96+
let error = await #expect(throws: TestMessengerError.self) {
97+
try await client.listen(to: serverMessenger.stream)
12498
}
12599
#expect(
126-
messages == ["\(ErrorCode.unauthorized): Unauthorized"]
100+
error
101+
== TestMessengerError(
102+
code: ErrorCode.unauthorized.rawValue,
103+
message: "Unauthorized"
104+
)
127105
)
128106
}
129107

@@ -149,7 +127,7 @@ struct GraphqlTransportWSTests {
149127
).get()
150128
}
151129
)
152-
let (messageStream, messageContinuation) = AsyncThrowingStream<String, any Error>
130+
let (messageStream, messageContinuation) = AsyncThrowingStream<Data, any Error>
153131
.makeStream()
154132
let serverMessageStream = serverMessenger.stream.map { message in
155133
messageContinuation.yield(message)
@@ -191,7 +169,7 @@ struct GraphqlTransportWSTests {
191169
)
192170
try await client.listen(to: serverMessageStream)
193171

194-
let messages = try await messageStream.reduce(into: [String]()) { result, message in
172+
let messages = try await messageStream.reduce(into: [Data]()) { result, message in
195173
result.append(message)
196174
}
197175
#expect(
@@ -226,7 +204,7 @@ struct GraphqlTransportWSTests {
226204
return subscription
227205
}
228206
)
229-
let (messageStream, messageContinuation) = AsyncThrowingStream<String, any Error>
207+
let (messageStream, messageContinuation) = AsyncThrowingStream<Data, any Error>
230208
.makeStream()
231209
// Used to extract the server messages
232210
let serverMessageStream = serverMessenger.stream.map { message in
@@ -282,7 +260,7 @@ struct GraphqlTransportWSTests {
282260
)
283261
try await client.listen(to: serverMessageStream)
284262

285-
let messages = try await messageStream.reduce(into: [String]()) { result, message in
263+
let messages = try await messageStream.reduce(into: [Data]()) { result, message in
286264
result.append(message)
287265
}
288266
#expect(

0 commit comments

Comments
 (0)