Skip to content

Commit 681e213

Browse files
author
Jay Herron
committed
Adds subscription result and test stub
1 parent 7a4abcf commit 681e213

3 files changed

Lines changed: 179 additions & 106 deletions

File tree

‎Sources/GraphQL/Subscription/SimplePubSub.swift‎

Lines changed: 0 additions & 67 deletions
This file was deleted.

‎Sources/GraphQL/Subscription/Subscribe.swift‎

Lines changed: 22 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ func subscribe(
3838
operationName: String? = nil,
3939
fieldResolver: GraphQLFieldResolve,
4040
subscribeFieldResolver: GraphQLFieldResolve
41-
) -> EventLoopFuture<Any?> { // This is either an AsyncIterator or a GraphQLResult
41+
) -> EventLoopFuture<SubscriptionResult> {
4242

4343

4444
let sourceFuture = createSourceEventStream(
@@ -77,16 +77,18 @@ func subscribe(
7777
operationName: operationName
7878
)
7979
}
80-
return sourceFuture.flatMap({ (resultOrStream) -> EventLoopFuture<Any?> in
81-
if resultOrStream is GraphQLResult { // Return the result directly
82-
return eventLoopGroup.next().makeSucceededFuture(resultOrStream)
83-
} else { // We can assume that it's an AsyncIterable
84-
let stream:AsyncIterable<GraphQLResult>! = resultOrStream
85-
return MappedAsyncIterator(from: stream) { payload -> EventLoopFuture<GraphQLResult> in
86-
mapSourceToResponse(payload: payload)
80+
return sourceFuture.flatMap{ subscriptionResult -> EventLoopFuture<SubscriptionResult> in
81+
do {
82+
let subscriptionObserver = try subscriptionResult.get()
83+
let eventObserver = subscriptionObserver.map { eventPayload -> GraphQLResult in
84+
return try! mapSourceToResponse(payload: eventPayload).wait() // TODO Remove this wait
8785
}
86+
// TODO Making a future here feels it indicates a mistake...
87+
return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.success(eventObserver))
88+
} catch let graphQLError as GraphQLError {
89+
return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.failure(graphQLError))
8890
}
89-
})
91+
}
9092
}
9193

9294
/**
@@ -130,7 +132,7 @@ func createSourceEventStream(
130132
variableValues: [String: Map] = [:],
131133
operationName: String? = nil,
132134
subscribeFieldResolver: GraphQLFieldResolve
133-
) -> EventLoopFuture<Any?> { // This is either an AsyncIterator or a GraphQLResult
135+
) -> EventLoopFuture<SubscriptionResult> {
134136

135137
let executeStarted = instrumentation.now
136138
let exeContext: ExecutionContext
@@ -168,9 +170,9 @@ func createSourceEventStream(
168170
result: nil
169171
)
170172

171-
return eventLoopGroup.next().makeSucceededFuture(GraphQLResult(errors: [error]))
173+
return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.failure(error))
172174
} catch {
173-
return eventLoopGroup.next().makeSucceededFuture(GraphQLResult(errors: [GraphQLError(error)]))
175+
return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.failure(GraphQLError(error)))
174176
}
175177

176178
return try! executeSubscription(context: exeContext, eventLoopGroup: eventLoopGroup)
@@ -179,7 +181,7 @@ func createSourceEventStream(
179181
func executeSubscription(
180182
context: ExecutionContext,
181183
eventLoopGroup: EventLoopGroup
182-
) throws -> EventLoopFuture<Any?> { // This is either an AsyncIterator or a GraphQLResult
184+
) throws -> EventLoopFuture<SubscriptionResult> {
183185

184186
// Get the first node
185187
let type = try getOperationRootType(schema: context.schema, operation: context.operation)
@@ -226,7 +228,7 @@ func executeSubscription(
226228
let contextValue = context.context
227229

228230
// Call the `subscribe()` resolver or the default resolver to produce an
229-
// AsyncIterable yielding raw payloads.
231+
// Observable yielding raw payloads.
230232
let resolve = fieldDef.subscribe ?? fieldDef.resolve ?? defaultResolve
231233

232234
// Get the resolve func, regardless of if its result is normal
@@ -247,33 +249,14 @@ func executeSubscription(
247249
info: info,
248250
path: path,
249251
result: result
250-
).flatMap { value -> EventLoopFuture<Any?> in
251-
if let asyncIterable = value as? EventLoopFuture<AsyncIterable> {
252-
return asyncIterable
252+
).map { value -> SubscriptionResult in
253+
if let observable = value as? Observable<GraphQLResult> {
254+
return SubscriptionResult.success(observable)
253255
} else {
254-
context.append(error: GraphQLError(message: "Subscription field must return AsyncIterable."))
255-
return context.eventLoopGroup.next().makeSucceededFuture(nil)
256+
context.append(error: GraphQLError(message: "Subscription field resolver must return Observable of GraphQLResults."))
257+
return SubscriptionResult.failure(GraphQLError(message: "Subscription field resolver must return Observable of GraphQLResults."))
256258
}
257259
}
258260
}
259261

260-
protocol AsyncIterable {
261-
associatedtype Item
262-
func next() -> EventLoopFuture<Item>
263-
}
264-
265-
class MappedAsyncIterator<OrigType: AsyncIterable, MappedType> : AsyncIterable {
266-
let origIterable: OrigType
267-
let callback: (OrigType.Item) -> Future<MappedType>
268-
269-
init(from: OrigType, by: @escaping (OrigType.Item) -> Future<MappedType>) {
270-
origIterable = from
271-
callback = by
272-
}
273-
274-
func next() -> EventLoopFuture<MappedType> {
275-
origIterable.next().flatMap { origResult -> Future<MappedType> in
276-
self.callback(origResult)
277-
}
278-
}
279-
}
262+
typealias SubscriptionResult = Result<Observable<GraphQLResult>, GraphQLError>

‎Tests/GraphQLTests/Subscription/SubscriptionTests.swift‎

Lines changed: 157 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,157 @@
1+
import XCTest
2+
import NIO
3+
import RxSwift
4+
@testable import GraphQL
5+
6+
class SubscriptionTests : XCTestCase {
7+
8+
private func createSubscription(
9+
pubsub:Observable<Email>,
10+
schema:GraphQLSchema = EmailSchema,
11+
document:Document = defaultSubscriptionAST
12+
) -> EventLoopFuture<SubscriptionResult> {
13+
14+
let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: System.coreCount)
15+
16+
// TODO figure out how to generate the subscription
17+
// return subscribe(
18+
// schema: schema,
19+
// documentAST: document,
20+
// eventLoopGroup: eventLoopGroup
21+
// )
22+
23+
// TODO Remove placeholder below
24+
return eventLoopGroup.next().makeSucceededFuture(SubscriptionResult.failure(GraphQLError(message:"PLACEHOLDER")))
25+
}
26+
}
27+
28+
let defaultSubscriptionAST = try! parse(source: """
29+
subscription ($priority: Int = 0) {
30+
importantEmail(priority: $priority) {
31+
email {
32+
from
33+
subject
34+
}
35+
inbox {
36+
unread
37+
total
38+
}
39+
}
40+
}
41+
""")
42+
43+
// MARK: Types
44+
struct Email {
45+
let from:String
46+
let subject:String
47+
let message:String
48+
let unread:Bool
49+
}
50+
51+
struct Inbox {
52+
let emails:[Email]
53+
}
54+
55+
struct EmailEvent {
56+
let email:Email
57+
let inbox:Inbox
58+
}
59+
60+
let emails = [
61+
Email(
62+
from: "joe@graphql.org",
63+
subject: "Hello",
64+
message: "Hello World",
65+
unread: false
66+
)
67+
]
68+
69+
func importantEmail(priority: Int) -> Observable<EmailEvent> {
70+
let inbox = Inbox(emails: emails)
71+
let emailObs = Observable.from(emails)
72+
let emailEventObs = emailObs.map { email -> EmailEvent in
73+
return EmailEvent(email: email, inbox: inbox)
74+
}
75+
return emailEventObs
76+
}
77+
78+
// MARK: Schema
79+
let EmailType = try! GraphQLObjectType(
80+
name: "Email",
81+
fields: [
82+
"from": GraphQLField(
83+
type: GraphQLString
84+
),
85+
"subject": GraphQLField(
86+
type: GraphQLString
87+
),
88+
"message": GraphQLField(
89+
type: GraphQLString
90+
),
91+
"unread": GraphQLField(
92+
type: GraphQLBoolean
93+
),
94+
]
95+
)
96+
let InboxType = try! GraphQLObjectType(
97+
name: "Inbox",
98+
fields: [
99+
"emails": GraphQLField(
100+
type: GraphQLList(EmailType)
101+
),
102+
"total": GraphQLField(
103+
type: GraphQLInt,
104+
resolve: { inbox, _, _, _ in
105+
(inbox as! Inbox).emails.count
106+
}
107+
),
108+
// TODO figure out how to do searches
109+
// "unread": GraphQLField(
110+
// type: GraphQLInt,
111+
// resolve: { inbox, _, _, _ in
112+
// (inbox as! InboxType).emails.
113+
// }
114+
// ),
115+
]
116+
)
117+
118+
let EmailEventType = try! GraphQLObjectType(
119+
name: "EmailEvent",
120+
fields: [
121+
"email": GraphQLField(
122+
type: EmailType
123+
),
124+
"inbox": GraphQLField(
125+
type: InboxType
126+
)
127+
]
128+
)
129+
130+
let EmailSchema = try! GraphQLSchema(
131+
query: try! GraphQLObjectType(
132+
name: "Query",
133+
fields: [
134+
"inbox": GraphQLField(
135+
type: InboxType
136+
)
137+
]
138+
),
139+
subscription: try! GraphQLObjectType(
140+
name: "Subscription",
141+
fields: [
142+
"importantEmail": GraphQLField(
143+
type: EmailEventType,
144+
args: [
145+
"priority": GraphQLArgument(
146+
type: GraphQLInt
147+
)
148+
],
149+
resolve: { _, arguments, _, _ in
150+
let priority = arguments["priority"].int!
151+
return importantEmail(priority: priority)
152+
}
153+
// subscribe: subscribeFn // TODO Fill in the subscribe function
154+
)
155+
]
156+
)
157+
)

0 commit comments

Comments
 (0)