Repository: opensistemas-hub/osbrain
Pull request #158 (2017-07-03 15:50:17)

Finer topic handling for SUB/SYNC_SUB
I still need to squash lots of commits and refactor a bit (maybe some parameter names and method signatures).

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

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

```diff
@@            Coverage Diff             @@
##           master     #158      +/-   ##
==========================================
- Coverage   98.39%   98.28%   -0.11%     
==========================================
  Files          25       25              
  Lines        3049     3150     +101     
  Branches      232      240       +8     
==========================================
+ Hits         3000     3096      +96     
  Misses         37       37              
- Partials       12       17       +5
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/address.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi9hZGRyZXNzLnB5) | `100% <100%> (ø)` | :arrow_up: |
| [osbrain/tests/test\_common.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi90ZXN0cy90ZXN0X2NvbW1vbi5weQ==) | `100% <100%> (ø)` | :arrow_up: |
| [osbrain/agent.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi9hZ2VudC5weQ==) | `95.76% <94.73%> (-0.12%)` | :arrow_down: |
| [osbrain/tests/test\_agent\_sync\_publications.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi90ZXN0cy90ZXN0X2FnZW50X3N5bmNfcHVibGljYXRpb25zLnB5) | `98.64% <95.38%> (-1.36%)` | :arrow_down: |
| [osbrain/tests/common.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi90ZXN0cy9jb21tb24ucHk=) | `100% <0%> (ø)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?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/158?src=pr&el=footer). Last update [0bb9de7...e12be29](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

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

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

```diff
@@            Coverage Diff             @@
##           master     #158      +/-   ##
==========================================
- Coverage   98.39%   98.28%   -0.11%     
==========================================
  Files          25       25              
  Lines        3049     3150     +101     
  Branches      232      240       +8     
==========================================
+ Hits         3000     3096      +96     
  Misses         37       37              
- Partials       12       17       +5
```


| [Impacted Files](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [osbrain/tests/test\_common.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi90ZXN0cy90ZXN0X2NvbW1vbi5weQ==) | `100% <100%> (ø)` | :arrow_up: |
| [osbrain/address.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi9hZGRyZXNzLnB5) | `100% <100%> (ø)` | :arrow_up: |
| [osbrain/agent.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi9hZ2VudC5weQ==) | `95.76% <94.73%> (-0.12%)` | :arrow_down: |
| [osbrain/tests/test\_agent\_sync\_publications.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi90ZXN0cy90ZXN0X2FnZW50X3N5bmNfcHVibGljYXRpb25zLnB5) | `98.64% <95.38%> (-1.36%)` | :arrow_down: |
| [osbrain/tests/common.py](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=tree#diff-b3NicmFpbi90ZXN0cy9jb21tb24ucHk=) | `100% <0%> (ø)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?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/158?src=pr&el=footer). Last update [0bb9de7...e12be29](https://codecov.io/gh/opensistemas-hub/osbrain/pull/158?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

Flood1993: As of now, internally, all handlers for SUB sockets are stored as a dictionary, of the form `{topic: handler, ...}`

For the special SUB case, we have to set the socket to listen to that topic, and also store the handler internally so we know what must be called when a message with that topic is received. In bb7e5a8, a bug occurs in which by overriding the handlers, we lose the first topic.

A real life example of that would be osmarkets, in which we might connect to different pairs at any point in time.

See #159
Flood1993: I'm wondering if it would be a good idea to make `_set_handler` a non private method. That way, the user could easily modify the handlers for other communication patterns other than PUB/SUB and SYNC_PUB/SYNC_SUB. What do you think @Peque ?
Flood1993: I have improved the tests a bit so as to improve their consistency and readability.
Flood1993: I pushed various commits with the following changes:
- Remove specifying the topics as bytes. You are right, it makes more sense to use strings plus the use of bytes is undocumented.
- I changed the tests to implement the new functionality you proposed, checking the tail of the messages.
- From the last point, I refactored the server agents to share the same timer for their publications, so as to try to guarantee that the tails are always ``...1, 2]``, or at least for most of the time, because it gave me problems keeping the timers separated.
- I also changed some strings from the test. I feel many of them are still meaningless, I feel like I should try to make the tests more readable for future maintenance, but I'm not sure if the proper way is with in-line comments or in the docstring of the test.

Let me know what you think @Peque so we can keep working on this PR 😄 
Flood1993: Uhh, I don't know why it's crashing on Travis (at least I'm able to reproduce the crashes in my Ubuntu VM). In Windows all tests are passing OK. I will investigate and come back when I find something.
Peque: @Flood1993 I just suggested you exactly that (it looked suspicious). :joy:
Peque: @Flood1993 Pushed one commit here to avoid setting the channel's UUID with a non-unique identifier (it kind of violates the whole purpose of a unique identifier :joy:).

Two tests fail due to the "shutdown many agents test". That was expected after the last shutdown modification. See #208.

Apart from that there are only a couple of minor comments from yesterday and this should be ready for merge. :blush:
Peque: @Flood1993 A fix for the "shutdown many agents test" is read in #209. Seems to work fine.
Flood1993: @Peque I did the changes to the documentation and also added a test for multiple unsubscriptions in the same call. I removed the option to pass the topics as a dictionary since it seems counter intuitive to me. Do you think I should leave it there?
Peque: @Flood1993 Seems `test_agent_sync_publications.py` coverage was reduced. You can correct those lambdas. Try with:

```python
addr = server.bind(..., lambda agent, message: None)
```

Or maybe even better, use an existing handler:

```python
addr = server.bind(..., handler=append_receive)
```
Peque: @Flood1993 There are also two lines not branch-covered in `agent.py`:

![screenshot from 2017-11-10 12-34-44](https://user-images.githubusercontent.com/526577/32656822-ace7bf66-c613-11e7-8d60-4beea724b23d.png)

Could you have a look at that? I would prefer to merge only code that either keeps or improves code coverage, but never decreases it (unless very special circumstances).

Patch:
diff --git a/docs/source/basic_patterns.rst b/docs/source/basic_patterns.rst
index 547b792..68c64f1 100644
--- a/docs/source/basic_patterns.rst
+++ b/docs/source/basic_patterns.rst
@@ -176,3 +176,25 @@ subscribed to topic ``a`` only and *Dave* subscribed to topic ``b`` only.
 
 Note how we can specify different handlers for different topics when
 subscribing agents.
+
+.. index:: more_on_filtering
+.. _more_on_filtering:
+
+More on filtering
+-----------------
+
+We can also easily modify the subscriptions at run-time.
+
+In the following example, *Alice* will be publishing messages using topics
+``a`` and ``b`` at the same time. Meanwhile, *Bob* will first subscribe to
+topic ``a``. After a few seconds, he will subscribe to topic ``b`` while
+unsubscribing from topic ``a``.
+
+.. literalinclude:: ../../examples/pub_sub_more_on_filter.py
+
+.. note:: Syntax regarding handlers in the ``subscribe()`` and
+   ``unsubscribe()`` methods is the same that the one used when specifying
+   the handlers on the ``connect()``/``bind()`` call of the SUB socket.
+
+.. warning:: Calls to the ``subscribe()`` method will always override the
+   previous handler for each of the specified topics, if any.
diff --git a/examples/pub_sub_more_on_filter.py b/examples/pub_sub_more_on_filter.py
new file mode 100644
index 0000000..93cfb9b
--- /dev/null
+++ b/examples/pub_sub_more_on_filter.py
@@ -0,0 +1,38 @@
+import time
+from osbrain import run_agent
+from osbrain import run_nameserver
+
+
+def log_a(agent, message):
+    agent.log_info('Log a: %s' % message)
+
+
+def log_b(agent, message):
+    agent.log_info('Log b: %s' % message)
+
+
+def send_messages(agent):
+    agent.send('main', 'Apple', topic='a')
+    agent.send('main', 'Banana', topic='b')
+
+
+if __name__ == '__main__':
+
+    # System deployment
+    ns = run_nameserver()
+    alice = run_agent('Alice')
+    bob = run_agent('Bob')
+
+    # System configuration
+    addr = alice.bind('PUB', alias='main')
+    alice.each(0.5, send_messages)
+    bob.connect(addr, alias='listener', handler={'a': log_a})
+
+    time.sleep(2)
+
+    bob.unsubscribe('listener', 'a')
+    bob.subscribe('listener', handlers={'b': log_b})
+
+    time.sleep(2)
+
+    ns.shutdown()
diff --git a/osbrain/address.py b/osbrain/address.py
index 3c33990..8f91906 100644
--- a/osbrain/address.py
+++ b/osbrain/address.py
@@ -389,7 +389,7 @@ class AgentChannel():
     receiver : str
         Second AgentAddress.
     """
-    def __init__(self, kind, receiver, sender):
+    def __init__(self, kind, receiver, sender, twin_uuid=None):
         self.kind = AgentChannelKind(kind)
         self.receiver = receiver
         self.sender = sender
@@ -398,6 +398,7 @@ class AgentChannel():
         self.serializer = \
             receiver.serializer if receiver else sender.serializer
         self.uuid = unique_identifier()
+        self.twin_uuid = twin_uuid
         # Set up pairs
         if sender:
             self.sender.channel = self
@@ -435,4 +436,5 @@ class AgentChannel():
         kind = self.kind.twin()
         sender = self.receiver.twin() if self.receiver is not None else None
         receiver = self.sender.twin() if self.sender is not None else None
-        return self.__class__(kind=kind, receiver=receiver, sender=sender)
+        return self.__class__(kind=kind, receiver=receiver, sender=sender,
+                              twin_uuid=self.uuid)
diff --git a/osbrain/agent.py b/osbrain/agent.py
index b06ba99..eb991a7 100644
--- a/osbrain/agent.py
+++ b/osbrain/agent.py
@@ -25,6 +25,7 @@ import zmq
 from . import config
 from .common import format_exception
 from .common import format_method_exception
+from .common import topic_to_bytes
 from .common import topics_to_bytes
 from .common import unbound_method
 from .common import validate_handler
@@ -556,14 +557,18 @@ class Agent():
             alias = address
         self.socket[alias] = socket
         self.socket[address] = socket
+        self.socket[socket] = socket
         self.address[alias] = address
         self.address[socket] = address
         self.address[address] = address
         if handler is not None:
             self.poller.register(socket, zmq.POLLIN)
-            self._set_handler(socket, handler)
+            if address.kind in ('SUB', 'SYNC_SUB'):
+                self.subscribe(socket, handler)
+            else:
+                self._set_handler(socket, handler)
 
-    def _set_handler(self, socket, handlers):
+    def _set_handler(self, socket, handler, update=False):
         """
         Set the socket handler(s).
 
@@ -571,17 +576,23 @@ class Agent():
         ----------
         socket : zmq.Socket
             Socket to set its handler(s).
-        handlers : function(s)
+        handler : function(s)
             Handler(s) for the socket. This can be a list or a dictionary too.
         """
-        self.handler[socket] = self._curated_handlers(handlers)
+        if update:
+            try:
+                self.handler[socket].update(self._curated_handlers(handler))
+            except KeyError:
+                self.handler[socket] = self._curated_handlers(handler)
+        else:
+            self.handler[socket] = self._curated_handlers(handler)
 
-    def _curated_handlers(self, handlers):
-        if isinstance(handlers, (list, tuple)):
-            return [self._curate_handler(h) for h in handlers]
-        if isinstance(handlers, dict):
-            return {k: self._curate_handler(v) for k, v in handlers.items()}
-        return self._curate_handler(handlers)
+    def _curated_handlers(self, handler):
+        if isinstance(handler, (list, tuple)):
+            return [self._curate_handler(h) for h in handler]
+        if isinstance(handler, dict):
+            return {k: self._curate_handler(v) for k, v in handler.items()}
+        return self._curate_handler(handler)
 
     def _curate_handler(self, handler):
         if isinstance(handler, str):
@@ -667,7 +678,7 @@ class Agent():
         self.register(socket, server_address, alias, handler)
         # SUB sockets are a special case
         if kind == 'SUB':
-            self._subscribe(server_address, handler)
+            self.subscribe(server_address, handler)
         return server_address
 
     def _bind_channel(self, kind, alias=None, handler=None, addr=None,
@@ -804,7 +815,7 @@ class Agent():
         if client_address.kind == 'SUB':
             if not alias:
                 alias = client_address
-            self._subscribe(alias, handler)
+            self.subscribe(alias, handler)
         return client_address
 
     def _connect_channel(self, channel, alias=None, handler=None):
@@ -933,29 +944,91 @@ class Agent():
         else:
             handler(self, response)
 
-    def _subscribe(self, alias: str, handlers: Dict[Union[bytes, str], Any]):
+    def subscribe(self, alias: str,
+                  handler: Dict[Union[bytes, str], Any]) -> None:
         """
-        Subscribe the agent to another agent.
+        Subscribe a SUB/SYNC_SUB socket given by its alias to the given
+        topics, and leave the handlers prepared internally.
 
         Parameters
         ----------
         alias
             Alias of the new subscriber socket.
-        handlers
+        handler
             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
             to any topic.
         """
-        if not isinstance(handlers, dict):
-            handlers = {'': handlers}
+        if not isinstance(handler, dict):
+            handler = {'': handler}
+
+        curated_handlers = topics_to_bytes(handler)
 
-        curated_handlers = topics_to_bytes(handlers)
         # Subscribe to topics
-        for topic in curated_handlers.keys():
-            self.socket[alias].setsockopt(zmq.SUBSCRIBE, topic)
+        for topic in curated_handlers:
+            self._subscribe_to_topic(alias, topic)
+
         # Reset handlers
-        self._set_handler(self.socket[alias], curated_handlers)
+        if isinstance(self.address[alias], AgentChannel):
+            channel = self.address[alias]
+            sub_address = channel.receiver
+            uuid = channel.twin_uuid
+            curated_handlers = topics_to_bytes(handler, uuid=uuid)
+            self._set_handler(self.socket[sub_address], curated_handlers,
+                              update=True)
+        else:
+            self._set_handler(self.socket[alias], curated_handlers,
+                              update=True)
+
+    def unsubscribe(self, alias: str, topic: Union[bytes, str]) -> None:
+        '''
+        Unsubscribe a SUB/SYNC_SUB socket given by its alias from a given
+        specific topic, and delete its entry from the handlers dictionary.
+
+        If instead of a single topic, a tuple or a list of topics is passed,
+        the agent will unsubscribe from all the supplied topics.
+        '''
+        if isinstance(topic, (tuple, list)):
+            for t in topic:
+                self.unsubscribe(alias, t)
+            return
+
+        topic = topic_to_bytes(topic)
+
+        if isinstance(self.address[alias], AgentAddress):
+            self.socket[alias].setsockopt(zmq.UNSUBSCRIBE, topic)
+            del self.handler[self.socket[alias]][topic]
+        elif isinstance(self.address[alias], AgentChannel):
+            channel = self.address[alias]
+            sub_address = channel.receiver
+            treated_topic = channel.twin_uuid + topic
+            self.socket[sub_address].setsockopt(zmq.UNSUBSCRIBE, treated_topic)
+            del self.handler[self.socket[sub_address]][treated_topic]
+        else:
+            raise NotImplementedError('Unsupported address type %s!' %
+                                      self.address[alias])
+
+    def _subscribe_to_topic(self, alias: str, topic: Union[bytes, str]):
+        '''
+        Do the actual ZeroMQ subscription of a socket given by its alias to
+        a specific topic. This method only makes sense to be called on
+        SUB/SYNC_SUB sockets.
+
+        Note that the handler is not set within this function.
+        '''
+        topic = topic_to_bytes(topic)
+
+        if isinstance(self.address[alias], AgentAddress):
+            self.socket[alias].setsockopt(zmq.SUBSCRIBE, topic)
+        elif isinstance(self.address[alias], AgentChannel):
+            channel = self.address[alias]
+            sub_address = channel.receiver
+            treated_topic = channel.uuid + topic
+            self.socket[sub_address].setsockopt(zmq.SUBSCRIBE, treated_topic)
+        else:
+            raise NotImplementedError('Unsupported address type %s!' %
+                                      self.address[alias])
 
     def idle(self):
         """
@@ -1335,8 +1408,7 @@ class Agent():
         if address.kind == 'PUB':
             if topic is None:
                 topic = ''
-            if isinstance(topic, str):
-                topic = topic.encode()
+            topic = topic_to_bytes(topic)
             message = compose_message(message=message,
                                       topic=topic,
                                       serializer=address.serializer)
@@ -1383,8 +1455,7 @@ class Agent():
                                     serializer=channel.serializer)
         if topic is None:
             topic = ''
-        if isinstance(topic, str):
-            topic = topic.encode()
+        topic = topic_to_bytes(topic)
         if general:
             topic = channel.uuid + topic
         message = compose_message(message=message,
@@ -1528,6 +1599,8 @@ class Agent():
         external_sockets = []
 
         for k, v in self.socket.items():
+            if isinstance(k, zmq.sugar.socket.Socket):
+                continue
             if isinstance(k, AgentAddress) and k.address in reserved:
                 continue
             if k in reserved:


Test patch:
diff --git a/osbrain/tests/test_agent_sync_publications.py b/osbrain/tests/test_agent_sync_publications.py
index c684557..6a58bea 100644
--- a/osbrain/tests/test_agent_sync_publications.py
+++ b/osbrain/tests/test_agent_sync_publications.py
@@ -11,6 +11,7 @@ from osbrain import run_logger
 from osbrain.helper import sync_agent_logger
 from osbrain.helper import logger_received
 from osbrain.helper import wait_agent_attr
+from osbrain.helper import wait_agent_condition
 
 from common import nsproxy  # pragma: no flakes
 from common import append_received
@@ -59,6 +60,13 @@ class ServerLate(Server):
         return 'x' + str(delay)
 
 
+class ServerNumbers(Agent):
+    def publish(self):
+        self.send('pub', 1, topic='negate')
+        self.send('pub', 2, topic=b'normal')
+        self.send('pub', 'not received', topic='topic_C')
+
+
 class Client(Agent):
     def on_init(self):
         self.received = []
@@ -73,6 +81,226 @@ def on_error(agent):
     agent.error_log.append('error')
 
 
+def match_tail(agent, data):
+    return agent.received[-len(data):] == data
+
+
+@pytest.mark.parametrize('socket_type', ['PUB', 'SYNC_PUB'])
+def test_unsubscribe(nsproxy, socket_type):
+    '''
+    Test that unsubscribing from topics works fine.
+
+    The server will be publishing the natural numbers, in ascending order,
+    starting at one. Odd numbers will be sent with `odd` topic and even
+    numbers with `even` topic.
+
+    A subscriber will subscribe to all numbers, and then it will unsubscribe
+    from the odd numbers.
+
+    Two non consecutive numbers must be received consecutively at that point,
+    since they both will be even.
+    '''
+    def check_non_consecutive(agent):
+        """
+        Check whether the last two received numbers are non-consecutive.
+        """
+        return agent.received[-1] - agent.received[-2] != 1
+
+    def publish(agent):
+        assert agent.count % 2 == 1
+        agent.send('pub', agent.count, topic='odd')
+        agent.send('pub', agent.count + 1, topic='even')
+        agent.count += 2
+
+    # Set up the agents
+    server = run_agent('server')
+    client = run_agent('client', base=Client)
+
+    server.set_attr(count=1)
+    addr = server.bind(socket_type, alias='pub', handler=append_received)
+    client.connect(addr, alias='sub', handler={'odd': append_received,
+                                               'even': append_received})
+
+    # Make sure the connection is established
+    server.each(0.1, 'send', 'pub', 'connecting...', topic='odd', alias='tmp')
+    assert wait_agent_attr(client, data='connecting...')
+    server.stop_timer('tmp')
+
+    # Wait until client receives some numbers
+    client.set_attr(received=[])
+    server.each(0.1, publish)
+    assert wait_agent_attr(client, length=5)
+    assert not client.execute_as_method(check_non_consecutive)
+
+    # Unsubscribe from odd numbers (receiving only even numbers)
+    client.unsubscribe('sub', 'odd')
+
+    assert wait_agent_condition(client, check_non_consecutive)
+    assert client.get_attr('received')[-1] % 2 == 0
+
+
+@pytest.mark.parametrize('socket_type', ['PUB', 'SYNC_PUB'])
+def test_unsubscribe_various(nsproxy, socket_type):
+    '''
+    Test that unsubscribing from various topics works fine.
+
+    The server will be publishing zeroes through two topics quickly, and
+    ones through a single topic slowly.
+
+    A subscriber will subscribe to all numbers, and then unsubscribe from the
+    topics related to the zeroes, therefore receiving three (or more) ones in
+    a row.
+    '''
+    def check_sum_three(agent):
+        """
+        Check whether the last three received numbers sum to three.
+
+        For this test in particular, the last three numbers should be one,
+        received through the 'one' topic.
+        """
+        return sum(agent.received[-3:]) == 3
+
+    def publish_zeroes(agent):
+        agent.send('pub', 0, topic='zero')
+        agent.send('pub', 0, topic='other_zero')
+
+    def publish_one(agent):
+        agent.send('pub', 1, topic='one')
+
+    # Set up the agents
+    server = run_agent('server')
+    client = run_agent('client', base=Client)
+
+    addr = server.bind(socket_type, alias='pub', handler=append_received)
+    client.connect(addr, alias='sub', handler={'zero': append_received,
+                                               'other_zero': append_received,
+                                               'one': append_received})
+
+    # Make sure the connection is established
+    server.each(0.1, 'send', 'pub', 'connecting...', topic='zero', alias='tmp')
+    assert wait_agent_attr(client, data='connecting...')
+    server.stop_timer('tmp')
+
+    # Wait until client receives some numbers
+    client.set_attr(received=[])
+    server.each(0.1, publish_zeroes)
+    server.each(0.2, publish_one)
+    assert wait_agent_attr(client, length=5)
+    assert not client.execute_as_method(check_sum_three)
+
+    # Unsubscribe from zeroes numbers (receiving only ones)
+    client.unsubscribe('sub', ('zero', 'other_zero'))
+
+    assert wait_agent_condition(client, check_sum_three)
+
+
+@pytest.mark.parametrize('socket_type', ['PUB', 'SYNC_PUB'])
+@pytest.mark.parametrize('subscribe_separately', [True, False])
+def test_subscribe(nsproxy, socket_type, subscribe_separately):
+    '''
+    Test subscribing to various topics/handlers works fine.
+    '''
+    server = run_agent('server', base=ServerNumbers)
+    client = run_agent('client')
+
+    addr = server.bind(socket_type, alias='pub', handler=append_received)
+    client.set_attr(received=[])
+    client.connect(addr, alias='sub', handler=append_received)
+
+    # Establish a connection
+    server.each(0.1, 'send', 'pub', 'connecting...', alias='tmp')
+    assert wait_agent_attr(client, data='connecting...')
+    server.stop_timer('tmp')
+
+    # Subscribe to two topics. Client should only receive -1 and 2.
+    client.unsubscribe('sub', '')
+    if subscribe_separately:
+        client.subscribe('sub', handler={'negate': receive_negate})
+        client.subscribe('sub', handler={'normal': append_received})
+    else:
+        client.subscribe('sub', handler={'negate': receive_negate,
+                                         'normal': append_received})
+
+    server.each(0.1, 'publish')
+
+    assert wait_agent_condition(client, match_tail, [-1, 2] * 4)
+
+
+@pytest.mark.parametrize('socket_type', ['PUB', 'SYNC_PUB'])
+def test_resubscribe(nsproxy, socket_type):
+    '''
+    After subscribing to the same topic with a different handler after a
+    subscription to that topic was already made, we should override the
+    original handler.
+
+    Note that we will only override explicitly given topics (previous
+    subscriptions of other topics will remain untouched).
+    '''
+    server = run_agent('server', base=ServerNumbers)
+    client = run_agent('client')
+
+    addr = server.bind(socket_type, alias='pub', handler=append_received)
+    client.set_attr(received=[])
+    client.connect(addr, alias='sub', handler=append_received)
+
+    # Establish a connection
+    server.each(0.1, 'send', 'pub', 'connecting...', alias='tmp')
+    assert wait_agent_attr(client, data='connecting...')
+    server.stop_timer('tmp')
+
+    # Subscribe to two topics
+    client.unsubscribe('sub', '')
+    client.subscribe('sub', handler={'negate': receive_negate,
+                                     'normal': append_received})
+
+    server.each(0.1, 'publish')
+
+    assert wait_agent_condition(client, match_tail, [-1, 2] * 4)
+
+    # Resubscribe, so as to store the 'normal' (2) messages negated
+    client.subscribe('sub', handler={'normal': receive_negate})
+
+    assert wait_agent_condition(client, match_tail, [-1, -2] * 4)
+
+
+@pytest.mark.parametrize('socket_type', ['PUB', 'SYNC_PUB'])
+def test_unsubscribe_and_subscribe_again(nsproxy, socket_type):
+    '''
+    Test resubscribing to topics after unsubscribing works fine.
+    '''
+    server = run_agent('server', base=ServerNumbers)
+    client = run_agent('client')
+
+    addr = server.bind(socket_type, alias='pub', handler=append_received)
+    client.set_attr(received=[])
+    client.connect(addr, alias='sub', handler=append_received)
+
+    # Establish a connection
+    server.each(0.1, 'send', 'pub', 'connecting...', alias='tmp')
+    assert wait_agent_attr(client, data='connecting...')
+    server.stop_timer('tmp')
+
+    # Subscribe to two topics
+    client.unsubscribe('sub', '')
+    client.subscribe('sub', handler={'negate': receive_negate,
+                                     'normal': append_received})
+
+    # Make server to start publishing messages
+    server.each(0.1, 'publish')
+
+    assert wait_agent_condition(client, match_tail, [-1, 2] * 4)
+
+    # Unsubscribe from one topic
+    client.unsubscribe('sub', 'negate')
+
+    assert wait_agent_condition(client, match_tail, [2] * 8)
+
+    # Subscribe again
+    client.subscribe('sub', handler={'negate': receive_negate})
+
+    assert wait_agent_condition(client, match_tail, [-1, 2] * 4)
+
+
 @pytest.mark.parametrize('server', [Server, PubServer])
 def test_simple_pub_single_handler(nsproxy, server):
     """