Repository: broadwaylamb/OpenCombine (swift)

Issue:
Missing `Future` publisher
I've tried to port [GRDBCombine](https://github.com/groue/GRDBCombine) to OpenCombine and it seems like the biggest missing piece there is a lack of the [`Future`](https://developer.apple.com/documentation/combine/future) publisher. Just adding it as a separate issue here for now to hopefully bump its priority on a possible OpenCombine roadmap 🙂

Pull request:
Resolves #105

Patch:
diff --git a/RemainingCombineInterface.swift b/RemainingCombineInterface.swift
index a8e85e4c2..413ca6dbf 100644
--- a/RemainingCombineInterface.swift
+++ b/RemainingCombineInterface.swift
@@ -328,22 +328,6 @@ extension Publisher {
     public func combineLatest<P, Q, R, T>(_ publisher1: P, _ publisher2: Q, _ publisher3: R, _ transform: @escaping (Self.Output, P.Output, Q.Output, R.Output) -> T) -> Publishers.Map<Publishers.CombineLatest4<Self, P, Q, R>, T> where P : Publisher, Q : Publisher, R : Publisher, Self.Failure == P.Failure, P.Failure == Q.Failure, Q.Failure == R.Failure
 }
 
-/// A publisher that eventually produces one value and then finishes or fails.
-final public class Future<Output, Failure> : Publisher where Failure : Error {
-
-    public typealias Promise = (Result<Output, Failure>) -> Void
-
-    public init(_ attemptToFulfill: @escaping (@escaping Future<Output, Failure>.Promise) -> Void)
-
-    /// This function is called to attach the specified `Subscriber` to this `Publisher` by `subscribe(_:)`
-    ///
-    /// - SeeAlso: `subscribe(_:)`
-    /// - Parameters:
-    ///     - subscriber: The subscriber to attach to this `Publisher`.
-    ///                   once attached it can begin to receive values.
-    final public func receive<S>(subscriber: S) where Output == S.Input, Failure == S.Failure, S : Subscriber
-}
-
 extension Publishers {
 
     /// A strategy for collecting received elements.
diff --git a/Sources/OpenCombine/Future.swift b/Sources/OpenCombine/Future.swift
new file mode 100644
index 000000000..7903509cf
--- /dev/null
+++ b/Sources/OpenCombine/Future.swift
@@ -0,0 +1,120 @@
+//
+//  Future.swift
+//  
+//
+//  Created by Max Desiatov on 24/11/2019.
+//
+
+import COpenCombineHelpers
+
+/// A publisher that eventually produces one value and then finishes or fails.
+public final class Future<Output, Failure>: Publisher where Failure: Error {
+
+    public typealias Promise = (Result<Output, Failure>) -> Void
+
+    private let _lock = UnfairRecursiveLock.allocate()
+    private var _subscriptions: [Conduit] = []
+
+    private var result: Result<Output, Failure>?
+
+    public init(
+        _ attemptToFulfill: @escaping (@escaping Promise) -> Void
+    ) {
+        attemptToFulfill { result in
+            self._lock.do {
+                guard self.result == nil else { return }
+                self.result = result
+                self._publish(result)
+            }
+        }
+    }
+
+    deinit {
+        _lock.deallocate()
+    }
+
+    /// This function is called to attach the specified `Subscriber` to this
+    /// `Publisher` by `subscribe(_:)`
+    ///
+    /// - SeeAlso: `subscribe(_:)`
+    /// - Parameters:
+    ///     - subscriber: The subscriber to attach to this `Publisher`.
+    ///                   once attached it can begin to receive values.
+    public func receive<Downstream: Subscriber>(
+        subscriber: Downstream
+    ) where Output == Downstream.Input, Failure == Downstream.Failure {
+        let subscription = Conduit(parent: self,
+                                   downstream: AnySubscriber(subscriber))
+
+        _subscriptions.append(subscription)
+       subscriber.receive(subscription: subscription)
+    }
+
+    private func _acknowledgeDownstreamDemand() {
+        _lock.do {
+            guard let result = result else { return }
+            _publish(result)
+        }
+    }
+
+    private func _publish(_ result: Result<Output, Failure>) {
+        for subscription in self._subscriptions where !subscription._isCompleted {
+            switch result {
+            case let .success(output) where subscription._demand > 0:
+                subscription._demand -= 1
+                subscription._demand += subscription._downstream?.receive(output) ?? .none
+                subscription._receive(completion: .finished)
+            case let .failure(error):
+                subscription._receive(completion: .failure(error))
+
+            // nothing to do if no demand
+            default: ()
+            }
+        }
+    }
+}
+
+extension Future {
+
+    fileprivate final class Conduit: Subscription {
+
+        fileprivate var _parent: Future<Output, Failure>?
+
+        fileprivate var _downstream: AnySubscriber<Output, Failure>?
+
+        fileprivate var _demand: Subscribers.Demand = .none
+
+        fileprivate var _isCompleted: Bool {
+            return _parent == nil
+        }
+
+        fileprivate init(parent: Future<Output, Failure>,
+                         downstream: AnySubscriber<Output, Failure>) {
+            _parent = parent
+            _downstream = downstream
+        }
+
+        fileprivate func _receive(completion: Subscribers.Completion<Failure>) {
+            if !_isCompleted {
+                _parent = nil
+                _downstream?.receive(completion: completion)
+            }
+        }
+
+        fileprivate func request(_ demand: Subscribers.Demand) {
+            demand.assertNonZero()
+            _parent?._lock.do {
+                _demand += demand
+            }
+            _parent?._acknowledgeDownstreamDemand()
+        }
+
+        fileprivate func cancel() {
+            _parent = nil
+        }
+    }
+}
+
+extension Future.Conduit: CustomStringConvertible {
+    fileprivate var description: String { return "Future" }
+}


Test patch:
diff --git a/Tests/OpenCombineTests/PublisherTests/FutureTests.swift b/Tests/OpenCombineTests/PublisherTests/FutureTests.swift
new file mode 100644
index 000000000..09f3d5ea6
--- /dev/null
+++ b/Tests/OpenCombineTests/PublisherTests/FutureTests.swift
@@ -0,0 +1,207 @@
+//
+//  FutureTests.swift
+//  
+//
+//  Created by Max Desiatov on 24/11/2019.
+//
+
+import XCTest
+
+#if OPENCOMBINE_COMPATIBILITY_TEST
+import Combine
+#else
+import OpenCombine
+#endif
+
+@available(macOS 10.15, iOS 13.0, *)
+final class FutureTests: XCTestCase {
+    private typealias Sut = Future<Int, TestingError>
+
+    func testFutureSuccess() {
+        var promise: Sut.Promise?
+
+        let future = Sut { promise = $0 }
+
+        let subscriber = TrackingSubscriber(receiveSubscription: { subscription in
+            subscription.request(.unlimited)
+        })
+        future.subscribe(subscriber)
+
+        promise?(.success(42))
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future"),
+            .value(42),
+            .completion(.finished)
+        ])
+    }
+
+    func testFutureFailure() {
+        var promise: Sut.Promise?
+
+        let future = Sut { promise = $0 }
+
+        let subscriber = TrackingSubscriber(
+            receiveSubscription: { subscription in
+                subscription.request(.unlimited)
+            }, receiveValue: { _ in
+                XCTFail("no value should be returned")
+                return .none
+            }
+        )
+        future.subscribe(subscriber)
+
+        let error = TestingError(description: "\(#function)")
+        promise?(.failure(error))
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future"),
+            .completion(.failure(error))
+        ])
+    }
+
+    func testResolvingMultipleTimes() {
+        var promise: Sut.Promise?
+
+        let future = Sut { promise = $0 }
+
+        let subscriber = TrackingSubscriber(receiveSubscription: { subscription in
+            subscription.request(.unlimited)
+        })
+        future.subscribe(subscriber)
+
+        promise?(.success(42))
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future"),
+            .value(42),
+            .completion(.finished)
+        ])
+
+        promise?(.success(41))
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future"),
+            .value(42),
+            .completion(.finished)
+        ])
+    }
+
+    func testCancellation() {
+        var promise: Sut.Promise?
+
+        let future = Sut { promise = $0 }
+
+        let subscriber = TrackingSubscriber(receiveSubscription: { subscription in
+            subscription.request(.unlimited)
+        })
+        future.subscribe(subscriber)
+
+        subscriber.subscriptions.forEach { $0.cancel() }
+
+        promise?(.success(42))
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future")
+        ])
+    }
+
+    func testSubscribeAfterResolution() {
+        var promise: Sut.Promise?
+
+        let future = Sut { promise = $0 }
+        promise?(.success(42))
+
+        let subscriber = TrackingSubscriber(receiveSubscription: { subscription in
+            subscription.request(.unlimited)
+        })
+        future.subscribe(subscriber)
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future"),
+            .value(42),
+            .completion(.finished)
+        ])
+    }
+
+    func testCrashesOnZeroDemand() {
+        let future = Sut { _ in }
+
+        let subscriber = TrackingSubscriber(receiveSubscription: { subscription in
+            self.assertCrashes {
+                subscription.request(.none)
+            }
+        })
+        future.subscribe(subscriber)
+    }
+
+    func testValueRecursion() {
+        var promise: Sut.Promise?
+        let future = Sut { promise = $0 }
+
+        let subscriber = TrackingSubscriber(receiveSubscription: { subscription in
+            subscription.request(.unlimited)
+        }, receiveValue: {
+            promise?(.success($0 + 1))
+            return .none
+        })
+        future.subscribe(subscriber)
+
+        promise?(.success(0))
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future"),
+            .value(0),
+            .completion(.finished)
+        ])
+    }
+
+    func testFinishedRecursion() {
+        var promise: Sut.Promise?
+        let future = Sut { promise = $0 }
+
+        let subscriber = TrackingSubscriber(receiveSubscription: { subscription in
+            subscription.request(.unlimited)
+        }, receiveCompletion: {
+            guard case .finished = $0 else {
+                XCTFail("future in \(#function) is not expected to fail")
+                return
+            }
+            promise?(.success(42))
+        })
+        future.subscribe(subscriber)
+
+        promise?(.success(0))
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future"),
+            .value(0),
+            .completion(.finished)
+        ])
+    }
+
+    func testFailureRecursion() {
+        var promise: Sut.Promise?
+        let future = Sut { promise = $0 }
+
+        let subscriber = TrackingSubscriber(receiveSubscription: { subscription in
+            subscription.request(.unlimited)
+        }, receiveCompletion: {
+            guard case .failure = $0 else {
+                XCTFail("future in \(#function) is expected to fail")
+                return
+            }
+            promise?(.success(42))
+        })
+        future.subscribe(subscriber)
+
+        let error = TestingError(description: "\(#function)")
+
+        promise?(.failure(error))
+
+        XCTAssertEqual(subscriber.history, [
+            .subscription("Future"),
+            .completion(.failure(error))
+        ])
+    }
+}