Extended Sample operation to support a default value (#1457)

.gitignore

@@ -52,6 +52,7 @@ Carthage/Build
 
 .build/
 Packages/
+.swiftpm
 
 # AppCode
 

CHANGELOG.md

@@ -21,6 +21,7 @@ All notable changes to this project will be documented in this file.
 * Add `UITextField.isSecureTextEntry` binder. #1968
 * Remove "custom" `Result` in favor of `Foundation.Resault`. #2006
 * Fix compilation error in `SharedSequence.createUnsafe`. #2014
+* Added `defaultValue` to `sample` to be returned when no new events occur between sampler ticks. #1457
 
 ## [5.0.1](https://github.com/ReactiveX/RxSwift/releases/tag/5.0.1)
 

RxSwift/Observables/Sample.swift

@@ -13,16 +13,18 @@ extension ObservableType {
 
      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, no element 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)
+    public func sample<Source: ObservableType>(_ sampler: Source, defaultValue: Element? = nil)
         -> Observable<Element> {
-            return Sample(source: self.asObservable(), sampler: sampler.asObservable())
+            return Sample(source: self.asObservable(), sampler: sampler.asObservable(), defaultValue: defaultValue)
     }
 }
 
@@ -51,7 +53,7 @@ final private class SamplerSink<Observer: ObserverType, SampleType>
     func synchronized_on(_ event: Event<Element>) {
         switch event {
         case .next, .completed:
-            if let element = parent.element {
+            if let element = parent.element ?? self.parent.defaultValue {
                 self.parent.element = nil
                 self.parent.forwardOn(.next(element))
             }
@@ -75,7 +77,8 @@ final private class SampleSequenceSink<Observer: ObserverType, SampleType>
     typealias Element = Observer.Element 
     typealias Parent = Sample<Element, SampleType>
 
-    private let parent: Parent
+    fileprivate let parent: Parent
+    fileprivate let defaultValue: Element?
 
     let lock = RecursiveLock()
 
@@ -85,8 +88,9 @@ final private class SampleSequenceSink<Observer: ObserverType, SampleType>
 
     private let sourceSubscription = SingleAssignmentDisposable()
 
-    init(parent: Parent, observer: Observer, cancel: Cancelable) {
+    init(parent: Parent, observer: Observer, cancel: Cancelable, defaultValue: Element? = nil) {
         self.parent = parent
+        self.defaultValue = defaultValue
         super.init(observer: observer, cancel: cancel)
     }
 
@@ -119,14 +123,16 @@ final private class SampleSequenceSink<Observer: ObserverType, SampleType>
 final private class Sample<Element, SampleType>: Producer<Element> {
     fileprivate let source: Observable<Element>
     fileprivate let sampler: Observable<SampleType>
-
-    init(source: Observable<Element>, 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)
+        let sink = SampleSequenceSink(parent: self, observer: observer, cancel: cancel, defaultValue: self.defaultValue)
         let subscription = sink.run()
         return (sink: sink, subscription: subscription)
     }

Sources/AllTestz/main.swift

@@ -1230,6 +1230,7 @@ final class ObservableSampleTest_ : ObservableSampleTest, RxTestCase {
     #endif
 
     static var allTests: [(String, (ObservableSampleTest_) -> () -> Void)] { return [
+    ("testSample_Sampler_DefaultValue", ObservableSampleTest.testSample_Sampler_DefaultValue),
     ("testSample_Sampler_SamplerThrows", ObservableSampleTest.testSample_Sampler_SamplerThrows),
     ("testSample_Sampler_Simple1", ObservableSampleTest.testSample_Sampler_Simple1),
     ("testSample_Sampler_Simple2", ObservableSampleTest.testSample_Sampler_Simple2),

Tests/RxSwiftTests/Observable+SampleTests.swift

@@ -14,6 +14,52 @@ class ObservableSampleTest : RxTest {
 }
 
 extension ObservableSampleTest {
+    func testSample_Sampler_DefaultValue() {
+        let scheduler = TestScheduler(initialClock: 0)
+
+        let xs = scheduler.createHotObservable([
+            .next(150, 1),
+            .next(220, 2),
+            .next(240, 3),
+            .next(290, 4),
+            .next(300, 5),
+            .next(310, 6),
+            .completed(400)
+            ])
+
+        let ys = scheduler.createHotObservable([
+            .next(150, ""),
+            .next(210, "bar"),
+            .next(250, "foo"),
+            .next(260, "qux"),
+            .next(320, "baz"),
+            .completed(500)
+            ])
+
+        let res = scheduler.start {
+            xs.sample(ys, defaultValue: 0)
+        }
+
+        let correct = Recorded.events(
+            .next(210, 0),
+            .next(250, 3),
+            .next(260, 0),
+            .next(320, 6),
+            .next(500, 0),
+            .completed(500)
+        )
+
+        XCTAssertEqual(res.events, correct)
+
+        XCTAssertEqual(xs.subscriptions, [
+            Subscription(200, 400)
+            ])
+
+        XCTAssertEqual(ys.subscriptions, [
+            Subscription(200, 500)
+            ])
+    }
+
     func testSample_Sampler_SamplerThrows() {
         let scheduler = TestScheduler(initialClock: 0)