Rename state_id to correlation_id

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,
                     },

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")