Add withLatestFrom operator

Author: serhiybutz

README.md

@@ -18,6 +18,8 @@ This repo contains XCombine, a Swift module, developed on top of the Combine fra
   - [Example of Usage](#example-of-usage)
 - [ShareReplay Operator](#sharereplay-operator)
   - [Example of Usage](#example-of-usage-sharereplay)
+- [WithLatestFrom Operator](#withlatestfrom-operator)
+  - [Example of Usage](#example-of-usage-withlatestfrom)
 - [License](#license)
 
 ## Installation
@@ -136,6 +138,44 @@ measurements.send(100)
 measurements.send(110)
 ```
 
+## WithLatestFrom Operator
+
+Another extension of the Combine framework that XCombine offers is the `withLatestFrom` operator which merges two publishers into one publisher by combining each element from self with the latest element from the second source. It returns a publisher that emits pairs of elements from the upstream publishers as tuples.
+
+### <a name="example-of-usage-withlatestfrom"></a>Example of Usage
+
+The following example demonstrates the use of the XCombine's `withLatestFrom` operator.
+
+```swift
+import Combine
+import XCombine
+
+let up1 = PassthroughSubject<Int, Never>()
+let up2 = PassthroughSubject<String, Never>()
+
+var subscriber = up1.x.withLatestFrom(up2)
+    .sink(
+        receiveCompletion: { completion in
+            print(completion)
+        },
+        receiveValue: { v1, v2 in
+            print(v1, v2)
+    })
+
+up1.send(1)
+up2.send("foo")
+up1.send(2)
+up1.send(3)
+up2.send("bar")
+up2.send("baz")
+up1.send(4)
+
+// The console output:
+// 2 foo
+// 3 foo
+// 4 baz
+```
+
 ## License
 
 This project is licensed under the MIT license.

Sources/XCombine/XCombine+UpstreamGroupCompleter.swift

@@ -0,0 +1,82 @@
+import Combine
+
+extension XCombine {
+    enum UpstreamCompletionPolicy {
+        case transparent
+        case completeAll
+    }
+
+    final class UpstreamGroupCompleter {
+        private var completionHandlers = [() -> Void]()
+        func register(_ handler: @escaping () -> Void) {
+            completionHandlers.append(handler)
+        }
+        func complete() {
+            completionHandlers.forEach { $0() }
+        }
+    }
+
+    final class UpstreamCompletionObserver<Upstream: Publisher>: Publisher
+    {
+        // MARK: - Types
+
+        typealias Output = Upstream.Output
+        typealias Failure = Upstream.Failure
+
+        // MARK: - Properties
+
+        private let upstream: Upstream
+        private var subscriber: AnySubscriber<Output, Failure>!
+
+        private let completer: UpstreamGroupCompleter
+        private let policy: UpstreamCompletionPolicy
+        private var isCompleted: Bool = false
+
+        // MARK: - Initialization
+
+        init(upstream: Upstream,
+             completer: UpstreamGroupCompleter,
+             policy: UpstreamCompletionPolicy)
+        {
+            self.upstream = upstream
+            self.completer = completer
+            self.policy = policy
+        }
+
+        // MARK: - Publisher Lifecycle
+
+        func receive<S: Subscriber>(subscriber: S)
+            where S.Failure == Failure, S.Input == Output
+        {
+            completer.register({
+                guard !self.isCompleted else { return; }
+                self.isCompleted = true
+                subscriber.receive(completion: .finished)
+            })
+
+            let innerSubscriber = AnySubscriber<Output, Failure>(
+                receiveSubscription: { subscription in
+                    subscriber.receive(subscription: subscription)
+            },
+                receiveValue: { value in
+                    subscriber.receive(value)
+            },
+                receiveCompletion: { completion in
+                    switch completion {
+                    case .finished:
+                        switch self.policy {
+                        case .transparent:
+                            self.isCompleted = true
+                            subscriber.receive(completion: completion)
+                        case .completeAll:
+                            self.completer.complete()
+                        }
+                    default:
+                        subscriber.receive(completion: completion)
+                    }
+            })
+
+            upstream.subscribe(innerSubscriber)
+        }
+    }
+}

Sources/XCombine/XCombine+WithLatestFrom.swift

@@ -0,0 +1,93 @@
+//
+//  XCombine+WithLatestFrom.swift
+//  XCombine
+//
+//  Created by Serge Bouts on 12/11/2019.
+//  Copyright © 2019 Serge Bouts. All rights reserved.
+//
+
+import Combine
+
+extension XCombine {
+    /// A "cold" publisher, that creates a new identity for each downstream subscriber.
+    public struct WithLatestFrom<Upstream: Publisher, Other: Publisher>: Publisher where Upstream.Failure == Other.Failure {
+        // MARK: - Types
+
+        public typealias Output = (Upstream.Output, Other.Output)
+        public typealias Failure = Upstream.Failure
+
+        // MARK: - Properties
+
+        private let upstream: Upstream
+        private let other: Other
+
+        private let completer = XCombine.UpstreamGroupCompleter()
+
+        // MARK: - Initialization
+
+        init(upstream: Upstream, other: Other) {
+            self.upstream = upstream
+            self.other = other
+        }
+
+        // MARK: - Publisher Lifecycle
+
+        public func receive<S: Subscriber>(subscriber: S) where S.Failure == Failure, S.Input == Output {
+            let merged = mergedStream(upstream, other)
+            let result = resultStream(from: merged)
+            result.subscribe(subscriber)
+        }
+    }
+}
+
+// MARK: - Helpers
+private extension XCombine.WithLatestFrom {
+    // MARK: - Types
+
+    enum MergedElement {
+        case upstream1(Upstream.Output)
+        case upstream2(Other.Output)
+    }
+
+    typealias ScanResult = (value1: Upstream.Output?, value2: Other.Output?, shouldEmit: Bool)
+
+    // MARK: - Pipelines
+
+    func mergedStream(_ upstream1: Upstream, _ upstream2: Other) -> AnyPublisher<MergedElement, Failure> {
+        let mergedElementUpstream1 = upstream1
+            .x.upstreamCompletionObserver(completer, policy: .completeAll)
+            .map { MergedElement.upstream1($0) }
+        let mergedElementUpstream2 = upstream2
+            .x.upstreamCompletionObserver(completer, policy: .transparent)
+            .map { MergedElement.upstream2($0) }
+        return mergedElementUpstream1
+            .merge(with: mergedElementUpstream2)
+            .eraseToAnyPublisher()
+    }
+
+    func resultStream(from mergedStream: AnyPublisher<MergedElement, Failure>) -> AnyPublisher<Output, Failure> {
+        mergedStream
+            .scan(nil) { (prevResult: ScanResult?, mergedElement: MergedElement) -> ScanResult? in
+                var newValue1: Upstream.Output?
+                var newValue2: Other.Output?
+                let shouldEmit: Bool
+
+                switch mergedElement {
+                case .upstream1(let v):
+                    newValue1 = v
+                    shouldEmit = prevResult?.value2 != nil
+                case .upstream2(let v):
+                    newValue2 = v
+                    shouldEmit = false
+                }
+
+                return ScanResult(value1: newValue1 ?? prevResult?.value1,
+                                  value2: newValue2 ?? prevResult?.value2,
+                                  shouldEmit: shouldEmit)
+            }
+            .compactMap { $0 }
+            .filter { $0.shouldEmit }
+            .map { Output($0.value1!, $0.value2!) }
+            .eraseToAnyPublisher()
+    }
+}

Sources/XCombine/XCombine.swift

@@ -34,4 +34,23 @@ extension XCombineNamespace {
     {
         return .init(upstream: self.upstream, capacity: replay)
     }
+
+    internal func upstreamCompletionObserver(
+        _ completer: XCombine.UpstreamGroupCompleter,
+        policy: XCombine.UpstreamCompletionPolicy
+    ) -> XCombine.UpstreamCompletionObserver<Upstream>
+    {
+        return .init(upstream: self.upstream,
+                     completer: completer,
+                     policy: policy)
+    }
+
+    /// Merges two publishers into one publisher by combining each element from self with the latest element from the second source, if any.
+    /// - Parameter other: Another publisher.
+    /// - Returns: A publisher that emits pairs of elements from the upstream publishers as tuples.
+    public func withLatestFrom<Other: Publisher>(_ other: Other)
+        -> XCombine.WithLatestFrom<Upstream, Other>
+    {
+        return .init(upstream: self.upstream, other: other)
+    }
 }