From c2a5f5412152f56e16c318218fe6a5d8d68f5e57 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 19 Sep 2023 11:05:37 +0100 Subject: [PATCH] Rename state_id to correlation_id --- permchain/pubsub.py | 7 ++++- tests/test_invoke.py | 65 ++++++++++++++++++++++++++++++++++++++------ 2 files changed, 62 insertions(+), 10 deletions(-) diff --git a/permchain/pubsub.py b/permchain/pubsub.py index 63b4f306b..2b3ddb068 100644 --- a/permchain/pubsub.py +++ b/permchain/pubsub.py @@ -72,7 +72,7 @@ class PubSub(Runnable[Any, Any], ABC): with get_executor_for_config(config) as executor: # Namespace topics for each run, default to run_id, ie. isolated - topic_prefix = str(config.get("state_id") or run_manager.run_id) + topic_prefix = str(config.get("correlation_id") or run_manager.run_id) # Track inflight futures inflight: Set[Future] = set() # Track exceptions @@ -143,6 +143,11 @@ class PubSub(Runnable[Any, Any], ABC): callbacks=run_manager.get_child(), run_name=f"Topic: {process.topic.name}", ), + "correlation_id": self.connection.full_topic_name( + topic_prefix, + process.topic.name, + self.processes.index(process), + ), CONFIG_SEND_KEY: partial(self.connection.send, topic_prefix), CONFIG_GET_KEY: get, }, diff --git a/tests/test_invoke.py b/tests/test_invoke.py index 9aaa67ba4..022034453 100644 --- a/tests/test_invoke.py +++ b/tests/test_invoke.py @@ -68,33 +68,33 @@ def test_invoke_two_processes_in_out_interrupt(mocker: MockerFixture): assert conn.listeners == {} # Then invoke both pubsubs, as a group # The second picks up where the first left off - state_id = uuid4() + correlation_id = uuid4() # invoke() step 1 - assert pubsub_one.invoke(2, {"state_id": state_id}) == [] + assert pubsub_one.invoke(2, {"correlation_id": correlation_id}) == [] # listeners are still cleared, even though state is preserved assert conn.listeners == {} # The log contains all messages published to all topics, in order - assert [{**m, "started_at": None} for m in conn.peek(state_id)] == [ + assert [{**m, "started_at": None} for m in conn.peek(correlation_id)] == [ {"message": 2, "topic_name": "__in__", "started_at": None}, {"message": 3, "topic_name": "one", "started_at": None}, ] # IN, OUT, one assert len(conn.topics) == 3 - topic_one_full_name = conn.full_topic_name(state_id, topic_one.name) + topic_one_full_name = conn.full_topic_name(correlation_id, topic_one.name) # the actual message publishd by chain_one, and a sentinel "end" value assert conn.topics[topic_one_full_name].qsize() == 2 # invoke() step 2 # this picks up where the first left off, and produces same result as # `test_invoke_two_processes_in_out` - assert pubsub_two.invoke(None, {"state_id": state_id}) == [4] + assert pubsub_two.invoke(None, {"correlation_id": correlation_id}) == [4] # listeners are still cleared, even though state is preserved assert conn.listeners == {} # The log contains all messages published to all topics, in order - assert [{**m, "started_at": None} for m in conn.peek(state_id)] == [ + assert [{**m, "started_at": None} for m in conn.peek(correlation_id)] == [ {"message": 2, "topic_name": "__in__", "started_at": None}, {"message": 3, "topic_name": "one", "started_at": None}, {"message": None, "topic_name": "__in__", "started_at": None}, @@ -214,7 +214,7 @@ def test_invoke_join_then_subscribe(mocker: MockerFixture): assert chain_two.invoke([2, 3]) == 5 assert chain_three.invoke(5) == 6 - state_id = uuid4() + correlation_id = uuid4() conn = InMemoryPubSubConnection(clear_on_disconnect=False) pubsub = PubSub((chain_one, chain_two, chain_three), connection=conn) @@ -225,8 +225,8 @@ def test_invoke_join_then_subscribe(mocker: MockerFixture): # Then invoke pubsub # We get a single array result as chain_four waits for all publishers to finish # before operating on all elements published to topic_two as an array - assert pubsub.invoke([2, 3], {"state_id": state_id}) == [26] - assert [{**m, "started_at": None} for m in conn.peek(state_id)] == [ + assert pubsub.invoke([2, 3], {"correlation_id": correlation_id}) == [26] + assert [{**m, "started_at": None} for m in conn.peek(correlation_id)] == [ {"message": [2, 3], "topic_name": "__in__", "started_at": None}, {"message": 12, "topic_name": "one", "started_at": None}, {"message": 13, "topic_name": "one", "started_at": None}, @@ -238,6 +238,53 @@ def test_invoke_join_then_subscribe(mocker: MockerFixture): assert conn.listeners == {} +@pytest.mark.skip("TODO") +def test_invoke_join_then_call_other_pubsub(mocker: MockerFixture): + conn = InMemoryPubSubConnection(clear_on_disconnect=False) + add_one = mocker.Mock(side_effect=lambda x: x + 1) + + inner_pubsub = PubSub( + (Topic.IN.subscribe() | add_one | Topic.OUT.publish(),), connection=conn + ) + + add_10_each = mocker.Mock(side_effect=lambda x: [y + 10 for y in x]) + + topic_one = Topic("one") + topic_two = Topic("two") + + chain_one = Topic.IN.subscribe() | add_10_each | topic_one.publish_each() + chain_two = topic_one.join() | inner_pubsub.map() | topic_two.publish() + chain_three = topic_two.subscribe() | Topic.OUT.publish() + + correlation_id = uuid4() + pubsub = PubSub((chain_one, chain_two, chain_three), connection=conn) + + # Using in-memory conn internals to make assertions about pubsub + # If we start with 0 listeners + assert conn.listeners == {} + + # Then invoke pubsub + # We get a single array result as chain_four waits for all publishers to finish + # before operating on all elements published to topic_two as an array + # assert pubsub.invoke([2, 3], {"correlation_id": correlation_id}) == [[13, 14]] + pubsub.invoke([2, 3], {"correlation_id": correlation_id}) + assert [{**m, "started_at": None} for m in conn.peek(correlation_id)] == [ + {"message": [2, 3], "topic_name": "__in__", "started_at": None}, + {"message": 12, "topic_name": "one", "started_at": None}, + {"message": 13, "topic_name": "one", "started_at": None}, + {"message": 25, "topic_name": "two", "started_at": None}, + {"message": 26, "topic_name": "__out__", "started_at": None}, + {"message": [2, 3], "started_at": None, "topic_name": "__in__"}, + {"message": 12, "started_at": None, "topic_name": "one"}, + {"message": 13, "started_at": None, "topic_name": "one"}, + {"message": 25, "started_at": None, "topic_name": "two"}, + {"message": 26, "started_at": None, "topic_name": "__out__"}, + ] + + # After invoke returns the listeners were cleaned up + assert conn.listeners == {} + + def test_invoke_two_processes_one_in_two_out(mocker: MockerFixture): add_one = mocker.Mock(side_effect=lambda x: x + 1) topic_one = Topic("one")