Scan with halt (#1957)

* HALT if scan reducer doesn't change value

* Updated docs to reflect new scan behaviour with HALT

docs/stream.md

@@ -120,11 +120,13 @@ Argument     | Type                 | Required | Description
 
 Creates a new stream with the results of calling the function on every value in the stream with an accumulator and the incoming value.
 
+Note that you can prevent dependent streams from being updated by returning the special value `stream.HALT` inside the accumulator function.
+
 `stream = Stream.scan(fn, accumulator, stream)`
 
 Argument      | Type                             | Required | Description
 ------------- | -------------------------------- | -------- | ---
-`fn`          | `(accumulator, value) -> result` | Yes      | A function that takes an accumulator and value parameter and returns a new accumulator value
+`fn`          | `(accumulator, value) -> result \| HALT` | Yes      | A function that takes an accumulator and value parameter and returns a new accumulator value
 `accumulator` | `any`                            | Yes      | The starting value for the accumulator
 `stream`      | `Stream`                         | Yes      | Stream containing the values
 **returns**   | `Stream`                         |          | Returns a new stream containing the result

stream/stream.js

@@ -117,7 +117,9 @@ function merge(streams) {
 
 function scan(reducer, seed, stream) {
 	var newStream = combine(function (s) {
-		return seed = reducer(seed, s._state.value)
+		var next = reducer(seed, s._state.value)
+		if (next !== HALT) return seed = next
+		return HALT
 	}, [stream])
 
 	if (newStream._state.state === 0) newStream(seed)

stream/tests/test-scan.js

@@ -30,4 +30,36 @@ o.spec("scan", function() {
 		o(result[2]).equals(undefined)
 		o(result[3]).deepEquals({a: 1})
 	})
+
+	o("reducer can return HALT to prevent child updates", function() {
+		var count = 0
+		var action = stream()
+		var store = stream.scan(function (arr, value) {
+			switch (typeof value) {
+				case "number":
+					return arr.concat(value)
+				default:
+					return stream.HALT
+			}
+		}, [], action)
+		var child = store.map(function (p) {
+			count++
+			return p
+		})
+		var result
+
+		action(7)
+		action("11")
+		action(undefined)
+		action({a: 1})
+
+		result = child()
+
+		// check we got the expect result
+		o(result[0]).equals(7)
+
+		// check child received minimum # of updates
+		o(count).equals(2)
+	})
+
 })