Added scan operator

Author: TheWalruzz

README.md

@@ -205,6 +205,53 @@ Sx.from(text_edit.text_changed).debounce(0.25).subscribe(func(): print(text_edit
 Sx.from(text_edit.text_changed).throttle(0.25).subscribe(func(): print(text_edit.text)) # text will be printed every 0.25 seconds when typing continuously.
 ```
 
+
+### Scan operator
+Sx allows for scanning and buffering incoming values inside a stateful operator. This operator behavves similar to reduce() in functional programming.
+
+```gdscript
+signal numbers(value: int)
+
+Sx.from(numbers).scan(
+	func(acc: int, value: int):
+		return acc + value,
+	0
+).subscribe(func(value: int): print(value))
+
+numbers.emit(3)
+numbers.emit(2)
+numbers.emit(7)
+
+# result:
+#	3
+#	5
+#	12
+```
+
+This can also be useful if you want to collect previous emissions:
+```gdscript
+signal numbers(value: int)
+
+Sx.from(numbers).scan(
+	func(acc: Array[int], value: int):
+		acc.append(value)
+		return acc,
+	[]
+).subscribe(func(value: Array[int]): print(value))
+
+numbers.emit(3)
+numbers.emit(2)
+numbers.emit(7)
+
+# result:
+#	[3]
+#	[3, 2]
+#	[3, 2, 7]
+```
+
+Due to the way the reducing function works, while multiple signal arguments will be passed to the function after the accumulator, this function should return one value.
+Subsequent operator after `scan` will receive ONE argument.
+
 ### On complete callback
 When you're subscribing, you can set an optional callback that will be fired when the signal completes (either naturally, or when signal is disposed).
 
@@ -455,6 +502,7 @@ for key in dict:
 * map
 * merge
 * merge_from
+* scan
 * skip
 * skip_while
 * start_with

addons/signal_extensions/operators/sx_scan_operator.gd

@@ -0,0 +1,21 @@
+extends Sx.Operator
+
+
+var _callable: Callable
+var _state: Variant
+var _initial_value: Variant
+
+
+func _init(callable: Callable, initial_value: Variant):
+	_callable = callable
+	_initial_value = initial_value
+	_state = initial_value
+	
+
+func clone() -> Sx.Operator:
+	return Sx.ScanOperator.new(_callable, _initial_value)
+
+
+func evaluate(args: Array[Variant]) -> Sx.OperatorResult:
+	_state = _callable.bindv(args).call(_state)
+	return Sx.OperatorResult.new(true, [_state])

addons/signal_extensions/plugin.cfg

@@ -3,5 +3,5 @@
 name="SignalExtensions"
 description="Rx-like extensions for Godot's signals."
 author="Paweł \"TheWalruzz\" Radej"
-version="1.7.0"
+version="1.8.0"
 script="plugin.gd"

addons/signal_extensions/signals/sx_signal.gd

@@ -54,15 +54,6 @@ func debounce(debounce_time: float, process_always := true, process_in_physics :
 	var cloned := clone()
 	cloned._operators.append(Sx.DebounceOperator.new(debounce_time, process_always, process_in_physics, ignore_timescale))
 	return cloned
-	
-
-## Throttles the signal by [b]throttle_time[/b].
-## This method has arguments consistent with Godot's [method SceneTree.create_timer] method.
-## See that method's documentation for more information.
-func throttle(throttle_time: float, process_always := true, process_in_physics := false, ignore_timescale := false) -> SxSignal:
-	var cloned := clone()
-	cloned._operators.append(Sx.ThrottleOperator.new(throttle_time, process_always, process_in_physics, ignore_timescale))
-	return cloned
 
 
 ## Delays item emission by [b]duration[/b]. 
@@ -121,6 +112,14 @@ func merge_from(signals: Array[Signal]) -> SxSignal:
 	return merge(converted)
 
 
+## Scans the emitted items and can reduce them to a single value over time.
+## Reducing function: func(acc: ACC_TYPE, ...args: Array[Variant]) -> ACC_TYPE
+func scan(callable: Callable, initial_value: Variant) -> SxSignal:
+	var cloned := clone()
+	cloned._operators.append(Sx.ScanOperator.new(callable, initial_value))
+	return cloned
+
+
 ## Skips the first [b]item_count[/b] items from the sequence.
 func skip(item_count: int) -> SxSignal:
 	var cloned := clone()
@@ -162,6 +161,15 @@ func take_while(callable: Callable) -> SxSignal:
 	operator.dispose_callback = cloned._set_dispose
 	cloned._operators.append(operator)
 	return cloned
+	
+	
+## Throttles the signal by [b]throttle_time[/b].
+## This method has arguments consistent with Godot's [method SceneTree.create_timer] method.
+## See that method's documentation for more information.
+func throttle(throttle_time: float, process_always := true, process_in_physics := false, ignore_timescale := false) -> SxSignal:
+	var cloned := clone()
+	cloned._operators.append(Sx.ThrottleOperator.new(throttle_time, process_always, process_in_physics, ignore_timescale))
+	return cloned
 
 
 func _clone() -> SxSignal:

addons/signal_extensions/sx_autoload.gd

@@ -11,6 +11,7 @@ const ElementAtOperator := preload("res://addons/signal_extensions/operators/sx_
 const FilterOperator := preload("res://addons/signal_extensions/operators/sx_filter_operator.gd")
 const FirstOperator := preload("res://addons/signal_extensions/operators/sx_first_operator.gd")
 const MapOperator := preload("res://addons/signal_extensions/operators/sx_map_operator.gd")
+const ScanOperator := preload("res://addons/signal_extensions/operators/sx_scan_operator.gd")
 const SkipOperator := preload("res://addons/signal_extensions/operators/sx_skip_operator.gd")
 const SkipWhileOperator := preload("res://addons/signal_extensions/operators/sx_skip_while_operator.gd")
 const TakeOperator := preload("res://addons/signal_extensions/operators/sx_take_operator.gd")

test/operators/sx_scan_operator_test.gd

@@ -0,0 +1,42 @@
+# GdUnit generated TestSuite
+class_name SxScanOperatorTest
+extends GdUnitTestSuite
+@warning_ignore('unused_parameter')
+@warning_ignore('return_value_discarded')
+
+# TestSuite generated from
+const __source = 'res://addons/signal_extensions/operators/sx_scan_operator.gd'
+
+
+var operator: Sx.ScanOperator
+
+
+func test_evaluate_returning_variant() -> void:
+	operator = Sx.ScanOperator.new(func(acc, value): return acc + value, 0)
+	var result := operator.evaluate([3])
+	assert_bool(result.ok).is_true()
+	assert_array(result.args).has_size(1).contains([3])
+	result = operator.evaluate([2])
+	assert_bool(result.ok).is_true()
+	assert_array(result.args).has_size(1).contains([5])
+	
+	
+func test_evaluate_returning_array() -> void:
+	operator = Sx.ScanOperator.new(func(acc, value):
+		acc.append(value)
+		return acc, 
+	[])
+	var result := operator.evaluate([3])
+	assert_bool(result.ok).is_true()
+	assert_array(result.args).has_size(1).contains([[3]])
+	result = operator.evaluate([2])
+	assert_bool(result.ok).is_true()
+	assert_array(result.args).has_size(1).contains([[3, 2]])
+
+
+func test_cloning() -> void:
+	operator = Sx.ScanOperator.new(func(acc, value): return acc + value, 0)
+	var cloned := operator.clone()
+	assert_object(cloned._callable).is_same(operator._callable)
+	assert_object(cloned._initial_value).is_same(operator._initial_value)
+	assert_object(cloned).is_not_same(operator)