Changes observer to resolve to a future

Author: Jay Herron

Sources/GraphQL/Subscription/Subscribe.swift

@@ -59,15 +59,15 @@ func subscribe(
     return sourceFuture.map{ subscriptionResult -> SubscriptionResult in
         do {
             let subscriptionObserver = try subscriptionResult.get()
-            let eventObserver = subscriptionObserver.map { eventPayload -> GraphQLResult in
+            let eventObserver = subscriptionObserver.map { eventPayload -> Future<GraphQLResult> in
 
                 // For each payload yielded from a subscription, map it over the normal
                 // GraphQL `execute` function, with `payload` as the rootValue.
                 // This implements the "MapSourceToResponseEvent" algorithm described in
                 // the GraphQL specification. The `execute` function provides the
                 // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
                 // "ExecuteQuery" algorithm, for which `execute` is also used.
-                let eventResolved = try execute(
+                return execute(
                     queryStrategy: queryStrategy,
                     mutationStrategy: mutationStrategy,
                     subscriptionStrategy: subscriptionStrategy,
@@ -79,10 +79,8 @@ func subscribe(
                     eventLoopGroup: eventLoopGroup,
                     variableValues: variableValues,
                     operationName: operationName
-                ).wait() // TODO remove this wait
-                return eventResolved
+                )
             }
-            // TODO Making a future here feels it indicates a mistake...
             return SubscriptionResult.success(eventObserver)
         } catch let graphQLError as GraphQLError {
             return SubscriptionResult.failure(graphQLError)
@@ -253,19 +251,21 @@ func executeSubscription(
             return SourceEventStreamResult.failure(context.errors.first!)
         } else if let error = resolved as? GraphQLError {
             return SourceEventStreamResult.failure(error)
-        } else if let observable = resolved as? Observable<Any> {
+        } else if let observable = resolved as? SourceEventStreamObservable {
             return SourceEventStreamResult.success(observable)
         } else if resolved == nil {
             return SourceEventStreamResult.failure(
                 GraphQLError(message: "Resolved subscription was nil")
             )
         } else {
             return SourceEventStreamResult.failure(
-                GraphQLError(message: "Subscription field resolver must return an Observable<Any>, not \(Swift.type(of:resolved))")
+                GraphQLError(message: "Subscription field resolver must return an SourceEventStreamObservable, not \(Swift.type(of:resolved))")
             )
         }
     }
 }
 
-typealias SubscriptionResult = Result<Observable<GraphQLResult>, GraphQLError>
-typealias SourceEventStreamResult = Result<Observable<Any>, GraphQLError>
+typealias SubscriptionObservable = Observable<Future<GraphQLResult>>
+typealias SubscriptionResult = Result<SubscriptionObservable, GraphQLError>
+typealias SourceEventStreamObservable = Observable<Any>
+typealias SourceEventStreamResult = Result<SourceEventStreamObservable, GraphQLError>

Tests/GraphQLTests/Subscription/SubscriptionTests.swift

@@ -25,7 +25,8 @@ class SubscriptionTests : XCTestCase {
             ]]
         )
         let _ = subscription.subscribe { event in
-            XCTAssertEqual(event.element, expected)
+            let payload = try! event.element!.wait()
+            XCTAssertEqual(payload, expected)
         }.disposed(by: disposeBag)
         pubsub.onNext(Email(
             from: "yuzhi@graphql.org",
@@ -105,7 +106,8 @@ class SubscriptionTests : XCTestCase {
             ]]
         )
         let _ = subscription.subscribe { event in
-            XCTAssertEqual(event.element, expected)
+            let payload = try! event.element!.wait()
+            XCTAssertEqual(payload, expected)
         }.disposed(by: disposeBag)
         pubsub.onNext(Email(
             from: "yuzhi@graphql.org",
@@ -158,7 +160,9 @@ class SubscriptionTests : XCTestCase {
             }
         """)
 
-        let _ = subscription.subscribe().disposed(by: disposeBag)
+        let _ = subscription.subscribe{ event in
+            let _ = try! event.element!.wait()
+        }.disposed(by: disposeBag)
         pubsub.onNext(Email(
             from: "yuzhi@graphql.org",
             subject: "Alright",
@@ -278,7 +282,7 @@ let defaultEmails = [
 private func createDbAndSubscription(
     pubsub:Observable<Any>,
     query:String
-) throws -> Observable<GraphQLResult> {
+) throws -> SubscriptionObservable {
 
     var emails = defaultEmails
 
@@ -304,7 +308,7 @@ private func createSubscription(
     pubsub: Observable<Any>,
     schema: GraphQLSchema,
     query: String
-) throws -> Observable<GraphQLResult> {
+) throws -> SubscriptionObservable {
     let document = try parse(source: query)
     let eventLoopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 1)
     let subscriptionOrError = try subscribe(
@@ -344,7 +348,7 @@ private func emailSchemaWithResolvers(resolve: GraphQLFieldResolve?, subscribe:
     )
 }
 
-private func extractSubscription(_ subscriptionResult: SubscriptionResult) throws -> Observable<GraphQLResult> {
+private func extractSubscription(_ subscriptionResult: SubscriptionResult) throws -> SubscriptionObservable {
     switch subscriptionResult {
     case .success(let subscription):
         return subscription