added op.Replace

broqer/op/__init__.py

@@ -12,6 +12,7 @@
 from .partition import Partition, partition
 from .pluck import Pluck, pluck
 from .reduce import Reduce, reduce
+from .replace import Replace, replace
 from .sink import Sink, sink
 from .sliding_window import SlidingWindow, sliding_window
 from .switch import Switch, switch
@@ -32,11 +33,11 @@
     'catch_exception', 'CombineLatest', 'combine_latest', 'Distinct',
     'distinct', 'Filter', 'filter', 'FromIterable', 'Just', 'Map', 'map',
     'Merge', 'merge', 'Pack', 'pack', 'Partition', 'partition', 'Pluck',
-    'pluck', 'Reduce', 'reduce', 'Sink', 'sink', 'SlidingWindow',
-    'sliding_window', 'Switch', 'switch', 'Unpack', 'unpack', 'Debounce',
-    'debounce', 'Delay', 'delay', 'FromPolling', 'Sample', 'sample',
-    'MapAsync', 'map_async', 'Mode', 'MapThreaded', 'map_threaded',
-    'Throttle', 'throttle', 'ToFuture', 'to_future'
+    'pluck', 'Reduce', 'reduce', 'Replace', 'replace', 'Sink', 'sink',
+    'SlidingWindow', 'sliding_window', 'Switch', 'switch', 'Unpack', 'unpack',
+    'Debounce', 'debounce', 'Delay', 'delay', 'FromPolling',
+    'Sample', 'sample', 'MapAsync', 'map_async', 'Mode', 'MapThreaded',
+    'map_threaded', 'Throttle', 'throttle', 'ToFuture', 'to_future'
 ]
 
 # TODO operators

broqer/op/replace.py

@@ -0,0 +1,23 @@
+"""
+When this operators gets emitted it's emitting a defined value
+"""
+
+from typing import Any
+
+from broqer import Publisher
+
+from ._operator import Operator, build_operator
+
+
+class Replace(Operator):
+    def __init__(self, publisher: Publisher, *args: Any) -> None:
+        Operator.__init__(self, publisher)
+
+        self._args = args
+
+    def emit(self, *args: Any, who: Publisher) -> None:
+        assert who == self._publisher, 'emit from non assigned publisher'
+        self._emit(self._args)
+
+
+replace = build_operator(Replace)