140 lines
4.5 KiB
Swift
140 lines
4.5 KiB
Swift
//
|
|
// Sample.swift
|
|
// RxSwift
|
|
//
|
|
// Created by Krunoslav Zaher on 5/1/15.
|
|
// Copyright © 2015 Krunoslav Zaher. All rights reserved.
|
|
//
|
|
|
|
extension ObservableType {
|
|
|
|
/**
|
|
Samples the source observable sequence using a sampler observable sequence producing sampling ticks.
|
|
|
|
Upon each sampling tick, the latest element (if any) in the source sequence during the last sampling interval is sent to the resulting sequence.
|
|
|
|
**In case there were no new elements between sampler ticks, you may provide a default value to be emitted, instead
|
|
to the resulting sequence otherwise no element is sent.**
|
|
|
|
- seealso: [sample operator on reactivex.io](http://reactivex.io/documentation/operators/sample.html)
|
|
|
|
- parameter sampler: Sampling tick sequence.
|
|
- parameter defaultValue: a value to return if there are no new elements between sampler ticks
|
|
- returns: Sampled observable sequence.
|
|
*/
|
|
public func sample<Source: ObservableType>(_ sampler: Source, defaultValue: Element? = nil)
|
|
-> Observable<Element> {
|
|
return Sample(source: self.asObservable(), sampler: sampler.asObservable(), defaultValue: defaultValue)
|
|
}
|
|
}
|
|
|
|
final private class SamplerSink<Observer: ObserverType, SampleType>
|
|
: ObserverType
|
|
, LockOwnerType
|
|
, SynchronizedOnType {
|
|
typealias Element = SampleType
|
|
|
|
typealias Parent = SampleSequenceSink<Observer, SampleType>
|
|
|
|
private let parent: Parent
|
|
|
|
var lock: RecursiveLock {
|
|
self.parent.lock
|
|
}
|
|
|
|
init(parent: Parent) {
|
|
self.parent = parent
|
|
}
|
|
|
|
func on(_ event: Event<Element>) {
|
|
self.synchronizedOn(event)
|
|
}
|
|
|
|
func synchronized_on(_ event: Event<Element>) {
|
|
switch event {
|
|
case .next, .completed:
|
|
if let element = parent.element ?? self.parent.defaultValue {
|
|
self.parent.element = nil
|
|
self.parent.forwardOn(.next(element))
|
|
}
|
|
|
|
if self.parent.atEnd {
|
|
self.parent.forwardOn(.completed)
|
|
self.parent.dispose()
|
|
}
|
|
case .error(let e):
|
|
self.parent.forwardOn(.error(e))
|
|
self.parent.dispose()
|
|
}
|
|
}
|
|
}
|
|
|
|
final private class SampleSequenceSink<Observer: ObserverType, SampleType>
|
|
: Sink<Observer>
|
|
, ObserverType
|
|
, LockOwnerType
|
|
, SynchronizedOnType {
|
|
typealias Element = Observer.Element
|
|
typealias Parent = Sample<Element, SampleType>
|
|
|
|
fileprivate let parent: Parent
|
|
fileprivate let defaultValue: Element?
|
|
|
|
let lock = RecursiveLock()
|
|
|
|
// state
|
|
fileprivate var element = nil as Element?
|
|
fileprivate var atEnd = false
|
|
|
|
private let sourceSubscription = SingleAssignmentDisposable()
|
|
|
|
init(parent: Parent, observer: Observer, cancel: Cancelable, defaultValue: Element? = nil) {
|
|
self.parent = parent
|
|
self.defaultValue = defaultValue
|
|
super.init(observer: observer, cancel: cancel)
|
|
}
|
|
|
|
func run() -> Disposable {
|
|
self.sourceSubscription.setDisposable(self.parent.source.subscribe(self))
|
|
let samplerSubscription = self.parent.sampler.subscribe(SamplerSink(parent: self))
|
|
|
|
return Disposables.create(sourceSubscription, samplerSubscription)
|
|
}
|
|
|
|
func on(_ event: Event<Element>) {
|
|
self.synchronizedOn(event)
|
|
}
|
|
|
|
func synchronized_on(_ event: Event<Element>) {
|
|
switch event {
|
|
case .next(let element):
|
|
self.element = element
|
|
case .error:
|
|
self.forwardOn(event)
|
|
self.dispose()
|
|
case .completed:
|
|
self.atEnd = true
|
|
self.sourceSubscription.dispose()
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
final private class Sample<Element, SampleType>: Producer<Element> {
|
|
fileprivate let source: Observable<Element>
|
|
fileprivate let sampler: Observable<SampleType>
|
|
fileprivate let defaultValue: Element?
|
|
|
|
init(source: Observable<Element>, sampler: Observable<SampleType>, defaultValue: Element? = nil) {
|
|
self.source = source
|
|
self.sampler = sampler
|
|
self.defaultValue = defaultValue
|
|
}
|
|
|
|
override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
|
|
let sink = SampleSequenceSink(parent: self, observer: observer, cancel: cancel, defaultValue: self.defaultValue)
|
|
let subscription = sink.run()
|
|
return (sink: sink, subscription: subscription)
|
|
}
|
|
}
|