Function to guarantee PUB/SUB channel in SYNC_PUB

This function will guarantee the PUB/SUB channel is already up and
running after returning, ready for sending data right away.

Author: Guillermo Alonso

osbrain/agent.py

@@ -912,6 +912,9 @@ def _handle_async_requests(self, data):
         else:
             handler(self, response)
 
+    def ugly(self, alias, handlers):
+        self._subscribe(alias, handlers)
+
     def _subscribe(self, alias: str, handlers: Dict[Union[bytes, str], Any]):
         """
         Subscribe the agent to another agent.
@@ -965,6 +968,9 @@ def set_attr(self, **kwargs):
     def get_attr(self, name):
         return getattr(self, name)
 
+    def del_attr(self, name):
+        delattr(self, name)
+
     def set_method(self, *args, **kwargs):
         """
         Set object methods.

osbrain/helper.py

@@ -192,3 +192,46 @@ def wait_agent_attr(agent, name='received', length=None, data=None, value=None,
             break
         time.sleep(0.01)
     return False
+
+
+def synchronize_sync_pub(server, server_alias, server_handler, client,
+                         client_alias, client_handler):
+    '''
+    Create a SYNC_PUB/SYNC_SUB channel and connect both agents.
+
+    Make sure they have stablished the PUB/SUB communication within the
+    SYNC_PUB/SYNC_SUB channel before returning. This will guarantee that
+    no PUB messages are lost.
+    '''
+    def assert_receive(agent, message, topic=None):
+        try:
+            agent.get_attr('_tmp_attr')
+            agent._tmp_attr = True
+        except AttributeError:  # Attribute already deleted
+            pass
+
+    addr = server.bind('SYNC_PUB', alias=server_alias, handler=server_handler)
+
+    client.set_attr(_tmp_attr=False)
+
+    client.connect(addr, alias=client_alias, handler=assert_receive)
+
+    # Send messages through the PUB socket until the client receives them
+    server.each(0.1, 'send', addr, 'Synchronize', alias='_tmp_timer')
+
+    assert wait_agent_attr(client, name='_tmp_attr', value=True, timeout=5)
+
+    server.stop_timer('_tmp_timer')
+
+    # The following is an ugly hack to get the uuid used as the alias for the
+    # SUB socket of the client in the SYNC_PUB channel.
+    channel = client.addr(client_alias)
+    client_addr = channel.twin().sender.twin()
+    addr_to_access_uuid = client.addr(client_addr)
+    client_async_req_uuid = client.get_attr('_async_req_uuid')
+    uuid = client_async_req_uuid[addr_to_access_uuid]
+
+    # Set the handler passed as a parameter, now that connection is guaranteed
+    client.ugly(uuid, client_handler)
+
+    client.del_attr('_tmp_attr')

osbrain/tests/test_helper.py

@@ -9,11 +9,45 @@
 from osbrain.helper import agent_dies
 from osbrain.helper import attribute_match_all
 from osbrain.helper import wait_agent_attr
+from osbrain.helper import synchronize_sync_pub
+from osbrain.address import AgentChannel
 
 from common import nsproxy  # pragma: no flakes
 from common import agent_logger  # pragma: no flakes
 
 
+def receive(agent, response):
+    agent.received.append(response)
+
+
+def test_synchronize_sync_pub(nsproxy):
+    """
+    All publications in SYNC_PUB/SYNC_SUB connections stablished through
+    `synchronize_sync_pub` should be received in the other end.
+    """
+    server = run_agent('server')
+    client = run_agent('client')
+
+    client.set_attr(received=[])
+
+    # Create a SYNC_PUB channel and guarantee the PUB/SUB is stablished
+    synchronize_sync_pub(server, 'sync_pub', receive,
+                         client, 'sync_sub', receive)
+
+    # Send the message only once
+    server.send('sync_pub', 'Hello')
+
+    assert wait_agent_attr(client, name='received', data='Hello')
+
+    # Check that no temporary attributes remain
+    assert 'Synchronize' not in client.get_attr('received')
+
+    with pytest.raises(AttributeError):
+        client.get_attr('_tmp_attr')
+
+    assert '_tmp_timer' not in server.list_timers()
+
+
 def test_agent_dies(nsproxy):
     """
     The function `agent_dies` should return `False` if the agent does not die