Swift 的 Combine 框架是否有类似于 RXSwift 或 Reactive Swift 中的 sample(on:) 运算符？

Question

有人知道如何在 Combine 中重新创建采样行为吗？

这是样本在 RXMarbles 中的行为图表

sample() 的要点是有两个流，当一个被触发时，如果另一个流的最新值尚未发送，则发送另一个流的最新值。

Answer 1

这是一个可能做你想做的事的游乐场。我没有对其进行大量测试，因此请谨慎行事：

import UIKit
import Combine
import PlaygroundSupport

struct SamplePublisher<DataSeq, Trigger, E> : Publisher
where DataSeq : Publisher,
      Trigger : Publisher,
      DataSeq.Failure == Trigger.Failure,
      E == DataSeq.Failure,
      DataSeq.Output : Equatable {

    typealias Output = DataSeq.Output
    typealias Failure = E

    // The two sequences we are observing, the data sequence and the
    // trigger sequence.  When the trigger fires it will send the
    // latest value from the dataSequence UNLESS it hasn't changed
    let dataPublisher : DataSeq
    let triggerPublisher : Trigger

    struct SamplePublisherSubscription : Subscription {
        var combineIdentifier = CombineIdentifier()

        let dataSubscription : AnyCancellable
        let triggerSubscription : Subscription

        func request(_ demand: Subscribers.Demand) {
            triggerSubscription.request(demand)
        }

        func cancel() {
            dataSubscription.cancel()
            triggerSubscription.cancel()
        }
    }

    func receive<S>(subscriber: S) where S : Subscriber, E == S.Failure, DataSeq.Output == S.Input {
        var latestData : DataSeq.Output?
        var lastSent : DataSeq.Output?
        var triggerSubscription : Subscription?

        // Compares the latest value sent to the last one that was sent.
        // If they don't match then it sends the latest value along.
        // IF they do match, or if no value has been sent on the data stream yet
        // Don't emit a new value.
        func emitIfNeeded() -> Subscribers.Demand {
            guard let latest = latestData else { return .unlimited }

            if nil == lastSent ||
                lastSent! != latest {
                lastSent = latest
                return subscriber.receive(latest)
            } else {
                return .unlimited
            }
        }

        // Here we watch the data stream for new values and simply
        // record them.  If the data stream ends, or erors we
        // pass that on to our subscriber.
        let dataSubscription = dataPublisher.sink(
            receiveCompletion: {
                switch $0 {
                    case .finished:
                        subscriber.receive(completion: .finished)
                    case .failure(let error):
                        subscriber.receive(completion: .failure(error))
                }
            },
            receiveValue: {
                latestData = $0
            })

        // The thing that subscribes to the trigger sequence.
        // When it receives a value, we emit the latest value from the data stream (if any).
        // If the trigger stream ends or errors, that will also end or error this publisher.
        let triggerSubscriber = AnySubscriber<Trigger.Output,Trigger.Failure>(
            receiveSubscription: { subscription in triggerSubscription = subscription },
            receiveValue: { _ in emitIfNeeded() },
            receiveCompletion: {
                switch $0 {
                    case .finished :
                        emitIfNeeded()
                        subscriber.receive(completion: .finished)
                    case .failure(let error) :
                        subscriber.receive(completion: .failure(error))
                }
            })

        // subscribe to the trigger sequence
        triggerPublisher.subscribe(triggerSubscriber)

        // Record relevant information and return the subscription to the subscriber.
        subscriber.receive(subscription: SamplePublisherSubscription(
            dataSubscription: dataSubscription,
            triggerSubscription: triggerSubscription!))
    }
}

extension Publisher {

    // A utility function that lets you create a stream that is triggered by
    // a value being emitted from another stream
    func sample<Trigger, E>(trigger: Trigger) -> SamplePublisher<Self, Trigger, E>
    where Trigger : Publisher,
          Self.Failure == Trigger.Failure,
          E == Self.Failure,
    Self.Output : Equatable {
        return SamplePublisher( dataPublisher : self, triggerPublisher : trigger)
    }
}

var count = 0
let timer = Timer.publish(every: 5.0, on: RunLoop.current, in: .common).autoconnect().eraseToAnyPublisher()
let data = Timer.publish(every: 1.0, on: RunLoop.current, in: .common)
    .autoconnect()
    .scan(0) { total, _ in total + 1}


var subscriptions = Set<AnyCancellable>()
data.sample(trigger: timer).print()
    .sink(receiveCompletion: {
        debugPrint($0)
    }, receiveValue: {
        debugPrint($0)
    }).store(in: &subscriptions)

PlaygroundSupport.PlaygroundPage.current.needsIndefiniteExecution = true

Answer 2

CombineExt 库具有 withLatestFrom 运算符，可以满足您的需求，以及许多其他有用的运算符。