Repository: opensistemas-hub/osbrain
Pull request #110 (2017-04-10 12:17:06)

Make bytes the default type for topics and UUIDs
Implements #91.

codecov[bot]: # [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=h1) Report
> Merging [#110](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=desc) into [master](https://codecov.io/gh/opensistemas-hub/osbrain/commit/c8bb10397235bb35715eeeee493e11fae20e5fdf?src=pr&el=desc) will **decrease** coverage by `0.14%`.
> The diff coverage is `95.23%`.

[![Impacted file tree graph](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110/graphs/tree.svg?height=150&width=650&src=pr&token=LfvMoNOrOc)](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #110      +/-   ##
==========================================
- Coverage   97.43%   97.29%   -0.15%     
==========================================
  Files           8        8              
  Lines        1171     1181      +10     
  Branches      173      178       +5     
==========================================
+ Hits         1141     1149       +8     
  Misses         21       21              
- Partials        9       11       +2
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/address.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=tree#diff-b3NicmFpbi9hZGRyZXNzLnB5) | `100% <100%> (ø)` | :arrow_up: |
| [osbrain/common.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=tree#diff-b3NicmFpbi9jb21tb24ucHk=) | `100% <100%> (ø)` | :arrow_up: |
| [osbrain/logging.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=tree#diff-b3NicmFpbi9sb2dnaW5nLnB5) | `100% <100%> (ø)` | :arrow_up: |
| [osbrain/agent.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=tree#diff-b3NicmFpbi9hZ2VudC5weQ==) | `95.5% <94.44%> (-0.28%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=footer). Last update [c8bb103...8e3509b](https://codecov.io/gh/opensistemas-hub/osbrain/pull/110?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

Patch:
diff --git a/.coveragerc b/.coveragerc
index 0708ed6..62c64c4 100644
--- a/.coveragerc
+++ b/.coveragerc
@@ -10,6 +10,7 @@ exclude_lines =
     def __hash__
     if .debug:
     raise NotImplementedError
+    if sys.version_info
     if __name__ == .__main__.:
 omit =
     setup.py
diff --git a/osbrain/address.py b/osbrain/address.py
index 36f2199..f3e5cdf 100644
--- a/osbrain/address.py
+++ b/osbrain/address.py
@@ -1,11 +1,12 @@
 """
 Implementation of address-related features.
 """
-from uuid import uuid4
 from ipaddress import ip_address
 
 import zmq
 
+from .common import unique_identifier
+
 
 def address_to_host_port(addr):
     """
@@ -396,7 +397,7 @@ class AgentChannel():
             receiver.transport if receiver else sender.transport
         self.serializer = \
             receiver.serializer if receiver else sender.serializer
-        self.uuid = uuid4().hex
+        self.uuid = unique_identifier()
         # Set up pairs
         if sender:
             self.sender.channel = self
diff --git a/osbrain/agent.py b/osbrain/agent.py
index 097aa7f..8dd5e21 100644
--- a/osbrain/agent.py
+++ b/osbrain/agent.py
@@ -12,7 +12,9 @@ import signal
 import sys
 import time
 import types
-from uuid import uuid4
+from typing import Any
+from typing import Dict
+from typing import Union
 
 import dill
 import Pyro4
@@ -27,8 +29,10 @@ from .common import LogLevel
 from .common import repeat
 from .common import after
 from .common import get_linger
+from .common import unique_identifier
 from .address import AgentAddress
 from .address import AgentAddressKind
+from .address import AgentAddressSerializer
 from .address import AgentChannel
 from .address import address_to_host_port
 from .address import guess_kind
@@ -39,14 +43,6 @@ from .proxy import NSProxy
 TOPIC_SEPARATOR = b'\x80'
 
 
-def str2bytes(message):
-    return message.encode('ascii')
-
-
-def bytes2str(message):
-    return message.decode('ascii')
-
-
 def serialize_message(message, serializer):
     """
     Check if a message needs to be serialized and do it if that is the
@@ -68,7 +64,7 @@ def serialize_message(message, serializer):
     if serializer == 'pickle':
         return pickle.dumps(message, -1)
     if serializer == 'json':
-        return str2bytes(json.dumps(message))
+        return json.dumps(message).encode()
     if serializer == 'raw':
         return message
     raise ValueError('Serializer not supported for serialization')
@@ -95,13 +91,14 @@ def deserialize_message(message, serializer):
     if serializer == 'pickle':
         return pickle.loads(message)
     if serializer == 'json':
-        return json.loads(bytes2str(bytes(message)))
+        return json.loads(bytes(message).decode())
     if serializer == 'raw':
         return message
     raise ValueError('Serializer not supported for deserialization')
 
 
-def compose_message(message, topic, serializer):
+def compose_message(message: bytes, topic: bytes,
+                    serializer: AgentAddressSerializer) -> bytes:
     """
     Compose a message and leave it ready to be sent through a socket.
 
@@ -110,20 +107,17 @@ def compose_message(message, topic, serializer):
 
     Parameters
     ----------
-    message : bytes
+    message
         Message to be composed.
-    topic : str
+    topic
         Topic to combine the message with.
-    serializer : AgentAddressSerializer
+    serializer
         Serialization for the message part.
 
     Returns
     -------
-    bytes
         The bytes representation of the final message to be sent.
     """
-    assert isinstance(topic, str), 'Topic must be of `str` type!'
-    topic = str2bytes(topic)
     if serializer.requires_separator:
         return topic + TOPIC_SEPARATOR + message
     return topic + message
@@ -196,7 +190,7 @@ class Agent():
         Set to `True` if the agent is running (executing the main loop).
     """
     def __init__(self, name=None, host=None, serializer=None, transport=None):
-        self.uuid = uuid4().hex
+        self.uuid = unique_identifier()
         self.name = name
         self.host = host
         if not self.host:
@@ -341,7 +335,7 @@ class Agent():
         timer = repeat(period, self._loopback,
                        'EXECUTE_METHOD', (method, args, kwargs))
         if not alias:
-            alias = uuid4().hex
+            alias = unique_identifier()
         self._timer[alias] = timer
         return alias
 
@@ -372,7 +366,7 @@ class Agent():
         timer = after(delay, self._loopback,
                       'EXECUTE_METHOD', (method, args, kwargs))
         if not alias:
-            alias = uuid4().hex
+            alias = unique_identifier()
         self._timer[alias] = timer
         return alias
 
@@ -732,7 +726,7 @@ class Agent():
                 socket.bind('tcp://%s' % (addr))
         else:
             if not addr:
-                addr = str(uuid4())
+                addr = str(unique_identifier())
             if transport == 'ipc':
                 addr = config['IPC_DIR'] / addr
             socket.bind('%s://%s' % (transport, addr))
@@ -833,7 +827,7 @@ class Agent():
         self._connect_and_register(pull_address.twin(), alias=alias,
                                    register_as=channel)
         # Create socket for receiving responses
-        uuid = uuid4().hex
+        uuid = unique_identifier()
         addr = self.bind('PULL', alias=uuid,
                          handler=self._handle_async_requests)
         self._async_req_uuid[pull_address] = uuid
@@ -865,10 +859,12 @@ class Agent():
         # Create socket for receiving responses
         pub_address = channel.sender
         assert pub_address.kind == 'PUB'
-        uuid = uuid4().hex
+        uuid = unique_identifier()
         topic_handlers = {}
         if isinstance(handler, dict):
             for key, value in handler.items():
+                if isinstance(key, str):
+                    key = key.encode()
                 topic_handlers[channel.uuid + key] = value
         else:
             topic_handlers[channel.uuid] = handler
@@ -908,15 +904,15 @@ class Agent():
         handler = self._pending_requests.pop(uuid)
         handler(self, response)
 
-    def subscribe(self, alias, handlers):
+    def subscribe(self, alias: str, handlers: Dict[Union[bytes, str], Any]):
         """
         Subscribe the agent to another agent.
 
         Parameters
         ----------
-        alias : str
+        alias
             Alias of the new subscriber socket.
-        handlers : dict
+        handlers
             A dictionary in which the keys represent the different topics
             and the values the actual handlers. If ,instead of a dictionary,
             a single handler is given, it will be used to subscribe the agent
@@ -924,12 +920,17 @@ class Agent():
         """
         if not isinstance(handlers, dict):
             handlers = {'': handlers}
-        for topic in handlers.keys():
-            assert isinstance(topic, str), 'Topic must be of type `str`!'
-            topic = str2bytes(topic)
+        # Convert all topics to bytes
+        curated_handlers = {}
+        for topic, value in handlers.items():
+            if isinstance(topic, str):
+                topic = topic.encode()
+            curated_handlers[topic] = value
+        # Subscribe to topics
+        for topic in curated_handlers.keys():
             self.socket[alias].setsockopt(zmq.SUBSCRIBE, topic)
         # Reset handlers
-        self._set_handler(self.socket[alias], handlers)
+        self._set_handler(self.socket[alias], curated_handlers)
 
     def iddle(self):
         """
@@ -1237,17 +1238,16 @@ class Agent():
 
         message = self._process_sub_message(addr.serializer, data)
 
-        for str_topic in handlers:
-            btopic = str2bytes(str_topic)
-            if not data.startswith(btopic):
+        for topic in handlers:
+            if not data.startswith(topic):
                 continue
             # Call the handler (with or without the topic)
-            handler = handlers[str_topic]
+            handler = handlers[topic]
             nparams = len(inspect.signature(handler).parameters)
             if nparams == 2:
                 handler(self, message)
             elif nparams == 3:
-                handler(self, message, str_topic)
+                handler(self, message, topic)
 
     def send(self, address, message, topic=None, handler=None, wait=None,
              on_error=None):
@@ -1295,6 +1295,8 @@ class Agent():
         if address.kind == 'PUB':
             if topic is None:
                 topic = ''
+            if isinstance(topic, str):
+                topic = topic.encode()
             message = compose_message(message=message,
                                       topic=topic,
                                       serializer=address.serializer)
@@ -1316,7 +1318,7 @@ class Agent():
         if kind == 'SYNC_SUB':
             address = channel.receiver
             address_uuid = self._async_req_uuid[address]
-            request_uuid = uuid4().hex
+            request_uuid = unique_identifier()
             self._pending_requests[request_uuid] = handler
             message = (address_uuid, request_uuid, message)
             self._send_address(channel.sender, message)
@@ -1327,7 +1329,7 @@ class Agent():
     def _send_channel_async_rep(self, channel, message, wait, on_error):
         address = channel.receiver
         address_uuid = self._async_req_uuid[address]
-        request_uuid = uuid4().hex
+        request_uuid = unique_identifier()
         self._pending_requests[request_uuid] = \
             self._async_req_handler[address_uuid]
         receiver_address = self.address[address_uuid]
@@ -1338,10 +1340,12 @@ class Agent():
         self._wait_received(wait, uuid=request_uuid, on_error=on_error)
 
     def _send_channel_sync_pub(self, channel, message, topic, general=True):
-        if general:
-            topic = channel.uuid + topic
         message = serialize_message(message=message,
                                     serializer=channel.serializer)
+        if isinstance(topic, str):
+            topic = topic.encode()
+        if general:
+            topic = channel.uuid + topic
         message = compose_message(message=message,
                                   topic=topic,
                                   serializer=channel.serializer)
diff --git a/osbrain/common.py b/osbrain/common.py
index b5f1dcf..c1e6c11 100644
--- a/osbrain/common.py
+++ b/osbrain/common.py
@@ -5,10 +5,22 @@ import time
 from threading import Event
 from threading import Thread
 import traceback
+from uuid import uuid4
 
 from . import config
 
 
+def unique_identifier() -> bytes:
+    """
+    Returns
+    -------
+        A unique identifier that is safe to use in PUB-SUB communication
+        patterns (i.e.: does not contain the `osbrain.TOPIC_SEPARATOR`
+        character).
+    """
+    return uuid4().hex.encode()
+
+
 def format_exception():
     """
     Represent a traceback exception as a string in which all lines start
diff --git a/osbrain/logging.py b/osbrain/logging.py
index e549a3a..a195c2d 100644
--- a/osbrain/logging.py
+++ b/osbrain/logging.py
@@ -38,6 +38,7 @@ class Logger(Agent):
         """
         Handle incoming log messages.
         """
+        topic = topic.decode()
         if topic == 'ERROR':
             self.log_history_error.append(message)
         elif topic == 'WARNING':
diff --git a/setup.py b/setup.py
index 5776eaf..d80b81e 100644
--- a/setup.py
+++ b/setup.py
@@ -2,6 +2,7 @@
 Setup module.
 """
 import re
+import sys
 from os.path import join as pjoin
 from setuptools import setup
 
@@ -10,6 +11,11 @@ with open(pjoin('osbrain', '__init__.py')) as f:
     line = next(l for l in f if l.startswith('__version__'))
     version = re.match('__version__ = [\'"]([^\'"]+)[\'"]', line).group(1)
 
+# While Python 3.4 is supported...
+install_requires_compat = []
+if sys.version_info < (3, 5):
+    install_requires_compat = ['typing']
+
 setup(
     name='osbrain',
     version=version,
@@ -35,7 +41,11 @@ setup(
     ],
     keywords='osbrain multi-agent system',
     packages=['osbrain'],
-    install_requires=['Pyro4>=4.42', 'pyzmq>=15.2.0', 'dill>=0.2.0'],
+    install_requires=[
+        'Pyro4>=4.42',
+        'pyzmq>=15.2.0',
+        'dill>=0.2.0',
+    ] + install_requires_compat,
     extras_require={
         'dev': [],
         'test': ['tox'],


Test patch:
diff --git a/osbrain/tests/test_address.py b/osbrain/tests/test_address.py
index 871a116..067a71d 100644
--- a/osbrain/tests/test_address.py
+++ b/osbrain/tests/test_address.py
@@ -6,6 +6,7 @@ from collections import namedtuple
 import zmq
 import pytest
 
+from osbrain.agent import TOPIC_SEPARATOR
 from osbrain.address import address_to_host_port
 from osbrain.address import guess_kind
 from osbrain.address import SocketAddress
@@ -239,6 +240,21 @@ def test_agentchannelkind_value_error():
         AgentChannelKind('FOO')
 
 
+def test_agentchannel_uuid():
+    """
+    An AgentChannel's unique identifier must never contain the
+    `osbrain.TOPIC_SEPARATOR` to avoid errors when using this identifier as
+    topic in PUB-SUB communication patterns.
+    """
+    sender = AgentAddress('ipc', 'addr0', 'PUB', 'server', 'pickle')
+    receiver = AgentAddress('ipc', 'addr0', 'PULL', 'server', 'pickle')
+    for i in range(1000):
+        channel = AgentChannel('SYNC_PUB', sender=sender, receiver=receiver)
+        identifier = channel.uuid
+        assert isinstance(identifier, bytes)
+        assert TOPIC_SEPARATOR not in identifier
+
+
 def test_agentchannel_async_rep():
     """
     Test basic ASYNC_REP AgentChannel operations: initialization, equivalence
diff --git a/osbrain/tests/test_agent.py b/osbrain/tests/test_agent.py
index 021afe0..62c952a 100644
--- a/osbrain/tests/test_agent.py
+++ b/osbrain/tests/test_agent.py
@@ -38,7 +38,7 @@ def test_agent_uuid():
     N = 1000
     bunch = set(Agent().uuid for i in range(N))
     assert len(bunch) == N
-    assert all(isinstance(identifier, str) for identifier in bunch)
+    assert all(isinstance(identifier, bytes) for identifier in bunch)
 
 
 def test_early_agent_proxy(nsproxy):
diff --git a/osbrain/tests/test_agent_serialization.py b/osbrain/tests/test_agent_serialization.py
index db3525c..aba0914 100644
--- a/osbrain/tests/test_agent_serialization.py
+++ b/osbrain/tests/test_agent_serialization.py
@@ -20,23 +20,22 @@ def set_received(agent, message, topic=None):
     agent.received = message
 
 
-def test_message_composer():
+def test_compose_message():
     """
     Test correct message composing.
     """
     message = b'message'
-    topic = 'test topic'
-    topic_bytes = b'test topic'
+    topic = b'test topic'
 
     # Basic composing
     for serializer in AgentAddressSerializer.SERIALIZER_SIMPLE:
         serializer = AgentAddressSerializer(serializer)
         assert compose_message(message, topic, serializer) \
-            == topic_bytes + message
+            == topic + message
     for serializer in AgentAddressSerializer.SERIALIZER_SEPARATOR:
         serializer = AgentAddressSerializer(serializer)
         assert compose_message(message, topic, serializer) \
-            == topic_bytes + TOPIC_SEPARATOR + message
+            == topic + TOPIC_SEPARATOR + message
 
     # Raise with wrong serializer
     with pytest.raises(Exception):
diff --git a/osbrain/tests/test_agent_sync_publications.py b/osbrain/tests/test_agent_sync_publications.py
index f352e86..fa1e27d 100644
--- a/osbrain/tests/test_agent_sync_publications.py
+++ b/osbrain/tests/test_agent_sync_publications.py
@@ -15,12 +15,7 @@ from osbrain.helper import wait_agent_attr
 from common import nsproxy  # pragma: no flakes
 
 
-class Server(Agent):
-    def on_init(self):
-        self.count = 0
-        self.received = []
-        self.bind('SYNC_PUB', alias='publish', handler='reply')
-
+class BaseServer(Agent):
     def publish(self):
         self.count += 1
         self.send('publish', self.count)
@@ -29,12 +24,27 @@ class Server(Agent):
         self.count += 1
         self.send('publish', str(self.count), topic='positive')
         self.send('publish', str(-self.count), topic='negative')
+        self.send('publish', 'bytes...', topic=b'\xeb')
 
     def reply(self, request):
         self.received.append((self.count, request))
         return self.count + 0.5
 
 
+class Server(BaseServer):
+    def on_init(self):
+        self.count = 0
+        self.received = []
+        self.bind('SYNC_PUB', alias='publish', handler='reply')
+
+
+class PubServer(BaseServer):
+    def on_init(self):
+        self.count = 0
+        self.received = []
+        self.bind('PUB', alias='publish', handler='reply')
+
+
 class ServerYield(Server):
     def reply(self, request):
         yield self.count + 0.5
@@ -66,24 +76,72 @@ def on_error(agent):
     agent.error_log.append('error')
 
 
-def test_simple_pub(nsproxy):
+@pytest.mark.parametrize('server', [Server, PubServer])
+def test_simple_pub_single_handler(nsproxy, server):
     """
-    SYNC_PUB should work just like a normal PUB.
+    SYNC_PUB should work just like a normal PUB. This test checks normal
+    behavior when using a single handler (i.e.: no filtering).
 
     When clients connect to a SYNC_PUB server, as long as they do not make
     requests, this communication pattern should behave exactly like a normal
     PUB-SUB pattern.
     """
-    server = run_agent('server', base=Server)
+    server = run_agent('server', base=server)
+    alltopics = run_agent('alltopics', base=Client)
+
+    # Connect clients
+    addr = server.addr('publish')
+    addr_alltopics = alltopics.connect(addr, handler=receive)
+    assert addr_alltopics == addr.twin()
+
+    # Publish from server
+    server.each(0, 'publish_str')
+
+    # Wait for clients to receive some data
+    N = 10
+    assert wait_agent_attr(alltopics, length=N)
+
+    # alltopics
+    received = [int(x) for x in alltopics.get_attr('received')
+                if x != 'bytes...']
+    assert len(received) >= N
+    for i in range(2, len(received)):
+        if received[i] > 0:
+            if received[i - 1] == 'bytes...':
+                assert received[i - 2] == 1 - received[i]
+            else:
+                assert received[i - 1] == 1 - received[i]
+        else:
+            if received[i - 1] == 'bytes...':
+                assert received[i - 2] == -received[i]
+            else:
+                assert received[i - 1] == -received[i]
+
+
+@pytest.mark.parametrize('server', [Server, PubServer])
+def test_simple_pub_dict_handler(nsproxy, server):
+    """
+    SYNC_PUB should work just like a normal PUB. This test checks normal
+    behavior when using multiple handlers (i.e.: filtering).
+
+    When clients connect to a SYNC_PUB server, as long as they do not make
+    requests, this communication pattern should behave exactly like a normal
+    PUB-SUB pattern.
+    """
+    server = run_agent('server', base=server)
     both = run_agent('both', base=Client)
     positive = run_agent('positive', base=Client)
+    bytestopic = run_agent('bytestopic', base=Client)
 
     # Connect clients
     addr = server.addr('publish')
-    addr_both = both.connect(addr, handler=receive)
+    addr_both = both.connect(addr, handler={'positive': receive,
+                                            'negative': receive})
     addr_positive = positive.connect(addr, handler={'positive': receive})
+    addr_bytestopic = bytestopic.connect(addr, handler={b'\xeb': receive})
     assert addr_both == addr.twin()
     assert addr_positive == addr.twin()
+    assert addr_bytestopic == addr.twin()
 
     # Publish from server
     server.each(0, 'publish_str')
@@ -92,6 +150,7 @@ def test_simple_pub(nsproxy):
     N = 10
     assert wait_agent_attr(both, length=N)
     assert wait_agent_attr(positive, length=N)
+    assert wait_agent_attr(bytestopic, length=N)
 
     # both
     received = [int(x) for x in both.get_attr('received')]
@@ -107,6 +166,11 @@ def test_simple_pub(nsproxy):
     assert len(received) >= N
     assert received == list(range(received[0], received[-1] + 1))
 
+    # bytestopic
+    received = bytestopic.get_attr('received')
+    assert len(received) >= N
+    assert all(x == 'bytes...' for x in received)
+
 
 @pytest.mark.parametrize('server', [Server, ServerYield])
 def test_request(nsproxy, server):
diff --git a/osbrain/tests/test_common.py b/osbrain/tests/test_common.py
index 2d2a385..3626fd0 100644
--- a/osbrain/tests/test_common.py
+++ b/osbrain/tests/test_common.py
@@ -3,9 +3,23 @@ Test file for common module.
 """
 import pytest
 
+from osbrain.agent import TOPIC_SEPARATOR
+from osbrain.common import unique_identifier
 from osbrain.common import LogLevel
 
 
+def test_unique_identifier():
+    """
+    The `unique_identifier` function must return an identifier that does not
+    contain the `osbrain.TOPIC_SEPARATOR`, so that it is safe to use in
+    PUB-SUB communication patterns.
+    """
+    for i in range(1000):
+        identifier = unique_identifier()
+        assert isinstance(identifier, bytes)
+        assert TOPIC_SEPARATOR not in identifier
+
+
 def test_loglevel():
     """
     Test LogLevel class initialization.