Add a recurring collect function and associated unit tests

CHANGELOG.md

@@ -1,6 +1,7 @@
 # master
 *Please add new entries at the top.*
 
+1. New method `collect(every:on:skipEmpty:discardWhenCompleted:)` which delivers all values that occurred during a time interval (#619, kudos to @Qata)
 1. Result now interoperates with SignalProducer n-ary operators as a constant producer (#606, kudos to @Qata)
 1. New property operator: `filter` (#586, kudos to @iv-mexx)
 1. New operator `merge(with:)` (#600, kudos to @ra1028)

Sources/Event.swift

@@ -861,6 +861,48 @@ extension Signal.Event {
 			}
 		}
 	}
+
+	internal static func collect(every interval: DispatchTimeInterval, on scheduler: DateScheduler, skipEmpty: Bool, discardsWhenCompleted: Bool) -> Transformation<[Value], Error> {
+		return { action, lifetime in
+			let values = Atomic<[Value]>([])
+			let d = SerialDisposable()
+			var lastDelivered: Date = scheduler.currentDate
+
+			d.inner = scheduler.schedule(after: scheduler.currentDate.addingTimeInterval(interval), interval: interval, leeway: interval * 0.1) {
+				let currentValues = values.swap([])
+				if !(currentValues.isEmpty && skipEmpty) {
+					action(.value(currentValues))
+					lastDelivered = scheduler.currentDate
+				}
+			}
+
+			lifetime.observeEnded {
+				d.dispose()
+				scheduler.schedule { action(.interrupted) }
+			}
+
+			return { event in
+				switch event {
+				case let .value(value):
+					values.modify { $0.append(value) }
+				case let .failed(error):
+					d.inner = scheduler.schedule { action(.failed(error)) }
+				case .completed:
+					let currentValues = values.swap([])
+					if !discardsWhenCompleted, !(currentValues.isEmpty && skipEmpty) {
+						d.inner = scheduler.schedule(after: lastDelivered.addingTimeInterval(interval)) {
+							action(.value(currentValues))
+							action(.completed)
+						}
+					} else {
+						d.inner = scheduler.schedule { action(.completed) }
+					}
+				case .interrupted:
+					d.inner = scheduler.schedule { action(.interrupted) }
+				}
+			}
+		}
+	}
 }
 
 private struct ThrottleState<Value> {

Sources/Signal.swift

@@ -765,6 +765,28 @@ extension Signal {
 		return flatMapEvent(Signal.Event.collect(shouldEmit))
 	}
 
+	/// Forward the latest values on `scheduler` every `interval`.
+	///
+	/// - note: If `self` terminates while values are being accumulated,
+	///         the behaviour will be determined by `discardsWhenCompleted`.
+	///         If `true`, the values will be discarded and the returned signal
+	///         will terminate immediately.
+	///         If `false`, that values will be delivered at the next interval.
+	///
+	/// - parameters:
+	///   - interval: A repetition interval.
+	///   - scheduler: A scheduler to send values on.
+	///   - skipEmpty: Whether empty arrays should be sent if no values were
+	///     accumulated during the interval.
+	///   - discardsWhenCompleted: A boolean to indicate if the latest unsent
+	///     values should be discarded on completion.
+	///
+	/// - returns: A signal that sends all values that are sent from `self` at
+	///            `interval` seconds apart.
+	public func collect(every interval: DispatchTimeInterval, on scheduler: DateScheduler, skipEmpty: Bool = false, discardsWhenCompleted: Bool = true) -> Signal<[Value], Error> {
+		return flatMapEvent(Signal.Event.collect(every: interval, on: scheduler, skipEmpty: skipEmpty, discardsWhenCompleted: discardsWhenCompleted))
+	}
+
 	/// Forward all events onto the given scheduler, instead of whichever
 	/// scheduler they originally arrived upon.
 	///

Sources/SignalProducer.swift

@@ -1055,6 +1055,28 @@ extension SignalProducer {
 		return core.flatMapEvent(Signal.Event.collect(shouldEmit))
 	}
 
+	/// Forward the latest values on `scheduler` every `interval`.
+	///
+	/// - note: If `self` terminates while values are being accumulated,
+	///         the behaviour will be determined by `discardsWhenCompleted`.
+	///         If `true`, the values will be discarded and the returned producer
+	///         will terminate immediately.
+	///         If `false`, that values will be delivered at the next interval.
+	///
+	/// - parameters:
+	///   - interval: A repetition interval.
+	///   - scheduler: A scheduler to send values on.
+	///   - skipEmpty: Whether empty arrays should be sent if no values were
+	///     accumulated during the interval.
+	///   - discardsWhenCompleted: A boolean to indicate if the latest unsent
+	///     values should be discarded on completion.
+	///
+	/// - returns: A producer that sends all values that are sent from `self`
+	///            at `interval` seconds apart.
+	public func collect(every interval: DispatchTimeInterval, on scheduler: DateScheduler, skipEmpty: Bool = false, discardsWhenCompleted: Bool = true) -> SignalProducer<[Value], Error> {
+		return core.flatMapEvent(Signal.Event.collect(every: interval, on: scheduler, skipEmpty: skipEmpty, discardsWhenCompleted: discardsWhenCompleted))
+	}
+
 	/// Forward all events onto the given scheduler, instead of whichever
 	/// scheduler they originally arrived upon.
 	///