Add FlatMapStrategy.throttle (#713)

* Add `FlatMapStrategy.first`

* Add doc-comments on `FlattenStrategy.first`

* [Test] Remove `fdescribe`

* Update CHANGELOG.md

* Update Tests/ReactiveSwiftTests/SignalProducerSpec.swift

Co-Authored-By: Anders Ha <[email protected]>

* Rename to `FlattenStrategy.throttle`

* Update CHANGELOG.md

Author: inamiy

CHANGELOG.md

@@ -1,10 +1,11 @@
 # master
 *Please add new entries at the top.*
 
+1. `FlattenStrategy.throttle` is introduced. (#713, kudos to @inamiy)
 1. Updated `README.md` to reflect Swift 5.1 compatibility and point snippets to 6.1.0 (#763, kudos to @Marcocanc)
 1. Update travis to Xcode 11.1 and Swift 5.1 (#764, kudos @petrpavlik)
-2. [SwiftPM] Add platforms (#761, kudos to @ikesyo)
-3. Renamed `filterMap` to `compactMap` and deprecated `filterMap` (#746, kudos to @Marcocanc)
+1. [SwiftPM] Add platforms (#761, kudos to @ikesyo)
+1. Renamed `filterMap` to `compactMap` and deprecated `filterMap` (#746, kudos to @Marcocanc)
 
 # 6.1.0
 
@@ -25,7 +26,6 @@
 * Replace all cases where `NoError` was used in a `Signal` or `SignalProducer` with `Never`
 * Replace all cases where `AnyError` was used in a `Signal` or `SignalProducer` with `Swift.Error`
 
-
 # 5.0.1
 1. Fix warnings in Xcode 10.2
 

Sources/Flatten.swift

@@ -12,6 +12,7 @@ public struct FlattenStrategy {
 		case concurrent(limit: UInt)
 		case latest
 		case race
+		case throttle
 	}
 
 	fileprivate let kind: Kind
@@ -96,6 +97,21 @@ public struct FlattenStrategy {
 	/// Any failure from the inner streams is propagated immediately to the flattened
 	/// stream of values.
 	public static let race = FlattenStrategy(kind: .race)
+
+	/// Forward only events from the "first inner stream" that sends an event if not exists.
+	/// Other inner streams is disposed of until the first inner stream is completed.
+	/// Note that next inner stream after previous completion can become
+	/// first inner stream again.
+	///
+	/// The flattened stream of values completes only when the stream of streams has completed,
+	/// and first inner stream has completed if exists.
+	///
+	/// Any interruption of inner streams is propagated immediately to the flattened
+	/// stream of values.
+	///
+	/// Any failure from the inner streams is propagated immediately to the flattened
+	/// stream of values.
+	public static let throttle = FlattenStrategy(kind: .throttle)
 }
 
 extension Signal where Value: SignalProducerConvertible, Error == Value.Error {
@@ -120,6 +136,9 @@ extension Signal where Value: SignalProducerConvertible, Error == Value.Error {
 
 		case .race:
 			return self.race()
+
+		case .throttle:
+			return self.throttle()
 		}
 	}
 }
@@ -162,6 +181,9 @@ extension Signal where Value: SignalProducerConvertible, Error == Never, Value.E
 
 		case .race:
 			return self.race()
+
+		case .throttle:
+			return self.throttle()
 		}
 	}
 }
@@ -205,6 +227,9 @@ extension SignalProducer where Value: SignalProducerConvertible, Error == Value.
 
 		case .race:
 			return self.race()
+
+		case .throttle:
+			return self.throttle()
 		}
 	}
 }
@@ -247,6 +272,9 @@ extension SignalProducer where Value: SignalProducerConvertible, Error == Never,
 
 		case .race:
 			return self.race()
+
+		case .throttle:
+			return self.throttle()
 		}
 	}
 }
@@ -819,6 +847,112 @@ private struct RaceState {
 	var isActivated = false
 }
 
+extension Signal where Value: SignalProducerConvertible, Error == Value.Error {
+	/// Returns a signal that forwards values from the "first inner producer" if not exists,
+	/// ignoring values sent from other inner producers until first inner producer is completed.
+	/// Note that next inner producer after previous completion can become
+	/// first inner producer again.
+	///
+	/// An error sent on `self` or the first inner producer will be sent on the
+	/// returned signal.
+	///
+	/// The returned signal completes when `self` is completed, and also first inner producer
+	/// is completed if it exists.
+	fileprivate func throttle() -> Signal<Value.Value, Error> {
+		return Signal<Value.Value, Error> { observer, lifetime in
+			let relayDisposable = CompositeDisposable()
+			lifetime += relayDisposable
+			lifetime += self.observeThrottle(observer, relayDisposable)
+		}
+	}
+
+	fileprivate func observeThrottle(_ observer: Signal<Value.Value, Error>.Observer, _ relayDisposable: CompositeDisposable) -> Disposable? {
+		let state = Atomic(ThrottleState())
+
+		return self.observe { event in
+			switch event {
+			case let .value(innerProducer):
+				let isFirstInnerProducer: Bool = state.modify { state in
+					guard !state.hasFirstInnerProducer else {
+						return false
+					}
+
+					state.hasFirstInnerProducer = true
+					return true
+				}
+
+				// Ignore consecutive `innerProducer`s while `isFirstInnerProducer` is true.
+				guard isFirstInnerProducer else { return }
+
+				innerProducer.producer.startWithSignal { innerSignal, innerDisposable in
+					relayDisposable.add(innerDisposable)
+
+					innerSignal.observe { event in
+						switch event {
+						case .completed:
+							let shouldComplete: Bool = state.modify { state in
+								state.hasFirstInnerProducer = false
+								return state.outerSignalComplete
+							}
+
+							if shouldComplete {
+								observer.sendCompleted()
+							}
+
+						case .value, .failed, .interrupted:
+							observer.send(event)
+						}
+					}
+				}
+
+			case let .failed(error):
+				observer.send(error: error)
+
+			case .completed:
+				let shouldComplete: Bool = state.modify { state in
+					state.outerSignalComplete = true
+					return !state.hasFirstInnerProducer
+				}
+
+				if shouldComplete {
+					observer.sendCompleted()
+				}
+
+			case .interrupted:
+				observer.sendInterrupted()
+			}
+		}
+	}
+}
+
+extension SignalProducer where Value: SignalProducerConvertible, Error == Value.Error {
+	/// Returns a producer that forwards values from the "first inner producer" if not exists,
+	/// ignoring values sent from other inner producers until first inner producer is completed.
+	/// Note that next inner producer after previous completion can become first inner producer again.
+	///
+	/// An error sent on `self` or the first inner producer will be sent on the
+	/// returned producer.
+	///
+	/// The returned signal completes when `self` is completed, and also first inner producer
+	/// is completed if it exists.
+	fileprivate func throttle() -> SignalProducer<Value.Value, Error> {
+		return SignalProducer<Value.Value, Error> { observer, lifetime in
+			let relayDisposable = CompositeDisposable()
+			lifetime += relayDisposable
+
+			self.startWithSignal { signal, signalDisposable in
+				lifetime += signalDisposable
+				lifetime += signal.observeThrottle(observer, relayDisposable)
+			}
+		}
+	}
+}
+
+private struct ThrottleState {
+	var outerSignalComplete = false
+	var hasFirstInnerProducer = false
+}
+
 extension Signal {
 	/// Maps each event from `signal` to a new signal, then flattens the
 	/// resulting producers (into a signal of values), according to the

Tests/ReactiveSwiftTests/SignalProducerSpec.swift

@@ -1781,6 +1781,135 @@ class SignalProducerSpec: QuickSpec {
 				}
 			}
 
+			describe("FlattenStrategy.throttle") {
+				it("should forward values from the first and third inner producer to send an event") {
+					let (outer, outerObserver) = SignalProducer<SignalProducer<Int, TestError>, TestError>.pipe()
+					let (firstInner, firstInnerObserver) = SignalProducer<Int, TestError>.pipe()
+					let (secondInner, secondInnerObserver) = SignalProducer<Int, TestError>.pipe()
+					let (thirdInner, thirdInnerObserver) = SignalProducer<Int, TestError>.pipe()
+
+					var receivedValues: [Int] = []
+					var errored = false
+					var completed = false
+
+					outer.flatten(.throttle).start { event in
+						switch event {
+						case let .value(value):
+							receivedValues.append(value)
+						case .completed:
+							completed = true
+						case .failed:
+							errored = true
+						case .interrupted:
+							break
+						}
+					}
+
+					outerObserver.send(value: firstInner)
+					outerObserver.send(value: secondInner)
+
+					firstInnerObserver.send(value: 1)
+					secondInnerObserver.send(value: 2)
+
+					expect(receivedValues) == [ 1 ]
+					expect(errored) == false
+					expect(completed) == false
+
+					secondInnerObserver.send(value: 3)
+					secondInnerObserver.sendCompleted()
+
+					expect(receivedValues) == [ 1 ]
+					expect(errored) == false
+					expect(completed) == false
+
+					firstInnerObserver.sendCompleted()
+
+					expect(receivedValues) == [ 1 ]
+					expect(errored) == false
+					expect(completed) == false
+
+					outerObserver.send(value: thirdInner)
+					thirdInnerObserver.send(value: 4)
+
+					// NOTE:
+					// `4` will be observed because `firstInner` is completed then `thirdInner` is emitted,
+					// which is also considered as "first" producer.
+					expect(receivedValues) == [ 1, 4 ]
+					expect(errored) == false
+					expect(completed) == false
+
+					outerObserver.sendCompleted()
+
+					expect(receivedValues) == [ 1, 4 ]
+					expect(errored) == false
+					expect(completed) == false
+
+					thirdInnerObserver.sendCompleted()
+
+					expect(receivedValues) == [ 1, 4 ]
+					expect(errored) == false
+					expect(completed) == true
+				}
+
+				it("should forward an error from the first inner producer to send an error") {
+					let inner = SignalProducer<Int, TestError>(error: .default)
+					let outer = SignalProducer<SignalProducer<Int, TestError>, TestError>(value: inner)
+
+					let result = outer.flatten(.throttle).first()
+					expect(result?.error) == TestError.default
+				}
+
+				it("should forward an error from the outer producer") {
+					let outer = SignalProducer<SignalProducer<Int, TestError>, TestError>(error: .default)
+
+					let result = outer.flatten(.throttle).first()
+					expect(result?.error) == TestError.default
+				}
+
+				it("should complete when the 'outer producer' and 'first inner producer to send an event' have completed") {
+					let inner = SignalProducer<Int, TestError>.empty
+					let outer = SignalProducer<SignalProducer<Int, TestError>, TestError>(value: inner)
+
+					var completed = false
+					outer.flatten(.throttle).startWithCompleted {
+						completed = true
+					}
+
+					expect(completed) == true
+				}
+
+				it("should complete when the outer producer completes before sending any inner producers") {
+					let outer = SignalProducer<SignalProducer<Int, TestError>, TestError>.empty
+
+					var completed = false
+					outer.flatten(.throttle).startWithCompleted {
+						completed = true
+					}
+
+					expect(completed) == true
+				}
+
+				it("should not complete when the outer producer completes after sending an inner producer but it doesn't send an event") {
+					let inner = SignalProducer<Int, TestError>.never
+					let outer = SignalProducer<SignalProducer<Int, TestError>, TestError>(value: inner)
+
+					var completed = false
+					outer.flatten(.throttle).startWithCompleted {
+						completed = true
+					}
+
+					expect(completed) == false
+				}
+
+				it("should not deadlock") {
+					let producer = SignalProducer<Int, Never>(value: 1)
+						.flatMap(.throttle) { _ in SignalProducer(value: 10) }
+
+					let result = producer.take(first: 1).last()
+					expect(result?.value) == 10
+				}
+			}
+
 			describe("interruption") {
 				var innerObserver: Signal<(), Never>.Observer!
 				var outerObserver: Signal<SignalProducer<(), Never>, Never>.Observer!