Rename reduce() to join()

README.md

@@ -29,7 +29,7 @@ Check `tests` and `examples` for more examples.
 - [x] Add Connection.peek() to monitor past messages from all topics
 - [x] Enable resuming PubSub from the "middle" of the computation
 - [x] Add test for .peek()
-- [x] Add "wait until topic X is done" pattern, aka. `Topic.reduce()`
+- [x] Add "wait until topic X is done" pattern, aka. `Topic.join()`
 - [ ] Move tracking of inflight processes/messages to Connection
   - [ ] Use this to build retry mechanism, where any inflight messages are moved back to the respective topics when restarting
   - [ ] But this would require being able to replay a message for a single listener only, which maybe requires a larger redesign of PubSub<>Connection contract than what I wanted to do here

permchain/topic.py

@@ -46,9 +46,9 @@ class Topic(Serializable, Generic[T], ABC):
 
         return RunnableSubscriber(topic=self)
 
-    def reduce(self) -> RunnableReducer[T]:
+    def join(self) -> RunnableReducer[T]:
         if self.name == OUTPUT_TOPIC:
-            raise ValueError("Cannot reduce on output topic")
+            raise ValueError("Cannot join on output topic")
 
         return RunnableReducer(topic=self)
 

tests/test_invoke.py

@@ -169,7 +169,7 @@ def test_invoke_two_processes_two_in_two_out(mocker: MockerFixture):
     assert conn.listeners == {}
 
 
-def test_invoke_two_processes_two_in_reduce_two_out(mocker: MockerFixture):
+def test_invoke_two_processes_two_in_join_two_out(mocker: MockerFixture):
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     add_10_each = mocker.Mock(side_effect=lambda x: [y + 10 for y in x])
     topic_one = Topic("one")
@@ -177,7 +177,7 @@ def test_invoke_two_processes_two_in_reduce_two_out(mocker: MockerFixture):
     chain_one = Topic.IN.subscribe() | add_one | topic_one.publish()
     chain_two = topic_one.subscribe() | add_one | topic_two.publish()
     chain_three = Topic.IN.subscribe() | add_one | topic_two.publish()
-    chain_four = topic_two.reduce() | add_10_each | Topic.OUT.publish()
+    chain_four = topic_two.join() | add_10_each | Topic.OUT.publish()
 
     # Chains can be invoked directly for testing
     assert chain_one.invoke(2) == 3
@@ -199,7 +199,7 @@ def test_invoke_two_processes_two_in_reduce_two_out(mocker: MockerFixture):
     assert conn.listeners == {}
 
 
-def test_invoke_reduce_then_subscribe(mocker: MockerFixture):
+def test_invoke_join_then_subscribe(mocker: MockerFixture):
     add_one = mocker.Mock(side_effect=lambda x: x + 1)
     add_10_each = mocker.Mock(side_effect=lambda x: [y + 10 for y in x])
 
@@ -207,7 +207,7 @@ def test_invoke_reduce_then_subscribe(mocker: MockerFixture):
     topic_two = Topic("two")
 
     chain_one = Topic.IN.subscribe() | add_10_each | topic_one.publish_each()
-    chain_two = topic_one.reduce() | sum | topic_two.publish()
+    chain_two = topic_one.join() | sum | topic_two.publish()
     chain_three = topic_two.subscribe() | add_one | Topic.OUT.publish()
 
     # Chains can be invoked directly for testing