From 99e128b87df189010a04406bc7932af27e22cb64 Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Wed, 17 Feb 2021 13:09:15 -0700
Subject: [PATCH 01/15] Adds subscription Schema support

---
 .../Graphiti/Subscription/Subscription.swift  | 53 +++++++++++++++++++
 .../HelloWorldTests/HelloWorldTests.swift     | 28 ++++++++++
 2 files changed, 81 insertions(+)
 create mode 100644 Sources/Graphiti/Subscription/Subscription.swift

diff --git a/Sources/Graphiti/Subscription/Subscription.swift b/Sources/Graphiti/Subscription/Subscription.swift
new file mode 100644
index 00000000..76b64590
--- /dev/null
+++ b/Sources/Graphiti/Subscription/Subscription.swift
@@ -0,0 +1,53 @@
+import GraphQL
+
+public final class Subscription<Resolver, Context> : Component<Resolver, Context> {
+    let fields: [FieldComponent<Resolver, Context>]
+    
+    let isTypeOf: GraphQLIsTypeOf = { source, _, _ in
+        return source is Resolver
+    }
+    
+    override func update(typeProvider: SchemaTypeProvider) throws {
+        typeProvider.subscription = try GraphQLObjectType(
+            name: name,
+            description: description,
+            fields: fields(typeProvider: typeProvider),
+            isTypeOf: isTypeOf
+        )
+    }
+    
+    func fields(typeProvider: TypeProvider) throws -> GraphQLFieldMap {
+        var map: GraphQLFieldMap = [:]
+        
+        for field in fields {
+            let (name, field) = try field.field(typeProvider: typeProvider)
+            map[name] = field
+        }
+        
+        return map
+    }
+    
+    private init(
+        name: String,
+        fields: [FieldComponent<Resolver, Context>]
+    ) {
+        self.fields = fields
+        super.init(name: name)
+    }
+}
+
+public extension Subscription {
+    convenience init(
+        as name: String = "Subscription",
+        @FieldComponentBuilder<Resolver, Context> _ fields: () -> FieldComponent<Resolver, Context>
+    ) {
+        self.init(name: name, fields: [fields()])
+    }
+    
+    convenience init(
+        as name: String = "Subscription",
+        @FieldComponentBuilder<Resolver, Context> _ fields: () -> [FieldComponent<Resolver, Context>]
+    ) {
+        self.init(name: name, fields: fields())
+    }
+}
diff --git a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
index 63898783..c52cd304 100644
--- a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
+++ b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
@@ -130,6 +130,10 @@ struct HelloAPI : API {
                 Argument("user", at: \.user)
             }
         }
+        
+        Subscription {
+            Field("hello", at: HelloResolver.hello)
+        }
     }
 }
 
@@ -318,6 +322,30 @@ class HelloWorldTests : XCTestCase {
         
         wait(for: [expectation], timeout: 10)
     }
+    
+    // TODO Adjust this to expect an Observable when GraphQL is piped up correctly.
+    func testSubscription() throws {
+        let subscription = """
+        subscription {
+            hello
+        }
+        """
+        
+        let expected = GraphQLResult(data: ["hello": "world"])
+        
+        let expectation = XCTestExpectation()
+        
+        api.execute(
+            request: subscription,
+            context: api.context,
+            on: group
+        ).whenSuccess { result in
+            XCTAssertEqual(result, expected)
+            expectation.fulfill()
+        }
+        
+        wait(for: [expectation], timeout: 10)
+    }
 }
 
 extension HelloWorldTests {

From 83413a65970ab08deb8993cb60de9a14dfe26e05 Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Tue, 23 Feb 2021 17:46:48 -0700
Subject: [PATCH 02/15] Adds RxSwift dependency

---
 Package.swift | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/Package.swift b/Package.swift
index dce5cc34..9b245755 100644
--- a/Package.swift
+++ b/Package.swift
@@ -8,9 +8,10 @@ let package = Package(
     ],
     dependencies: [
         .package(url: "https://github.com/GraphQLSwift/GraphQL.git", .upToNextMajor(from: "1.1.7")),
+        .package(url: "https://github.com/ReactiveX/RxSwift.git", .upToNextMajor(from: "6.1.0"))
     ],
     targets: [
-        .target(name: "Graphiti", dependencies: ["GraphQL"]),
+        .target(name: "Graphiti", dependencies: ["GraphQL", "RxSwift"]),
         .testTarget(name: "GraphitiTests", dependencies: ["Graphiti"]),
     ]
 )

From b38bbceb045e6c965eac8676df0e5a46fda95ef1 Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Tue, 23 Feb 2021 17:47:21 -0700
Subject: [PATCH 03/15] Adds API and Schema subscription functions

---
 Sources/Graphiti/API/API.swift       | 18 ++++++++++++++++++
 Sources/Graphiti/Schema/Schema.swift | 24 ++++++++++++++++++++++++
 2 files changed, 42 insertions(+)

diff --git a/Sources/Graphiti/API/API.swift b/Sources/Graphiti/API/API.swift
index 6bb3e90f..d8567850 100644
--- a/Sources/Graphiti/API/API.swift
+++ b/Sources/Graphiti/API/API.swift
@@ -1,5 +1,6 @@
 import GraphQL
 import NIO
+import RxSwift
 
 public protocol API {
     associatedtype Resolver
@@ -25,4 +26,21 @@ extension API {
             operationName: operationName
         )
     }
+    
+    public func subscribe(
+        request: String,
+        context: ContextType,
+        on eventLoopGroup: EventLoopGroup,
+        variables: [String: Map] = [:],
+        operationName: String? = nil
+    ) -> EventLoopFuture<SubscriptionResult> {
+        return schema.subscribe(
+            request: request,
+            resolver: resolver,
+            context: context,
+            eventLoopGroup: eventLoopGroup,
+            variables: variables,
+            operationName: operationName
+        )
+    }
 }
diff --git a/Sources/Graphiti/Schema/Schema.swift b/Sources/Graphiti/Schema/Schema.swift
index 8d9a4d7e..de013ada 100644
--- a/Sources/Graphiti/Schema/Schema.swift
+++ b/Sources/Graphiti/Schema/Schema.swift
@@ -1,5 +1,6 @@
 import GraphQL
 import NIO
+import RxSwift
 
 public final class Schema<Resolver, Context> {
     public let schema: GraphQLSchema
@@ -56,4 +57,27 @@ public extension Schema {
             return eventLoopGroup.next().makeFailedFuture(error)
         }
     }
+    
+    func subscribe(
+        request: String,
+        resolver: Resolver,
+        context: Context,
+        eventLoopGroup: EventLoopGroup,
+        variables: [String: Map] = [:],
+        operationName: String? = nil
+    ) -> EventLoopFuture<SubscriptionResult> {
+        do {
+            return try graphqlSubscribe(
+                schema: schema,
+                request: request,
+                rootValue: resolver,
+                context: context,
+                eventLoopGroup: eventLoopGroup,
+                variableValues: variables,
+                operationName: operationName
+            )
+        } catch {
+            return eventLoopGroup.next().makeFailedFuture(error)
+        }
+    }
 }

From e7548b4e5ff3c2d8fbc3d7eed1edf361e5d3fb02 Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Tue, 23 Feb 2021 17:47:54 -0700
Subject: [PATCH 04/15] Adds new subscription field and resolve types

---
 .../Subscription/SubscribeField.swift         | 293 ++++++++++++++++++
 .../Subscription/SubscribeResolve.swift       |   9 +
 2 files changed, 302 insertions(+)
 create mode 100644 Sources/Graphiti/Subscription/SubscribeField.swift
 create mode 100644 Sources/Graphiti/Subscription/SubscribeResolve.swift

diff --git a/Sources/Graphiti/Subscription/SubscribeField.swift b/Sources/Graphiti/Subscription/SubscribeField.swift
new file mode 100644
index 00000000..565268ef
--- /dev/null
+++ b/Sources/Graphiti/Subscription/SubscribeField.swift
@@ -0,0 +1,293 @@
+import GraphQL
+import Runtime
+import RxSwift
+
+public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType, Arguments : Decodable> : FieldComponent<ObjectType, Context> {
+    let name: String
+    let arguments: [ArgumentComponent<Arguments>]
+    let resolve: GraphQLFieldResolve
+    let subscribe: GraphQLFieldResolve
+    
+    override func field(typeProvider: TypeProvider) throws -> (String, GraphQLField) {
+        let field = GraphQLField(
+            type: try typeProvider.getOutputType(from: FieldType.self, field: name),
+            description: description,
+            deprecationReason: deprecationReason,
+            args: try arguments(typeProvider: typeProvider),
+            resolve: resolve,
+            subscribe: subscribe
+        )
+        
+        return (name, field)
+    }
+    
+    func arguments(typeProvider: TypeProvider) throws -> GraphQLArgumentConfigMap {
+        var map: GraphQLArgumentConfigMap = [:]
+        
+        for argument in arguments {
+            let (name, argument) = try argument.argument(typeProvider: typeProvider)
+            map[name] = argument
+        }
+        
+        return map
+    }
+    
+    init(
+        name: String,
+        arguments: [ArgumentComponent<Arguments>],
+        resolve: @escaping GraphQLFieldResolve,
+        subscribe: @escaping GraphQLFieldResolve
+    ) {
+        self.name = name
+        self.arguments = arguments
+        self.resolve = resolve
+        self.subscribe = subscribe
+    }
+
+    convenience init<ResolveType>(
+        name: String,
+        arguments: [ArgumentComponent<Arguments>],
+        asyncResolve: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>
+    ) {
+        let resolve: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
+            guard let s = source as? SubscriptionType else {
+                throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
+            }
+            
+            guard let c = context as? Context else {
+                throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
+            }
+        
+            let a = try MapDecoder().decode(Arguments.self, from: arguments)
+            return try asyncResolve(s)(c, a, eventLoopGroup).map({ $0 })
+        }
+        
+        let subscribe: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
+            guard let s = source as? ObjectType else {
+                throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
+            }
+        
+            guard let c = context as? Context else {
+                throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
+            }
+        
+            let a = try MapDecoder().decode(Arguments.self, from: arguments)
+            return try asyncSubscribe(s)(c, a, eventLoopGroup).map({ $0 })
+        }
+        self.init(name: name, arguments: arguments, resolve: resolve, subscribe: subscribe)
+    }
+    
+    convenience init<ResolveType>(
+        name: String,
+        arguments: [ArgumentComponent<Arguments>],
+        simpleAsyncResolve: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>
+    ) {
+        let asyncResolve: AsyncResolve<SubscriptionType, Context, Arguments, ResolveType> = { type in
+            { context, arguments, group in
+                // We hop to guarantee that the future will
+                // return in the same event loop group of the execution.
+                try simpleAsyncResolve(type)(context, arguments).hop(to: group.next())
+            }
+        }
+        
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>> = { type in
+            { context, arguments, group in
+                // We hop to guarantee that the future will
+                // return in the same event loop group of the execution.
+                try simpleAsyncSubscribe(type)(context, arguments).hop(to: group.next())
+            }
+        }
+        self.init(name: name, arguments: arguments, asyncResolve: asyncResolve, asyncSubscribe: asyncSubscribe)
+    }
+    
+    convenience init<ResolveType>(
+        name: String,
+        arguments: [ArgumentComponent<Arguments>],
+        syncResolve: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>
+    ) {
+        let asyncResolve: AsyncResolve<SubscriptionType, Context, Arguments, ResolveType> = { type in
+            { context, arguments, group in
+                let result = try syncResolve(type)(context, arguments)
+                return group.next().makeSucceededFuture(result)
+            }
+        }
+        
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>> = { type in
+            { context, arguments, group in
+                let result = try syncSubscribe(type)(context, arguments)
+                return group.next().makeSucceededFuture(result)
+            }
+        }
+        self.init(name: name, arguments: arguments, asyncResolve: asyncResolve, asyncSubscribe: asyncSubscribe)
+    }
+}
+
+// MARK: AsyncResolve Initializers
+
+public extension SubscriptionField where FieldType : Encodable {
+    convenience init(
+        _ name: String,
+        at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], asyncResolve: function, asyncSubscribe: subFunc)
+    }
+    
+    convenience init(
+        _ name: String,
+        at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), asyncResolve: function, asyncSubscribe: subFunc)
+    }
+}
+
+public extension SubscriptionField {
+    convenience init<ResolveType>(
+        _ name: String,
+        at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        as: FieldType.Type,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], asyncResolve: function, asyncSubscribe: subFunc)
+    }
+    
+    convenience init<ResolveType>(
+        _ name: String,
+        at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        as: FieldType.Type,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), asyncResolve: function, asyncSubscribe: subFunc)
+    }
+}
+
+// MARK: SimpleAsyncResolve Initializers
+
+public extension SubscriptionField where FieldType : Encodable {
+    convenience init(
+        _ name: String,
+        at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
+    }
+    
+    convenience init(
+        _ name: String,
+        at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
+    }
+}
+
+public extension SubscriptionField {
+    convenience init<ResolveType>(
+        _ name: String,
+        at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        as: FieldType.Type,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
+    }
+    
+    convenience init<ResolveType>(
+        _ name: String,
+        at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        as: FieldType.Type,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
+    }
+}
+
+// MARK: SyncResolve Initializers
+
+public extension SubscriptionField where FieldType : Encodable {
+    convenience init(
+        _ name: String,
+        at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], syncResolve: function, syncSubscribe: subFunc)
+    }
+    
+    convenience init(
+        _ name: String,
+        at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), syncResolve: function, syncSubscribe: subFunc)
+    }
+}
+
+public extension SubscriptionField {
+    convenience init<ResolveType>(
+        _ name: String,
+        at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        as: FieldType.Type,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], syncResolve: function, syncSubscribe: subFunc)
+    }
+    
+    convenience init<ResolveType>(
+        _ name: String,
+        at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        as: FieldType.Type,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), syncResolve: function, syncSubscribe: subFunc)
+    }
+}
+
+// TODO Determine if we can use keypaths to initialize
+
+// MARK: Keypath Initializers
+
+//public extension SubscriptionField where Arguments == NoArguments {
+//    convenience init(
+//        _ name: String,
+//        at keyPath: KeyPath<ObjectType, FieldType>
+//    ) {
+//        let syncResolve: SyncResolve<SubscriptionType, Context, Arguments, ResolveType> = { type in
+//            { context, _ in
+//                type[keyPath: keyPath]
+//            }
+//        }
+//
+//        self.init(name: name, arguments: [], syncResolve: syncResolve)
+//    }
+//}
+//
+//public extension SubscriptionField where Arguments == NoArguments {
+//    convenience init<ResolveType>(
+//        _ name: String,
+//        at keyPath: KeyPath<ObjectType, ResolveType>,
+//        as: FieldType.Type
+//    ) {
+//        let syncResolve: SyncResolve<ObjectType, Context, NoArguments, ResolveType> = { type in
+//            return { context, _ in
+//                return type[keyPath: keyPath]
+//            }
+//        }
+//
+//        self.init(name: name, arguments: [], syncResolve: syncResolve)
+//    }
+//}
diff --git a/Sources/Graphiti/Subscription/SubscribeResolve.swift b/Sources/Graphiti/Subscription/SubscribeResolve.swift
new file mode 100644
index 00000000..5779e8f2
--- /dev/null
+++ b/Sources/Graphiti/Subscription/SubscribeResolve.swift
@@ -0,0 +1,9 @@
+import NIO
+
+public typealias SubscribeResolve<ObjectType, Context, Arguments, ResolveType> = (
+    _ object: ObjectType
+)  -> (
+    _ context: Context,
+    _ arguments: Arguments,
+    _ eventLoopGroup: EventLoopGroup
+) throws -> EventLoopFuture<ResolveType>

From 35e471f2205374d8ab4f1b6cc9057a208d281673 Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Tue, 23 Feb 2021 17:48:57 -0700
Subject: [PATCH 05/15] Adds in start of testing

Running into type invariance issues with GraphQL
---
 .../HelloWorldTests/HelloWorldTests.swift     | 72 +++++++++++++++----
 1 file changed, 60 insertions(+), 12 deletions(-)

diff --git a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
index c52cd304..7ab0dd3b 100644
--- a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
+++ b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
@@ -1,7 +1,10 @@
 import XCTest
-@testable import Graphiti
 import GraphQL
 import NIO
+import RxSwift
+@testable import Graphiti
+
+let pubsub = PublishSubject<User>()
 
 struct ID : Codable {
     let id: String
@@ -41,12 +44,22 @@ struct UserInput : Codable {
     let name: String?
 }
 
+struct UserEvent : Codable {
+    let user: User
+}
+
 final class HelloContext {
     func hello() -> String {
         "world"
     }
 }
 
+extension User {
+    func toEvent(context: HelloContext, arguments: NoArguments) throws -> UserEvent {
+        return UserEvent(user: self)
+    }
+}
+
 struct HelloResolver {
     func hello(context: HelloContext, arguments: NoArguments) -> String {
         context.hello()
@@ -87,6 +100,10 @@ struct HelloResolver {
     func addUser(context: HelloContext, arguments: AddUserArguments) -> User {
         User(arguments.user)
     }
+    
+    func subscribeUser(context: HelloContext, arguments: NoArguments) -> Observable<User> {
+        pubsub
+    }
 }
 
 struct HelloAPI : API {
@@ -110,6 +127,10 @@ struct HelloAPI : API {
             InputField("name", at: \.name)
         }
         
+        Type(UserEvent.self) {
+            Field("user", at: \.user)
+        }
+        
         Query {
             Field("hello", at: HelloResolver.hello)
             Field("asyncHello", at: HelloResolver.asyncHello)
@@ -132,7 +153,7 @@ struct HelloAPI : API {
         }
         
         Subscription {
-            Field("hello", at: HelloResolver.hello)
+            SubscriptionField("subscribeUser", at: User.toEvent, atSub: HelloResolver.subscribeUser)
         }
     }
 }
@@ -327,24 +348,51 @@ class HelloWorldTests : XCTestCase {
     func testSubscription() throws {
         let subscription = """
         subscription {
-            hello
+            subscribeUser {
+                user {
+                    id
+                    name
+                }
+            }
         }
         """
         
-        let expected = GraphQLResult(data: ["hello": "world"])
-        
-        let expectation = XCTestExpectation()
+        let expected = GraphQLResult(data: [
+            "subscribeUser": [
+                "user": [
+                    "name": "John Doe",
+                    "id": "123"
+                ]
+            ]
+        ])
         
-        api.execute(
+        let result = try api.subscribe(
             request: subscription,
             context: api.context,
             on: group
-        ).whenSuccess { result in
-            XCTAssertEqual(result, expected)
-            expectation.fulfill()
-        }
+        ).wait()
+        print(result)
+//        let observable = try api.subscribe(
+//            request: subscription,
+//            context: api.context,
+//            on: group
+//        ).wait().observable!
         
-        wait(for: [expectation], timeout: 10)
+        let expectation = XCTestExpectation()
+        
+//        var currentResult = GraphQLResult()
+//        let _ = observable.subscribe { event in
+//            event.element!.whenSuccess { result in
+//                currentResult = result
+//                expectation.fulfill()
+//            }
+//        }
+//
+//        pubsub.onNext(User(id: "124", name: "Jerry"))
+//
+//        wait(for: [expectation], timeout: 10)
+//
+//        XCTAssertEqual(currentResult, expected)
     }
 }
 

From 217981ed9daf8fe5ceedde45cbe4d846902b798a Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Tue, 23 Feb 2021 18:23:40 -0700
Subject: [PATCH 06/15] Works out subscription field type issues

---
 .../Subscription/SubscribeField.swift         | 46 +++++++++----------
 1 file changed, 23 insertions(+), 23 deletions(-)

diff --git a/Sources/Graphiti/Subscription/SubscribeField.swift b/Sources/Graphiti/Subscription/SubscribeField.swift
index 565268ef..0400644b 100644
--- a/Sources/Graphiti/Subscription/SubscribeField.swift
+++ b/Sources/Graphiti/Subscription/SubscribeField.swift
@@ -48,7 +48,7 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         asyncResolve: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
-        asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>
+        asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
     ) {
         let resolve: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
             guard let s = source as? SubscriptionType else {
@@ -82,7 +82,7 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         simpleAsyncResolve: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
-        simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>
+        simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
     ) {
         let asyncResolve: AsyncResolve<SubscriptionType, Context, Arguments, ResolveType> = { type in
             { context, arguments, group in
@@ -92,7 +92,7 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
             }
         }
         
-        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>> = { type in
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<Any>> = { type in
             { context, arguments, group in
                 // We hop to guarantee that the future will
                 // return in the same event loop group of the execution.
@@ -106,7 +106,7 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         syncResolve: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
-        syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>
+        syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>
     ) {
         let asyncResolve: AsyncResolve<SubscriptionType, Context, Arguments, ResolveType> = { type in
             { context, arguments, group in
@@ -115,7 +115,7 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
             }
         }
         
-        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>> = { type in
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<Any>> = { type in
             { context, arguments, group in
                 let result = try syncSubscribe(type)(context, arguments)
                 return group.next().makeSucceededFuture(result)
@@ -131,7 +131,7 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init(
         _ name: String,
         at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], asyncResolve: function, asyncSubscribe: subFunc)
@@ -139,8 +139,8 @@ public extension SubscriptionField where FieldType : Encodable {
     
     convenience init(
         _ name: String,
-        at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        at function: @escaping AsyncResolve<Any, Context, Arguments, FieldType>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), asyncResolve: function, asyncSubscribe: subFunc)
@@ -152,7 +152,7 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], asyncResolve: function, asyncSubscribe: subFunc)
@@ -162,7 +162,7 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), asyncResolve: function, asyncSubscribe: subFunc)
@@ -175,16 +175,16 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
     }
-    
+
     convenience init(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
@@ -196,17 +196,17 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
     }
-    
+
     convenience init<ResolveType>(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
@@ -219,16 +219,16 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init(
         _ name: String,
         at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], syncResolve: function, syncSubscribe: subFunc)
     }
-    
+
     convenience init(
         _ name: String,
         at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), syncResolve: function, syncSubscribe: subFunc)
@@ -240,17 +240,17 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], syncResolve: function, syncSubscribe: subFunc)
     }
-    
+
     convenience init<ResolveType>(
         _ name: String,
         at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<SubscriptionType>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), syncResolve: function, syncSubscribe: subFunc)
@@ -266,7 +266,7 @@ public extension SubscriptionField {
 //        _ name: String,
 //        at keyPath: KeyPath<ObjectType, FieldType>
 //    ) {
-//        let syncResolve: SyncResolve<SubscriptionType, Context, Arguments, ResolveType> = { type in
+//        let syncResolve: SyncResolve<Any, Context, Arguments, ResolveType> = { type in
 //            { context, _ in
 //                type[keyPath: keyPath]
 //            }

From 89eb9f2c9077ef8b439f83453685cf338bd81224 Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Tue, 23 Feb 2021 18:24:04 -0700
Subject: [PATCH 07/15] Completes working subscription test

---
 .../HelloWorldTests/HelloWorldTests.swift     | 59 ++++++++-----------
 1 file changed, 26 insertions(+), 33 deletions(-)

diff --git a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
index 7ab0dd3b..078abecb 100644
--- a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
+++ b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
@@ -4,7 +4,7 @@ import NIO
 import RxSwift
 @testable import Graphiti
 
-let pubsub = PublishSubject<User>()
+let pubsub = PublishSubject<Any>()
 
 struct ID : Codable {
     let id: String
@@ -101,7 +101,7 @@ struct HelloResolver {
         User(arguments.user)
     }
     
-    func subscribeUser(context: HelloContext, arguments: NoArguments) -> Observable<User> {
+    func subscribeUser(context: HelloContext, arguments: NoArguments) -> Observable<Any> {
         pubsub
     }
 }
@@ -344,8 +344,9 @@ class HelloWorldTests : XCTestCase {
         wait(for: [expectation], timeout: 10)
     }
     
-    // TODO Adjust this to expect an Observable when GraphQL is piped up correctly.
     func testSubscription() throws {
+        let disposeBag = DisposeBag()
+        
         let subscription = """
         subscription {
             subscribeUser {
@@ -357,42 +358,34 @@ class HelloWorldTests : XCTestCase {
         }
         """
         
-        let expected = GraphQLResult(data: [
-            "subscribeUser": [
-                "user": [
-                    "name": "John Doe",
-                    "id": "123"
-                ]
-            ]
-        ])
-        
-        let result = try api.subscribe(
+        let observable = try api.subscribe(
             request: subscription,
             context: api.context,
             on: group
-        ).wait()
-        print(result)
-//        let observable = try api.subscribe(
-//            request: subscription,
-//            context: api.context,
-//            on: group
-//        ).wait().observable!
+        ).wait().observable!
         
         let expectation = XCTestExpectation()
         
-//        var currentResult = GraphQLResult()
-//        let _ = observable.subscribe { event in
-//            event.element!.whenSuccess { result in
-//                currentResult = result
-//                expectation.fulfill()
-//            }
-//        }
-//
-//        pubsub.onNext(User(id: "124", name: "Jerry"))
-//
-//        wait(for: [expectation], timeout: 10)
-//
-//        XCTAssertEqual(currentResult, expected)
+        var currentResult = GraphQLResult()
+        let _ = observable.subscribe { event in
+            event.element!.whenSuccess { result in
+                currentResult = result
+                expectation.fulfill()
+            }
+        }.disposed(by: disposeBag)
+
+        pubsub.onNext(User(id: "124", name: "Jerry"))
+
+        wait(for: [expectation], timeout: 10)
+
+        XCTAssertEqual(currentResult, GraphQLResult(data: [
+            "subscribeUser": [
+                "user": [
+                    "name": "Jerry",
+                    "id": "124"
+                ]
+            ]
+        ]))
     }
 }
 

From 4b3bb1104898a018d6fda417aa0915e86fe34549 Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Tue, 23 Feb 2021 18:27:54 -0700
Subject: [PATCH 08/15] Minor note on observable limitations

---
 Sources/Graphiti/Subscription/SubscribeField.swift | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/Sources/Graphiti/Subscription/SubscribeField.swift b/Sources/Graphiti/Subscription/SubscribeField.swift
index 0400644b..7ca0ed3c 100644
--- a/Sources/Graphiti/Subscription/SubscribeField.swift
+++ b/Sources/Graphiti/Subscription/SubscribeField.swift
@@ -2,6 +2,8 @@ import GraphQL
 import Runtime
 import RxSwift
 
+// Subscription resolver must return an Observer<Any>, not a specific type, due to lack of support for covariance generics in Swift
+
 public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType, Arguments : Decodable> : FieldComponent<ObjectType, Context> {
     let name: String
     let arguments: [ArgumentComponent<Arguments>]

From 321df427f45814c0d53a3b8963033b2bacefb8fb Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Wed, 24 Feb 2021 10:28:16 -0700
Subject: [PATCH 09/15] Adjusts type name for clarity

---
 .../Subscription/SubscribeField.swift         | 36 +++++++++----------
 1 file changed, 18 insertions(+), 18 deletions(-)

diff --git a/Sources/Graphiti/Subscription/SubscribeField.swift b/Sources/Graphiti/Subscription/SubscribeField.swift
index 7ca0ed3c..487b642b 100644
--- a/Sources/Graphiti/Subscription/SubscribeField.swift
+++ b/Sources/Graphiti/Subscription/SubscribeField.swift
@@ -4,7 +4,7 @@ import RxSwift
 
 // Subscription resolver must return an Observer<Any>, not a specific type, due to lack of support for covariance generics in Swift
 
-public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType, Arguments : Decodable> : FieldComponent<ObjectType, Context> {
+public class SubscriptionField<ObjectType, Context, SourceEventType, FieldType, Arguments : Decodable> : FieldComponent<ObjectType, Context> {
     let name: String
     let arguments: [ArgumentComponent<Arguments>]
     let resolve: GraphQLFieldResolve
@@ -49,11 +49,11 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
     convenience init<ResolveType>(
         name: String,
         arguments: [ArgumentComponent<Arguments>],
-        asyncResolve: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        asyncResolve: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
     ) {
         let resolve: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
-            guard let s = source as? SubscriptionType else {
+            guard let s = source as? SourceEventType else {
                 throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
             }
             
@@ -83,10 +83,10 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
     convenience init<ResolveType>(
         name: String,
         arguments: [ArgumentComponent<Arguments>],
-        simpleAsyncResolve: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        simpleAsyncResolve: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
     ) {
-        let asyncResolve: AsyncResolve<SubscriptionType, Context, Arguments, ResolveType> = { type in
+        let asyncResolve: AsyncResolve<SourceEventType, Context, Arguments, ResolveType> = { type in
             { context, arguments, group in
                 // We hop to guarantee that the future will
                 // return in the same event loop group of the execution.
@@ -107,10 +107,10 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
     convenience init<ResolveType>(
         name: String,
         arguments: [ArgumentComponent<Arguments>],
-        syncResolve: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        syncResolve: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>
     ) {
-        let asyncResolve: AsyncResolve<SubscriptionType, Context, Arguments, ResolveType> = { type in
+        let asyncResolve: AsyncResolve<SourceEventType, Context, Arguments, ResolveType> = { type in
             { context, arguments, group in
                 let result = try syncResolve(type)(context, arguments)
                 return group.next().makeSucceededFuture(result)
@@ -132,7 +132,7 @@ public class SubscriptionField<ObjectType, Context, SubscriptionType, FieldType,
 public extension SubscriptionField where FieldType : Encodable {
     convenience init(
         _ name: String,
-        at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
@@ -152,7 +152,7 @@ public extension SubscriptionField where FieldType : Encodable {
 public extension SubscriptionField {
     convenience init<ResolveType>(
         _ name: String,
-        at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
         atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
@@ -162,7 +162,7 @@ public extension SubscriptionField {
     
     convenience init<ResolveType>(
         _ name: String,
-        at function: @escaping AsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
         atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
@@ -176,7 +176,7 @@ public extension SubscriptionField {
 public extension SubscriptionField where FieldType : Encodable {
     convenience init(
         _ name: String,
-        at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
@@ -185,7 +185,7 @@ public extension SubscriptionField where FieldType : Encodable {
 
     convenience init(
         _ name: String,
-        at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
@@ -196,7 +196,7 @@ public extension SubscriptionField where FieldType : Encodable {
 public extension SubscriptionField {
     convenience init<ResolveType>(
         _ name: String,
-        at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
         atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
@@ -206,7 +206,7 @@ public extension SubscriptionField {
 
     convenience init<ResolveType>(
         _ name: String,
-        at function: @escaping SimpleAsyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
         atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
@@ -220,7 +220,7 @@ public extension SubscriptionField {
 public extension SubscriptionField where FieldType : Encodable {
     convenience init(
         _ name: String,
-        at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        at function: @escaping SyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
@@ -229,7 +229,7 @@ public extension SubscriptionField where FieldType : Encodable {
 
     convenience init(
         _ name: String,
-        at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, FieldType>,
+        at function: @escaping SyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
@@ -240,7 +240,7 @@ public extension SubscriptionField where FieldType : Encodable {
 public extension SubscriptionField {
     convenience init<ResolveType>(
         _ name: String,
-        at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        at function: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
         atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
@@ -250,7 +250,7 @@ public extension SubscriptionField {
 
     convenience init<ResolveType>(
         _ name: String,
-        at function: @escaping SyncResolve<SubscriptionType, Context, Arguments, ResolveType>,
+        at function: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
         atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}

From e5307cdddee357092ed6180c6c5852ba5a77049e Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Thu, 25 Feb 2021 12:46:31 -0700
Subject: [PATCH 10/15] Adjusts subscription field initializers

This enhances support for simple subscriptions that don't need to be transformed from their original source event type
---
 .../Subscription/SubscribeField.swift         | 148 ++++++++++++++++--
 .../HelloWorldTests/HelloWorldTests.swift     |  71 +++++++--
 2 files changed, 189 insertions(+), 30 deletions(-)

diff --git a/Sources/Graphiti/Subscription/SubscribeField.swift b/Sources/Graphiti/Subscription/SubscribeField.swift
index 487b642b..a685ebbb 100644
--- a/Sources/Graphiti/Subscription/SubscribeField.swift
+++ b/Sources/Graphiti/Subscription/SubscribeField.swift
@@ -4,7 +4,7 @@ import RxSwift
 
 // Subscription resolver must return an Observer<Any>, not a specific type, due to lack of support for covariance generics in Swift
 
-public class SubscriptionField<ObjectType, Context, SourceEventType, FieldType, Arguments : Decodable> : FieldComponent<ObjectType, Context> {
+public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decodable> : FieldComponent<ObjectType, Context> {
     let name: String
     let arguments: [ArgumentComponent<Arguments>]
     let resolve: GraphQLFieldResolve
@@ -46,7 +46,7 @@ public class SubscriptionField<ObjectType, Context, SourceEventType, FieldType,
         self.subscribe = subscribe
     }
 
-    convenience init<ResolveType>(
+    convenience init<SourceEventType, ResolveType>(
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         asyncResolve: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
@@ -80,7 +80,36 @@ public class SubscriptionField<ObjectType, Context, SourceEventType, FieldType,
         self.init(name: name, arguments: arguments, resolve: resolve, subscribe: subscribe)
     }
     
-    convenience init<ResolveType>(
+    convenience init(
+        name: String,
+        arguments: [ArgumentComponent<Arguments>],
+        as: FieldType.Type,
+        asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+    ) {
+        let resolve: GraphQLFieldResolve = { source, _, context, eventLoopGroup, _ in
+            guard let s = source as? FieldType else {
+                throw GraphQLError(message: "Expected source type \(FieldType.self) but got \(type(of: source))")
+            }
+            
+            return eventLoopGroup.next().makeSucceededFuture(s)
+        }
+        
+        let subscribe: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
+            guard let s = source as? ObjectType else {
+                throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
+            }
+        
+            guard let c = context as? Context else {
+                throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
+            }
+        
+            let a = try MapDecoder().decode(Arguments.self, from: arguments)
+            return try asyncSubscribe(s)(c, a, eventLoopGroup).map({ $0 })
+        }
+        self.init(name: name, arguments: arguments, resolve: resolve, subscribe: subscribe)
+    }
+    
+    convenience init<SourceEventType, ResolveType>(
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         simpleAsyncResolve: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
@@ -104,7 +133,23 @@ public class SubscriptionField<ObjectType, Context, SourceEventType, FieldType,
         self.init(name: name, arguments: arguments, asyncResolve: asyncResolve, asyncSubscribe: asyncSubscribe)
     }
     
-    convenience init<ResolveType>(
+    convenience init(
+        name: String,
+        arguments: [ArgumentComponent<Arguments>],
+        as: FieldType.Type,
+        simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+    ) {
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<Any>> = { type in
+            { context, arguments, group in
+                // We hop to guarantee that the future will
+                // return in the same event loop group of the execution.
+                try simpleAsyncSubscribe(type)(context, arguments).hop(to: group.next())
+            }
+        }
+        self.init(name: name, arguments: arguments, as: `as`, asyncSubscribe: asyncSubscribe)
+    }
+    
+    convenience init<SourceEventType, ResolveType>(
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         syncResolve: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
@@ -125,12 +170,27 @@ public class SubscriptionField<ObjectType, Context, SourceEventType, FieldType,
         }
         self.init(name: name, arguments: arguments, asyncResolve: asyncResolve, asyncSubscribe: asyncSubscribe)
     }
+    
+    convenience init(
+        name: String,
+        arguments: [ArgumentComponent<Arguments>],
+        as: FieldType.Type,
+        syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+    ) {
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<Any>> = { type in
+            { context, arguments, group in
+                let result = try syncSubscribe(type)(context, arguments)
+                return group.next().makeSucceededFuture(result)
+            }
+        }
+        self.init(name: name, arguments: arguments, as: `as`, asyncSubscribe: asyncSubscribe)
+    }
 }
 
 // MARK: AsyncResolve Initializers
 
 public extension SubscriptionField where FieldType : Encodable {
-    convenience init(
+    convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
@@ -139,9 +199,9 @@ public extension SubscriptionField where FieldType : Encodable {
         self.init(name: name, arguments: [argument()], asyncResolve: function, asyncSubscribe: subFunc)
     }
     
-    convenience init(
+    convenience init<SourceEventType>(
         _ name: String,
-        at function: @escaping AsyncResolve<Any, Context, Arguments, FieldType>,
+        at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
@@ -150,7 +210,25 @@ public extension SubscriptionField where FieldType : Encodable {
 }
 
 public extension SubscriptionField {
-    convenience init<ResolveType>(
+    convenience init(
+        _ name: String,
+        as: FieldType.Type,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], as: `as`, asyncSubscribe: subFunc)
+    }
+    
+    convenience init(
+        _ name: String,
+        as: FieldType.Type,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), as: `as`, asyncSubscribe: subFunc)
+    }
+    
+    convenience init<SourceEventType, ResolveType>(
         _ name: String,
         at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
@@ -160,7 +238,7 @@ public extension SubscriptionField {
         self.init(name: name, arguments: [argument()], asyncResolve: function, asyncSubscribe: subFunc)
     }
     
-    convenience init<ResolveType>(
+    convenience init<SourceEventType, ResolveType>(
         _ name: String,
         at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
@@ -174,7 +252,7 @@ public extension SubscriptionField {
 // MARK: SimpleAsyncResolve Initializers
 
 public extension SubscriptionField where FieldType : Encodable {
-    convenience init(
+    convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
@@ -183,7 +261,7 @@ public extension SubscriptionField where FieldType : Encodable {
         self.init(name: name, arguments: [argument()], simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
     }
 
-    convenience init(
+    convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
@@ -194,7 +272,25 @@ public extension SubscriptionField where FieldType : Encodable {
 }
 
 public extension SubscriptionField {
-    convenience init<ResolveType>(
+    convenience init(
+        _ name: String,
+        as: FieldType.Type,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], as: `as`, simpleAsyncSubscribe: subFunc)
+    }
+    
+    convenience init(
+        _ name: String,
+        as: FieldType.Type,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), as: `as`, simpleAsyncSubscribe: subFunc)
+    }
+    
+    convenience init<SourceEventType, ResolveType>(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
@@ -204,7 +300,7 @@ public extension SubscriptionField {
         self.init(name: name, arguments: [argument()], simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
     }
 
-    convenience init<ResolveType>(
+    convenience init<SourceEventType, ResolveType>(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
@@ -218,7 +314,7 @@ public extension SubscriptionField {
 // MARK: SyncResolve Initializers
 
 public extension SubscriptionField where FieldType : Encodable {
-    convenience init(
+    convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping SyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
@@ -227,7 +323,7 @@ public extension SubscriptionField where FieldType : Encodable {
         self.init(name: name, arguments: [argument()], syncResolve: function, syncSubscribe: subFunc)
     }
 
-    convenience init(
+    convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping SyncResolve<SourceEventType, Context, Arguments, FieldType>,
         atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
@@ -238,7 +334,25 @@ public extension SubscriptionField where FieldType : Encodable {
 }
 
 public extension SubscriptionField {
-    convenience init<ResolveType>(
+    convenience init(
+        _ name: String,
+        as: FieldType.Type,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
+    ) {
+        self.init(name: name, arguments: [argument()], as: `as`, syncSubscribe: subFunc)
+    }
+    
+    convenience init(
+        _ name: String,
+        as: FieldType.Type,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
+    ) {
+        self.init(name: name, arguments: arguments(), as: `as`, syncSubscribe: subFunc)
+    }
+    
+    convenience init<SourceEventType, ResolveType>(
         _ name: String,
         at function: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
@@ -248,7 +362,7 @@ public extension SubscriptionField {
         self.init(name: name, arguments: [argument()], syncResolve: function, syncSubscribe: subFunc)
     }
 
-    convenience init<ResolveType>(
+    convenience init<SourceEventType, ResolveType>(
         _ name: String,
         at function: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
diff --git a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
index 078abecb..39883d13 100644
--- a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
+++ b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
@@ -37,6 +37,10 @@ struct User : Codable {
         self.id = input.id
         self.name = input.name
     }
+    
+    func toEvent(context: HelloContext, arguments: NoArguments) throws -> UserEvent {
+        return UserEvent(user: self)
+    }
 }
 
 struct UserInput : Codable {
@@ -54,12 +58,6 @@ final class HelloContext {
     }
 }
 
-extension User {
-    func toEvent(context: HelloContext, arguments: NoArguments) throws -> UserEvent {
-        return UserEvent(user: self)
-    }
-}
-
 struct HelloResolver {
     func hello(context: HelloContext, arguments: NoArguments) -> String {
         context.hello()
@@ -153,7 +151,8 @@ struct HelloAPI : API {
         }
         
         Subscription {
-            SubscriptionField("subscribeUser", at: User.toEvent, atSub: HelloResolver.subscribeUser)
+            SubscriptionField("subscribeUser", as: User.self, atSub: HelloResolver.subscribeUser)
+            SubscriptionField("subscribeUserEvent", at: User.toEvent, atSub: HelloResolver.subscribeUser)
         }
     }
 }
@@ -344,12 +343,56 @@ class HelloWorldTests : XCTestCase {
         wait(for: [expectation], timeout: 10)
     }
     
-    func testSubscription() throws {
+    /// Tests subscription when the sourceEventStream type matches the resolved type (i.e. the normal resolution function should just short-circuit to the sourceEventStream object)
+    func testSubscriptionSelf() throws {
         let disposeBag = DisposeBag()
         
-        let subscription = """
+        let request = """
         subscription {
             subscribeUser {
+                id
+                name
+            }
+        }
+        """
+        
+        let result = try api.subscribe(
+            request: request,
+            context: api.context,
+            on: group
+        ).wait()
+        
+        let observable = result.observable!
+        
+        let expectation = XCTestExpectation()
+        
+        var currentResult = GraphQLResult()
+        let _ = observable.subscribe { event in
+            event.element!.whenSuccess { result in
+                currentResult = result
+                expectation.fulfill()
+            }
+        }.disposed(by: disposeBag)
+
+        pubsub.onNext(User(id: "124", name: "Jerry"))
+
+        wait(for: [expectation], timeout: 10)
+
+        XCTAssertEqual(currentResult, GraphQLResult(data: [
+            "subscribeUser": [
+                "name": "Jerry",
+                "id": "124"
+            ]
+        ]))
+    }
+    
+    /// Tests subscription when the sourceEventStream type does not match the resolved type (i.e. there is a non-trivial resolution function that transforms the sourceEventStream object)
+    func testSubscriptionEvent() throws {
+        let disposeBag = DisposeBag()
+        
+        let request = """
+        subscription {
+            subscribeUserEvent {
                 user {
                     id
                     name
@@ -358,11 +401,13 @@ class HelloWorldTests : XCTestCase {
         }
         """
         
-        let observable = try api.subscribe(
-            request: subscription,
+        let result = try api.subscribe(
+            request: request,
             context: api.context,
             on: group
-        ).wait().observable!
+        ).wait()
+        
+        let observable = result.observable!
         
         let expectation = XCTestExpectation()
         
@@ -379,7 +424,7 @@ class HelloWorldTests : XCTestCase {
         wait(for: [expectation], timeout: 10)
 
         XCTAssertEqual(currentResult, GraphQLResult(data: [
-            "subscribeUser": [
+            "subscribeUserEvent": [
                 "user": [
                     "name": "Jerry",
                     "id": "124"

From 605c27ecb11ee0252c12b1d7be1d858af3c3bfa4 Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Mon, 1 Mar 2021 13:58:56 -0700
Subject: [PATCH 11/15] Improves variable naming following review

---
 .../Subscription/SubscribeField.swift         | 30 +++++++++----------
 .../HelloWorldTests/HelloWorldTests.swift     | 20 +++++++++----
 2 files changed, 29 insertions(+), 21 deletions(-)

diff --git a/Sources/Graphiti/Subscription/SubscribeField.swift b/Sources/Graphiti/Subscription/SubscribeField.swift
index a685ebbb..faa9f5b7 100644
--- a/Sources/Graphiti/Subscription/SubscribeField.swift
+++ b/Sources/Graphiti/Subscription/SubscribeField.swift
@@ -2,7 +2,7 @@ import GraphQL
 import Runtime
 import RxSwift
 
-// Subscription resolver must return an Observer<Any>, not a specific type, due to lack of support for covariance generics in Swift
+// Subscription resolver MUST return an Observer<Any>, not a specific type, due to lack of support for covariance generics in Swift
 
 public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decodable> : FieldComponent<ObjectType, Context> {
     let name: String
@@ -53,29 +53,29 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
     ) {
         let resolve: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
-            guard let s = source as? SourceEventType else {
+            guard let source = source as? SourceEventType else {
                 throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
             }
             
-            guard let c = context as? Context else {
+            guard let context = context as? Context else {
                 throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
             }
         
-            let a = try MapDecoder().decode(Arguments.self, from: arguments)
-            return try asyncResolve(s)(c, a, eventLoopGroup).map({ $0 })
+            let args = try MapDecoder().decode(Arguments.self, from: arguments)
+            return try asyncResolve(source)(context, args, eventLoopGroup).map({ $0 })
         }
         
         let subscribe: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
-            guard let s = source as? ObjectType else {
+            guard let source = source as? ObjectType else {
                 throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
             }
         
-            guard let c = context as? Context else {
+            guard let context = context as? Context else {
                 throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
             }
         
-            let a = try MapDecoder().decode(Arguments.self, from: arguments)
-            return try asyncSubscribe(s)(c, a, eventLoopGroup).map({ $0 })
+            let args = try MapDecoder().decode(Arguments.self, from: arguments)
+            return try asyncSubscribe(source)(context, args, eventLoopGroup).map({ $0 })
         }
         self.init(name: name, arguments: arguments, resolve: resolve, subscribe: subscribe)
     }
@@ -87,24 +87,24 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
     ) {
         let resolve: GraphQLFieldResolve = { source, _, context, eventLoopGroup, _ in
-            guard let s = source as? FieldType else {
+            guard let source = source as? FieldType else {
                 throw GraphQLError(message: "Expected source type \(FieldType.self) but got \(type(of: source))")
             }
             
-            return eventLoopGroup.next().makeSucceededFuture(s)
+            return eventLoopGroup.next().makeSucceededFuture(source)
         }
         
         let subscribe: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
-            guard let s = source as? ObjectType else {
+            guard let source = source as? ObjectType else {
                 throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
             }
         
-            guard let c = context as? Context else {
+            guard let context = context as? Context else {
                 throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
             }
         
-            let a = try MapDecoder().decode(Arguments.self, from: arguments)
-            return try asyncSubscribe(s)(c, a, eventLoopGroup).map({ $0 })
+            let args = try MapDecoder().decode(Arguments.self, from: arguments)
+            return try asyncSubscribe(source)(context, args, eventLoopGroup).map({ $0 })
         }
         self.init(name: name, arguments: arguments, resolve: resolve, subscribe: subscribe)
     }
diff --git a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
index 39883d13..80244aaf 100644
--- a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
+++ b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
@@ -356,22 +356,26 @@ class HelloWorldTests : XCTestCase {
         }
         """
         
-        let result = try api.subscribe(
+        let subResult = try api.subscribe(
             request: request,
             context: api.context,
             on: group
         ).wait()
         
-        let observable = result.observable!
+        let observable = subResult.observable!
         
         let expectation = XCTestExpectation()
         
         var currentResult = GraphQLResult()
         let _ = observable.subscribe { event in
-            event.element!.whenSuccess { result in
+            let resultFuture = event.element!
+            resultFuture.whenSuccess { result in
                 currentResult = result
                 expectation.fulfill()
             }
+            resultFuture.whenFailure { error in
+                XCTFail()
+            }
         }.disposed(by: disposeBag)
 
         pubsub.onNext(User(id: "124", name: "Jerry"))
@@ -401,22 +405,26 @@ class HelloWorldTests : XCTestCase {
         }
         """
         
-        let result = try api.subscribe(
+        let subResult = try api.subscribe(
             request: request,
             context: api.context,
             on: group
         ).wait()
         
-        let observable = result.observable!
+        let observable = subResult.observable!
         
         let expectation = XCTestExpectation()
         
         var currentResult = GraphQLResult()
         let _ = observable.subscribe { event in
-            event.element!.whenSuccess { result in
+            let resultFuture = event.element!
+            resultFuture.whenSuccess { result in
                 currentResult = result
                 expectation.fulfill()
             }
+            resultFuture.whenFailure { error in
+                XCTFail()
+            }
         }.disposed(by: disposeBag)
 
         pubsub.onNext(User(id: "124", name: "Jerry"))

From 66e641ef8ed3688d3702dc0689aa527d7f519c8e Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Wed, 10 Mar 2021 20:03:28 -0700
Subject: [PATCH 12/15] Moves all RxSwift dependencies to GraphQLRxSwift

---
 Package.swift                                 |   5 +-
 Sources/Graphiti/API/API.swift                |   1 -
 Sources/Graphiti/Schema/Schema.swift          |   1 -
 .../Subscription/SubscribeField.swift         |  57 +++++----
 .../HelloWorldTests/HelloWorldTests.swift     | 110 ------------------
 5 files changed, 30 insertions(+), 144 deletions(-)

diff --git a/Package.swift b/Package.swift
index 9b245755..99dba8c6 100644
--- a/Package.swift
+++ b/Package.swift
@@ -7,11 +7,10 @@ let package = Package(
         .library(name: "Graphiti", targets: ["Graphiti"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/GraphQLSwift/GraphQL.git", .upToNextMajor(from: "1.1.7")),
-        .package(url: "https://github.com/ReactiveX/RxSwift.git", .upToNextMajor(from: "6.1.0"))
+        .package(url: "https://github.com/GraphQLSwift/GraphQL.git", .upToNextMajor(from: "1.1.7"))
     ],
     targets: [
-        .target(name: "Graphiti", dependencies: ["GraphQL", "RxSwift"]),
+        .target(name: "Graphiti", dependencies: ["GraphQL"]),
         .testTarget(name: "GraphitiTests", dependencies: ["Graphiti"]),
     ]
 )
diff --git a/Sources/Graphiti/API/API.swift b/Sources/Graphiti/API/API.swift
index d8567850..9264b83b 100644
--- a/Sources/Graphiti/API/API.swift
+++ b/Sources/Graphiti/API/API.swift
@@ -1,6 +1,5 @@
 import GraphQL
 import NIO
-import RxSwift
 
 public protocol API {
     associatedtype Resolver
diff --git a/Sources/Graphiti/Schema/Schema.swift b/Sources/Graphiti/Schema/Schema.swift
index de013ada..0aece3b0 100644
--- a/Sources/Graphiti/Schema/Schema.swift
+++ b/Sources/Graphiti/Schema/Schema.swift
@@ -1,6 +1,5 @@
 import GraphQL
 import NIO
-import RxSwift
 
 public final class Schema<Resolver, Context> {
     public let schema: GraphQLSchema
diff --git a/Sources/Graphiti/Subscription/SubscribeField.swift b/Sources/Graphiti/Subscription/SubscribeField.swift
index faa9f5b7..fe74aded 100644
--- a/Sources/Graphiti/Subscription/SubscribeField.swift
+++ b/Sources/Graphiti/Subscription/SubscribeField.swift
@@ -1,6 +1,5 @@
 import GraphQL
 import Runtime
-import RxSwift
 
 // Subscription resolver MUST return an Observer<Any>, not a specific type, due to lack of support for covariance generics in Swift
 
@@ -50,7 +49,7 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         asyncResolve: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
-        asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+        asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>
     ) {
         let resolve: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
             guard let source = source as? SourceEventType else {
@@ -84,7 +83,7 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         as: FieldType.Type,
-        asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+        asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>
     ) {
         let resolve: GraphQLFieldResolve = { source, _, context, eventLoopGroup, _ in
             guard let source = source as? FieldType else {
@@ -113,7 +112,7 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         simpleAsyncResolve: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
-        simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+        simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>
     ) {
         let asyncResolve: AsyncResolve<SourceEventType, Context, Arguments, ResolveType> = { type in
             { context, arguments, group in
@@ -123,7 +122,7 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
             }
         }
         
-        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<Any>> = { type in
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>> = { type in
             { context, arguments, group in
                 // We hop to guarantee that the future will
                 // return in the same event loop group of the execution.
@@ -137,9 +136,9 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         as: FieldType.Type,
-        simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+        simpleAsyncSubscribe: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>
     ) {
-        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<Any>> = { type in
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>> = { type in
             { context, arguments, group in
                 // We hop to guarantee that the future will
                 // return in the same event loop group of the execution.
@@ -153,7 +152,7 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         syncResolve: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
-        syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+        syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, EventStream<Any>>
     ) {
         let asyncResolve: AsyncResolve<SourceEventType, Context, Arguments, ResolveType> = { type in
             { context, arguments, group in
@@ -162,7 +161,7 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
             }
         }
         
-        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<Any>> = { type in
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>> = { type in
             { context, arguments, group in
                 let result = try syncSubscribe(type)(context, arguments)
                 return group.next().makeSucceededFuture(result)
@@ -175,9 +174,9 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         name: String,
         arguments: [ArgumentComponent<Arguments>],
         as: FieldType.Type,
-        syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>
+        syncSubscribe: @escaping SyncResolve<ObjectType, Context, Arguments, EventStream<Any>>
     ) {
-        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, Observable<Any>> = { type in
+        let asyncSubscribe: AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>> = { type in
             { context, arguments, group in
                 let result = try syncSubscribe(type)(context, arguments)
                 return group.next().makeSucceededFuture(result)
@@ -193,7 +192,7 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], asyncResolve: function, asyncSubscribe: subFunc)
@@ -202,7 +201,7 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), asyncResolve: function, asyncSubscribe: subFunc)
@@ -213,7 +212,7 @@ public extension SubscriptionField {
     convenience init(
         _ name: String,
         as: FieldType.Type,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], as: `as`, asyncSubscribe: subFunc)
@@ -222,7 +221,7 @@ public extension SubscriptionField {
     convenience init(
         _ name: String,
         as: FieldType.Type,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), as: `as`, asyncSubscribe: subFunc)
@@ -232,7 +231,7 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], asyncResolve: function, asyncSubscribe: subFunc)
@@ -242,7 +241,7 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping AsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), asyncResolve: function, asyncSubscribe: subFunc)
@@ -255,7 +254,7 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
@@ -264,7 +263,7 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
@@ -275,7 +274,7 @@ public extension SubscriptionField {
     convenience init(
         _ name: String,
         as: FieldType.Type,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], as: `as`, simpleAsyncSubscribe: subFunc)
@@ -284,7 +283,7 @@ public extension SubscriptionField {
     convenience init(
         _ name: String,
         as: FieldType.Type,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), as: `as`, simpleAsyncSubscribe: subFunc)
@@ -294,7 +293,7 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
@@ -304,7 +303,7 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping SimpleAsyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SimpleAsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), simpleAsyncResolve: function, simpleAsyncSubscribe: subFunc)
@@ -317,7 +316,7 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping SyncResolve<SourceEventType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], syncResolve: function, syncSubscribe: subFunc)
@@ -326,7 +325,7 @@ public extension SubscriptionField where FieldType : Encodable {
     convenience init<SourceEventType>(
         _ name: String,
         at function: @escaping SyncResolve<SourceEventType, Context, Arguments, FieldType>,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), syncResolve: function, syncSubscribe: subFunc)
@@ -337,7 +336,7 @@ public extension SubscriptionField {
     convenience init(
         _ name: String,
         as: FieldType.Type,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], as: `as`, syncSubscribe: subFunc)
@@ -346,7 +345,7 @@ public extension SubscriptionField {
     convenience init(
         _ name: String,
         as: FieldType.Type,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), as: `as`, syncSubscribe: subFunc)
@@ -356,7 +355,7 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ argument: () -> ArgumentComponent<Arguments>
     ) {
         self.init(name: name, arguments: [argument()], syncResolve: function, syncSubscribe: subFunc)
@@ -366,7 +365,7 @@ public extension SubscriptionField {
         _ name: String,
         at function: @escaping SyncResolve<SourceEventType, Context, Arguments, ResolveType>,
         as: FieldType.Type,
-        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, Observable<Any>>,
+        atSub subFunc: @escaping SyncResolve<ObjectType, Context, Arguments, EventStream<Any>>,
         @ArgumentComponentBuilder<Arguments> _ arguments: () -> [ArgumentComponent<Arguments>] = {[]}
     ) {
         self.init(name: name, arguments: arguments(), syncResolve: function, syncSubscribe: subFunc)
diff --git a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
index 80244aaf..cfe4fbeb 100644
--- a/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
+++ b/Tests/GraphitiTests/HelloWorldTests/HelloWorldTests.swift
@@ -1,11 +1,8 @@
 import XCTest
 import GraphQL
 import NIO
-import RxSwift
 @testable import Graphiti
 
-let pubsub = PublishSubject<Any>()
-
 struct ID : Codable {
     let id: String
     
@@ -98,10 +95,6 @@ struct HelloResolver {
     func addUser(context: HelloContext, arguments: AddUserArguments) -> User {
         User(arguments.user)
     }
-    
-    func subscribeUser(context: HelloContext, arguments: NoArguments) -> Observable<Any> {
-        pubsub
-    }
 }
 
 struct HelloAPI : API {
@@ -149,11 +142,6 @@ struct HelloAPI : API {
                 Argument("user", at: \.user)
             }
         }
-        
-        Subscription {
-            SubscriptionField("subscribeUser", as: User.self, atSub: HelloResolver.subscribeUser)
-            SubscriptionField("subscribeUserEvent", at: User.toEvent, atSub: HelloResolver.subscribeUser)
-        }
     }
 }
 
@@ -342,104 +330,6 @@ class HelloWorldTests : XCTestCase {
         
         wait(for: [expectation], timeout: 10)
     }
-    
-    /// Tests subscription when the sourceEventStream type matches the resolved type (i.e. the normal resolution function should just short-circuit to the sourceEventStream object)
-    func testSubscriptionSelf() throws {
-        let disposeBag = DisposeBag()
-        
-        let request = """
-        subscription {
-            subscribeUser {
-                id
-                name
-            }
-        }
-        """
-        
-        let subResult = try api.subscribe(
-            request: request,
-            context: api.context,
-            on: group
-        ).wait()
-        
-        let observable = subResult.observable!
-        
-        let expectation = XCTestExpectation()
-        
-        var currentResult = GraphQLResult()
-        let _ = observable.subscribe { event in
-            let resultFuture = event.element!
-            resultFuture.whenSuccess { result in
-                currentResult = result
-                expectation.fulfill()
-            }
-            resultFuture.whenFailure { error in
-                XCTFail()
-            }
-        }.disposed(by: disposeBag)
-
-        pubsub.onNext(User(id: "124", name: "Jerry"))
-
-        wait(for: [expectation], timeout: 10)
-
-        XCTAssertEqual(currentResult, GraphQLResult(data: [
-            "subscribeUser": [
-                "name": "Jerry",
-                "id": "124"
-            ]
-        ]))
-    }
-    
-    /// Tests subscription when the sourceEventStream type does not match the resolved type (i.e. there is a non-trivial resolution function that transforms the sourceEventStream object)
-    func testSubscriptionEvent() throws {
-        let disposeBag = DisposeBag()
-        
-        let request = """
-        subscription {
-            subscribeUserEvent {
-                user {
-                    id
-                    name
-                }
-            }
-        }
-        """
-        
-        let subResult = try api.subscribe(
-            request: request,
-            context: api.context,
-            on: group
-        ).wait()
-        
-        let observable = subResult.observable!
-        
-        let expectation = XCTestExpectation()
-        
-        var currentResult = GraphQLResult()
-        let _ = observable.subscribe { event in
-            let resultFuture = event.element!
-            resultFuture.whenSuccess { result in
-                currentResult = result
-                expectation.fulfill()
-            }
-            resultFuture.whenFailure { error in
-                XCTFail()
-            }
-        }.disposed(by: disposeBag)
-
-        pubsub.onNext(User(id: "124", name: "Jerry"))
-
-        wait(for: [expectation], timeout: 10)
-
-        XCTAssertEqual(currentResult, GraphQLResult(data: [
-            "subscribeUserEvent": [
-                "user": [
-                    "name": "Jerry",
-                    "id": "124"
-                ]
-            ]
-        ]))
-    }
 }
 
 extension HelloWorldTests {

From 30120d35fc67d750fb7e664295aacf4e958d708b Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Wed, 10 Mar 2021 20:04:05 -0700
Subject: [PATCH 13/15] Adds subscription documentation to readme

---
 README.md | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/README.md b/README.md
index a85838de..21f3f992 100644
--- a/README.md
+++ b/README.md
@@ -152,6 +152,16 @@ struct Resolver {
 }
 ```
 
+#### Subscription
+
+This library supports GraphQL subscriptions. To use them, you must create a concrete subclass of the `EventStream` class that implements event streaming
+functionality.
+
+If you don't feel like creating a subclass yourself, you can use the [GraphQLRxSwift](https://github.com/GraphQLSwift/GraphQLRxSwift) repository
+to integrate [RxSwift](https://github.com/ReactiveX/RxSwift) observables out-of-the-box. Or you can use that repository as a reference to connect a different 
+stream library like [ReactiveSwift](https://github.com/ReactiveCocoa/ReactiveSwift), [OpenCombine](https://github.com/OpenCombine/OpenCombine), or
+one that you've created yourself.
+
 ## Star Wars API example
 
 Check the [Star Wars API](Tests/GraphitiTests/StarWarsAPI/StarWarsAPI.swift) for a more complete example.

From b7a30f72af58235233d50027fc46a1d6076a076b Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Mon, 5 Apr 2021 19:34:52 -0600
Subject: [PATCH 14/15] Updates GraphQL version dependency

---
 Package.resolved | 4 ++--
 Package.swift    | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/Package.resolved b/Package.resolved
index 74e6ce22..70bd15d7 100644
--- a/Package.resolved
+++ b/Package.resolved
@@ -15,8 +15,8 @@
         "repositoryURL": "https://github.com/GraphQLSwift/GraphQL.git",
         "state": {
           "branch": null,
-          "revision": "69f2cd4835ff60b5fae841a680abdc1d56592dd4",
-          "version": "1.1.7"
+          "revision": "abf41c20c79331444ee3a290373d7c647c244383",
+          "version": "1.2.0"
         }
       },
       {
diff --git a/Package.swift b/Package.swift
index 45856d2e..270d8a5b 100644
--- a/Package.swift
+++ b/Package.swift
@@ -7,7 +7,7 @@ let package = Package(
         .library(name: "Graphiti", targets: ["Graphiti"]),
     ],
     dependencies: [
-        .package(url: "https://github.com/GraphQLSwift/GraphQL.git", .upToNextMajor(from: "1.1.8"))
+        .package(url: "https://github.com/GraphQLSwift/GraphQL.git", .upToNextMajor(from: "1.2.0"))
     ],
     targets: [
         .target(name: "Graphiti", dependencies: ["GraphQL"]),

From 9813c7516dcad730a20ee8d36a96520fa01e8cbf Mon Sep 17 00:00:00 2001
From: Jay Herron <jay@passivelogic.com>
Date: Wed, 14 Apr 2021 14:26:32 -0600
Subject: [PATCH 15/15] Fixes guard renaming issues

---
 .../Subscription/SubscribeField.swift         | 22 +++++++++----------
 1 file changed, 11 insertions(+), 11 deletions(-)

diff --git a/Sources/Graphiti/Subscription/SubscribeField.swift b/Sources/Graphiti/Subscription/SubscribeField.swift
index fe74aded..1dc47112 100644
--- a/Sources/Graphiti/Subscription/SubscribeField.swift
+++ b/Sources/Graphiti/Subscription/SubscribeField.swift
@@ -52,29 +52,29 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>
     ) {
         let resolve: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
-            guard let source = source as? SourceEventType else {
+            guard let _source = source as? SourceEventType else {
                 throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
             }
             
-            guard let context = context as? Context else {
+            guard let _context = context as? Context else {
                 throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
             }
         
             let args = try MapDecoder().decode(Arguments.self, from: arguments)
-            return try asyncResolve(source)(context, args, eventLoopGroup).map({ $0 })
+            return try asyncResolve(_source)(_context, args, eventLoopGroup).map({ $0 })
         }
         
         let subscribe: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
-            guard let source = source as? ObjectType else {
+            guard let _source = source as? ObjectType else {
                 throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
             }
         
-            guard let context = context as? Context else {
+            guard let _context = context as? Context else {
                 throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
             }
         
             let args = try MapDecoder().decode(Arguments.self, from: arguments)
-            return try asyncSubscribe(source)(context, args, eventLoopGroup).map({ $0 })
+            return try asyncSubscribe(_source)(_context, args, eventLoopGroup).map({ $0 })
         }
         self.init(name: name, arguments: arguments, resolve: resolve, subscribe: subscribe)
     }
@@ -86,24 +86,24 @@ public class SubscriptionField<ObjectType, Context, FieldType, Arguments : Decod
         asyncSubscribe: @escaping AsyncResolve<ObjectType, Context, Arguments, EventStream<Any>>
     ) {
         let resolve: GraphQLFieldResolve = { source, _, context, eventLoopGroup, _ in
-            guard let source = source as? FieldType else {
+            guard let _source = source as? FieldType else {
                 throw GraphQLError(message: "Expected source type \(FieldType.self) but got \(type(of: source))")
             }
             
-            return eventLoopGroup.next().makeSucceededFuture(source)
+            return eventLoopGroup.next().makeSucceededFuture(_source)
         }
         
         let subscribe: GraphQLFieldResolve = { source, arguments, context, eventLoopGroup, _ in
-            guard let source = source as? ObjectType else {
+            guard let _source = source as? ObjectType else {
                 throw GraphQLError(message: "Expected source type \(ObjectType.self) but got \(type(of: source))")
             }
         
-            guard let context = context as? Context else {
+            guard let _context = context as? Context else {
                 throw GraphQLError(message: "Expected context type \(Context.self) but got \(type(of: context))")
             }
         
             let args = try MapDecoder().decode(Arguments.self, from: arguments)
-            return try asyncSubscribe(source)(context, args, eventLoopGroup).map({ $0 })
+            return try asyncSubscribe(_source)(_context, args, eventLoopGroup).map({ $0 })
         }
         self.init(name: name, arguments: arguments, resolve: resolve, subscribe: subscribe)
     }