Add scanMap (#695)

* Add `transduce`

* Add documentation on `transduce`

* Update CHANGELOG.md

* Rename `transduce` to `scanMap`

* Use `scanMap` for `scan` implementation

* Sort method declaration order

* Fix documentation

CHANGELOG.md

@@ -45,6 +45,7 @@
 1. New operator `materializeResults` and `dematerializeResults` (#679, kudos to @ra1028)
 1. New convenience initializer for `Action` that takes a `ValidatingProperty` as its state (#637, kudos to @Marcocanc)
 1. Fix legacy date implementation. (#683, kudos to @shoheiyokoyama)
+1. New operator `scanMap`. (#695, kudos to @inamiy)
 
 # 4.0.0
 

Sources/Event.swift

@@ -690,33 +690,51 @@ extension Signal.Event {
 		}
 	}
 
-	internal static func scan<U>(into initialResult: U, _ nextPartialResult: @escaping (inout U, Value) -> Void) -> Transformation<U, Error> {
+	internal static func reduce<U>(into initialResult: U, _ nextPartialResult: @escaping (inout U, Value) -> Void) -> Transformation<U, Error> {
 		return { action, _ in
 			var accumulator = initialResult
 
 			return { event in
-				action(event.map { value in
+				switch event {
+				case let .value(value):
 					nextPartialResult(&accumulator, value)
-					return accumulator
-				})
+				case .completed:
+					action(.value(accumulator))
+					action(.completed)
+				case .interrupted:
+					action(.interrupted)
+				case let .failed(error):
+					action(.failed(error))
+				}
 			}
 		}
 	}
 
+	internal static func reduce<U>(_ initialResult: U, _ nextPartialResult: @escaping (U, Value) -> U) -> Transformation<U, Error> {
+		return reduce(into: initialResult) { $0 = nextPartialResult($0, $1) }
+	}
+
+	internal static func scan<U>(into initialResult: U, _ nextPartialResult: @escaping (inout U, Value) -> Void) -> Transformation<U, Error> {
+		return self.scanMap(into: initialResult, { result, value -> U in
+			nextPartialResult(&result, value)
+			return result
+		})
+	}
+
 	internal static func scan<U>(_ initialResult: U, _ nextPartialResult: @escaping (U, Value) -> U) -> Transformation<U, Error> {
 		return scan(into: initialResult) { $0 = nextPartialResult($0, $1) }
 	}
 
-	internal static func reduce<U>(into initialResult: U, _ nextPartialResult: @escaping (inout U, Value) -> Void) -> Transformation<U, Error> {
+	internal static func scanMap<State, U>(into initialState: State, _ next: @escaping (inout State, Value) -> U) -> Transformation<U, Error> {
 		return { action, _ in
-			var accumulator = initialResult
+			var accumulator = initialState
 
 			return { event in
 				switch event {
 				case let .value(value):
-					nextPartialResult(&accumulator, value)
+					let output = next(&accumulator, value)
+					action(.value(output))
 				case .completed:
-					action(.value(accumulator))
 					action(.completed)
 				case .interrupted:
 					action(.interrupted)
@@ -727,8 +745,12 @@ extension Signal.Event {
 		}
 	}
 
-	internal static func reduce<U>(_ initialResult: U, _ nextPartialResult: @escaping (U, Value) -> U) -> Transformation<U, Error> {
-		return reduce(into: initialResult) { $0 = nextPartialResult($0, $1) }
+	internal static func scanMap<State, U>(_ initialState: State, _ next: @escaping (State, Value) -> (State, U)) -> Transformation<U, Error> {
+		return scanMap(into: initialState) { state, value in
+			let new = next(state, value)
+			state = new.0
+			return new.1
+		}
 	}
 
 	internal static func observe(on scheduler: Scheduler) -> Transformation<Value, Error> {

Sources/Signal.swift

@@ -1324,6 +1324,34 @@ extension Signal {
 	public func scan<U>(into initialResult: U, _ nextPartialResult: @escaping (inout U, Value) -> Void) -> Signal<U, Error> {
 		return flatMapEvent(Signal.Event.scan(into: initialResult, nextPartialResult))
 	}
+
+	/// Accumulate all values from `self` as `State`, and send the value as `U`.
+	///
+	/// - parameters:
+	///   - initialState: The state to use as the initial accumulating state.
+	///   - next: A closure that combines the accumulating state and the latest value
+	///           from `self`. The result would be "next state" and "output" where
+	///           "output" would be forwarded and "next state" would be used in the
+	///           next call of `next`.
+	///
+	/// - returns: A producer that sends the output that is computed from the accumuation.
+	public func scanMap<State, U>(_ initialState: State, _ next: @escaping (State, Value) -> (State, U)) -> Signal<U, Error> {
+		return flatMapEvent(Signal.Event.scanMap(initialState, next))
+	}
+
+	/// Accumulate all values from `self` as `State`, and send the value as `U`.
+	///
+	/// - parameters:
+	///   - initialState: The state to use as the initial accumulating state.
+	///   - next: A closure that combines the accumulating state and the latest value
+	///           from `self`. The result would be "next state" and "output" where
+	///           "output" would be forwarded and "next state" would be used in the
+	///           next call of `next`.
+	///
+	/// - returns: A producer that sends the output that is computed from the accumuation.
+	public func scanMap<State, U>(into initialState: State, _ next: @escaping (inout State, Value) -> U) -> Signal<U, Error> {
+		return flatMapEvent(Signal.Event.scanMap(into: initialState, next))
+	}
 }
 
 extension Signal where Value: Equatable {

Sources/SignalProducer.swift

@@ -1493,6 +1493,34 @@ extension SignalProducer {
 		return core.flatMapEvent(Signal.Event.scan(into: initialResult, nextPartialResult))
 	}
 
+	/// Accumulate all values from `self` as `State`, and send the value as `U`.
+	///
+	/// - parameters:
+	///   - initialState: The state to use as the initial accumulating state.
+	///   - next: A closure that combines the accumulating state and the latest value
+	///           from `self`. The result would be "next state" and "output" where
+	///           "output" would be forwarded and "next state" would be used in the
+	///           next call of `next`.
+	///
+	/// - returns: A producer that sends the output that is computed from the accumuation.
+	public func scanMap<State, U>(_ initialState: State, _ next: @escaping (State, Value) -> (State, U)) -> SignalProducer<U, Error> {
+		return core.flatMapEvent(Signal.Event.scanMap(initialState, next))
+	}
+
+	/// Accumulate all values from `self` as `State`, and send the value as `U`.
+	///
+	/// - parameters:
+	///   - initialState: The state to use as the initial accumulating state.
+	///   - next: A closure that combines the accumulating state and the latest value
+	///           from `self`. The result would be "next state" and "output" where
+	///           "output" would be forwarded and "next state" would be used in the
+	///           next call of `next`.
+	///
+	/// - returns: A producer that sends the output that is computed from the accumuation.
+	public func scanMap<State, U>(into initialState: State, _ next: @escaping (inout State, Value) -> U) -> SignalProducer<U, Error> {
+		return core.flatMapEvent(Signal.Event.scanMap(into: initialState, next))
+	}
+
 	/// Forward only values from `self` that are not considered equivalent to its
 	/// immediately preceding value.
 	///

Tests/ReactiveSwiftTests/SignalProducerLiftingSpec.swift

@@ -423,6 +423,55 @@ class SignalProducerLiftingSpec: QuickSpec {
 			}
 		}
 
+		describe("scanMap(_:_:)") {
+			it("should update state and output separately") {
+				let (baseProducer, observer) = SignalProducer<Int, NoError>.pipe()
+				let producer = baseProducer.scanMap(false) { state, value -> (Bool, String) in
+					return (true, state ? "\(value)" : "initial")
+				}
+
+				var lastValue: String?
+
+				producer.startWithValues { lastValue = $0 }
+
+				expect(lastValue).to(beNil())
+
+				observer.send(value: 1)
+				expect(lastValue) == "initial"
+
+				observer.send(value: 2)
+				expect(lastValue) == "2"
+
+				observer.send(value: 3)
+				expect(lastValue) == "3"
+			}
+		}
+
+		describe("scanMap(into:_:)") {
+			it("should update state and output separately") {
+				let (baseProducer, observer) = SignalProducer<Int, NoError>.pipe()
+				let producer = baseProducer.scanMap(into: false) { (state: inout Bool, value: Int) -> String in
+					defer { state = true }
+					return state ? "\(value)" : "initial"
+				}
+
+				var lastValue: String?
+
+				producer.startWithValues { lastValue = $0 }
+
+				expect(lastValue).to(beNil())
+
+				observer.send(value: 1)
+				expect(lastValue) == "initial"
+
+				observer.send(value: 2)
+				expect(lastValue) == "2"
+
+				observer.send(value: 3)
+				expect(lastValue) == "3"
+			}
+		}
+
 		describe("reduce(_:_:)") {
 			it("should accumulate one value") {
 				let (baseProducer, observer) = SignalProducer<Int, Never>.pipe()

Tests/ReactiveSwiftTests/SignalSpec.swift

@@ -834,6 +834,55 @@ class SignalSpec: QuickSpec {
 			}
 		}
 
+		describe("scanMap(_:_:)") {
+			it("should update state and output separately") {
+				let (baseSignal, observer) = Signal<Int, NoError>.pipe()
+				let signal = baseSignal.scanMap(false) { state, value -> (Bool, String) in
+					return (true, state ? "\(value)" : "initial")
+				}
+
+				var lastValue: String?
+
+				signal.observeValues { lastValue = $0 }
+
+				expect(lastValue).to(beNil())
+
+				observer.send(value: 1)
+				expect(lastValue) == "initial"
+
+				observer.send(value: 2)
+				expect(lastValue) == "2"
+
+				observer.send(value: 3)
+				expect(lastValue) == "3"
+			}
+		}
+
+		describe("scanMap(into:_:)") {
+			it("should update state and output separately") {
+				let (baseSignal, observer) = Signal<Int, NoError>.pipe()
+				let signal = baseSignal.scanMap(into: false) { (state: inout Bool, value: Int) -> String in
+					defer { state = true }
+					return state ? "\(value)" : "initial"
+				}
+
+				var lastValue: String?
+
+				signal.observeValues { lastValue = $0 }
+
+				expect(lastValue).to(beNil())
+
+				observer.send(value: 1)
+				expect(lastValue) == "initial"
+
+				observer.send(value: 2)
+				expect(lastValue) == "2"
+
+				observer.send(value: 3)
+				expect(lastValue) == "3"
+			}
+		}
+
 		describe("reduce(_:_:)") {
 			it("should accumulate one value") {
 				let (baseSignal, observer) = Signal<Int, Never>.pipe()