From e093d989a03c42ff31ff8dac28a4afb7c1415e59 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Wed, 13 Sep 2023 22:44:56 +0100 Subject: [PATCH] Rename reduce() to join() --- README.md | 2 +- permchain/topic.py | 4 ++-- tests/test_invoke.py | 8 ++++---- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index ea0925f18..4629bce5d 100644 --- a/README.md +++ b/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 diff --git a/permchain/topic.py b/permchain/topic.py index 4a98fbc5c..b1369ff76 100644 --- a/permchain/topic.py +++ b/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) diff --git a/tests/test_invoke.py b/tests/test_invoke.py index 9cd2af9a7..967f90284 100644 --- a/tests/test_invoke.py +++ b/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