Adds wrapper class for Observables, generalizes

Author: Jay Herron

Sources/GraphQL/GraphQL.swift

@@ -42,16 +42,16 @@ public struct GraphQLResult : Equatable, Codable, CustomStringConvertible {
 
 /// SubscriptionResult wraps the observable and error data returned by the subscribe request.
 public struct SubscriptionResult {
-    public let observable: SubscriptionObservable?
+    public let stream: SubscriptionEventStream?
     public let errors: [GraphQLError]
 
-    public init(observable: SubscriptionObservable? = nil, errors: [GraphQLError] = []) {
-        self.observable = observable
+    public init(stream: SubscriptionEventStream? = nil, errors: [GraphQLError] = []) {
+        self.stream = stream
         self.errors = errors
     }
 }
 /// SubscriptionObservable represents an event stream of fully resolved GraphQL subscription results. Subscribers can be added to this stream.
-public typealias SubscriptionObservable = Observable<Future<GraphQLResult>>
+public typealias SubscriptionEventStream = EventStream<Future<GraphQLResult>>
 
 /// This is the primary entry point function for fulfilling GraphQL operations
 /// by parsing, validating, and executing a GraphQL document along side a

Sources/GraphQL/Subscription/EventStream.swift

@@ -0,0 +1,26 @@
+// Copyright (c) 2021 PassiveLogic, Inc.
+
+import RxSwift
+
+public class EventStream<Element> {
+    /// This class should be overridden
+    func transform<To>(_ closure: @escaping (Element) throws -> To) -> EventStream<To> {
+        fatalError("This function should be overridden by implementing classes")
+    }
+}
+
+//extension Observable: EventStream<Element> {
+//    func transform<To>(_ closure: @escaping (Element) throws -> To) -> EventStream<To> {
+//        return self.map(closure)
+//    }
+//}
+
+public class ObservableEventStream<Element> : EventStream<Element> {
+    var observable: Observable<Element>
+    init(observable: Observable<Element>) {
+        self.observable = observable
+    }
+    override func transform<To>(_ closure: @escaping (Element) throws -> To) -> EventStream<To> {
+        return ObservableEventStream<To>(observable: observable.map(closure))
+    }
+}

Sources/GraphQL/Subscription/Subscribe.swift

@@ -52,8 +52,8 @@ func subscribe(
     )
 
     return sourceFuture.map{ sourceResult -> SubscriptionResult in
-        if let sourceObservable = sourceResult.observable {
-            let subscriptionObservable = sourceObservable.map { eventPayload -> Future<GraphQLResult> in
+        if let sourceStream = sourceResult.stream {
+            let subscriptionStream = sourceStream.transform { eventPayload -> Future<GraphQLResult> in
 
                 // For each payload yielded from a subscription, map it over the normal
                 // GraphQL `execute` function, with `payload` as the rootValue.
@@ -75,7 +75,7 @@ func subscribe(
                     operationName: operationName
                 )
             }
-            return SubscriptionResult(observable: subscriptionObservable, errors: sourceResult.errors)
+            return SubscriptionResult(stream: subscriptionStream, errors: sourceResult.errors)
         } else {
             return SubscriptionResult(errors: sourceResult.errors)
         }
@@ -251,8 +251,8 @@ func executeSubscription(
             return SourceEventStreamResult(errors: context.errors)
         } else if let error = resolved as? GraphQLError {
             return SourceEventStreamResult(errors: [error])
-        } else if let observable = resolved as? SourceEventObservable {
-            return SourceEventStreamResult(observable: observable)
+        } else if let stream = resolved as? EventStream<Any> {
+            return SourceEventStreamResult(stream: stream)
         } else if resolved == nil {
             return SourceEventStreamResult(errors: [
                 GraphQLError(message: "Resolved subscription was nil")
@@ -261,23 +261,21 @@ func executeSubscription(
             let resolvedObj = resolved as AnyObject
             return SourceEventStreamResult(errors: [
                 GraphQLError(
-                    message: "Subscription field resolver must return SourceEventObservable. Received: '\(resolvedObj)'"
+                    message: "Subscription field resolver must return EventStream<Any>. Received: '\(resolvedObj)'"
                 )
             ])
         }
     }
 }
 
+// Subscription resolvers MUST return observables that are declared as 'Any' due to Swift not having covariant generic support for type
+// checking. Normal resolvers for subscription fields should handle type casting, same as resolvers for query fields.
 struct SourceEventStreamResult {
-    public let observable: SourceEventObservable?
+    public let stream: EventStream<Any>?
     public let errors: [GraphQLError]
 
-    public init(observable: SourceEventObservable? = nil, errors: [GraphQLError] = []) {
-        self.observable = observable
+    public init(stream: EventStream<Any>? = nil, errors: [GraphQLError] = []) {
+        self.stream = stream
         self.errors = errors
     }
 }
-
-// Subscription resolvers MUST return observables that are declared as 'Any' due to Swift not having covariant generic support for type
-// checking. Normal resolvers for subscription fields should handle type casting, same as resolvers for query fields.
-typealias SourceEventObservable = Observable<Any>

Tests/GraphQLTests/Subscription/SubscriptionTests.swift

@@ -32,11 +32,11 @@ class SubscriptionTests : XCTestCase {
             request: query,
             eventLoopGroup: eventLoopGroup
         ).wait()
-        
-        let observable = subscriptionResult.observable!
+        print(subscriptionResult)
+        let subscription = subscriptionResult.stream! as! ObservableEventStream<Future<GraphQLResult>>
 
         var currentResult = GraphQLResult()
-        let _ = observable.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             currentResult = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
 
@@ -123,10 +123,10 @@ class SubscriptionTests : XCTestCase {
                   }
                 }
               }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
         var currentResult = GraphQLResult()
-        let _ = subscription.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             currentResult = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
 
@@ -193,9 +193,9 @@ class SubscriptionTests : XCTestCase {
                 importantEmail
                 notImportantEmail
             }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
-        let _ = subscription.subscribe{ event in
+        let _ = subscription.observable.subscribe{ event in
             let _ = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
         db.trigger(email: Email(
@@ -261,7 +261,7 @@ class SubscriptionTests : XCTestCase {
             let graphQLError = error as! GraphQLError
             XCTAssertEqual(
                 graphQLError.message,
-                "Subscription field resolver must return SourceEventObservable. Received: 'test'"
+                "Subscription field resolver must return EventStream<Any>. Received: 'test'"
             )
         }
     }
@@ -360,10 +360,10 @@ class SubscriptionTests : XCTestCase {
                   }
                 }
               }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
         var currentResult = GraphQLResult()
-        let _ = subscription.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             currentResult = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
 
@@ -403,17 +403,17 @@ class SubscriptionTests : XCTestCase {
                   }
                 }
               }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
         // Subscription 1
         var sub1Value = GraphQLResult()
-        let _ = subscription.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             sub1Value = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
 
         // Subscription 2
         var sub2Value = GraphQLResult()
-        let _ = subscription.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             sub2Value = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
 
@@ -457,10 +457,10 @@ class SubscriptionTests : XCTestCase {
                   }
                 }
               }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
         var currentResult = GraphQLResult()
-        let _ = subscription.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             currentResult = try! event.element!.wait()
             print(currentResult)
         }.disposed(by: db.disposeBag)
@@ -521,10 +521,10 @@ class SubscriptionTests : XCTestCase {
                   }
                 }
               }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
         var currentResult = GraphQLResult()
-        let _ = subscription.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             currentResult = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
 
@@ -597,10 +597,10 @@ class SubscriptionTests : XCTestCase {
                   }
                 }
               }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
         var currentResult = GraphQLResult()
-        let subscriber = subscription.subscribe { event in
+        let subscriber = subscription.observable.subscribe { event in
             currentResult = try! event.element!.wait()
         }
 
@@ -671,10 +671,10 @@ class SubscriptionTests : XCTestCase {
                     }
                 }
             }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
         var currentResult = GraphQLResult()
-        let _ = subscription.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             currentResult = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
 
@@ -744,10 +744,10 @@ class SubscriptionTests : XCTestCase {
                   }
                 }
               }
-        """)
+        """) as! ObservableEventStream<Future<GraphQLResult>>
 
         var currentResult = GraphQLResult()
-        let _ = subscription.subscribe { event in
+        let _ = subscription.observable.subscribe { event in
             currentResult = try! event.element!.wait()
         }.disposed(by: db.disposeBag)
 
@@ -905,7 +905,7 @@ class EmailDb {
     func subscription (
         query:String,
         variableValues: [String: Map] = [:]
-    ) throws -> SubscriptionObservable {
+    ) throws -> SubscriptionEventStream {
         return try createSubscription(schema: defaultSchema(), query: query, variableValues: variableValues)
     }
 }
@@ -937,7 +937,7 @@ private func createSubscription(
     schema: GraphQLSchema,
     query: String,
     variableValues: [String: Map] = [:]
-) throws -> SubscriptionObservable {
+) throws -> SubscriptionEventStream {
     let document = try parse(source: query)
     let result = try subscribe(
         queryStrategy: SerialFieldExecutionStrategy(),
@@ -953,8 +953,8 @@ private func createSubscription(
         operationName: nil
     ).wait()
 
-    if let observable = result.observable {
-        return observable
+    if let stream = result.stream {
+        return stream
     } else {
         throw result.errors.first! // We may have more than one...
     }