Allow for multiple subscriptions in a single stream

Author: Humbedooh

CHANGELOG.md

@@ -1,3 +1,5 @@
+# 0.7.0
+- Allow for multiple subscriptions per stream
 
 # 0.6.3
 - Fixed an issue with payload delivery stalling due to client pipe timeouts

README.md

@@ -6,6 +6,7 @@
 - [Introduction](#introduction)
 - [Installing](#installing)
 - [Topics and publishing/subscribing](#topics-and-publishingsubscribing)
+  * [Subscribing to multiple topic streams](#subscribing-to-multiple-topic-streams)
 - [Pushing an event to PyPubSub](#pushing-an-event-to-pypubsub)
   * [Pushing an event via Python](#pushing-an-event-via-python)
 - [Listening for events](#listening-for-events)
@@ -52,6 +53,15 @@ The below matrix shows how subscription paths match topics:
 | fruits + apples + red | ✓ | ✓ | ✓ | ✗ | ✓ |
 | fruits + oranges | ✓ | ✗ | ✗ | ✓ | ✗ |
 
+### Subscribing to multiple topic streams
+As mentioned above, subscription topics are typically AND'ed together, with more topics narrowing the 
+event stream. Thus, if you wanted to subscribe to `bar` OR `foo`, you would need two streams.
+
+It is possible (from 0.7.0 onwards) to utilize a single stream to subscribe to multiple topic streams at once,
+using a comma as a delimiter between topic batches. For instance, to subscribe to both 
+`apples/red` events AND `oranges/ripe` events, you may subscribe to:
+
+`http://localhost:2069/apples/red,oranges/ripe`.
 
 ## Pushing an event to PyPubSub
 Event payloads requires that the IP or IP range (Ipv4 or IPv6) is listed in `pypubsub.yaml` under `payloaders` first.

pypubsub.py

@@ -33,7 +33,7 @@
 import plugins.sqs
 
 # Some consts
-PUBSUB_VERSION = '0.6.3'
+PUBSUB_VERSION = '0.7.0'
 PUBSUB_CONTENT_TYPE = 'application/vnd.pypubsub-stream'
 PUBSUB_DEFAULT_PORT = 2069
 PUBSUB_DEFAULT_IP = '0.0.0.0'
@@ -302,7 +302,10 @@ def __init__(self, server, connection, request):
         self.lock = asyncio.Lock()
 
         # Set topics subscribed to
-        self.topics = [x for x in request.path.split('/') if x]
+        self.topics = []
+        for topic_batch in request.path.split(','):
+            sub_to = [x for x in topic_batch.split('/') if x]
+            self.topics.append(sub_to)
 
         # Is the client old and expecting zero-terminators?
         self.old_school = False
@@ -393,16 +396,18 @@ async def publish(self, subscribers):
                 if not can_see:
                     continue
             # If subscribed to all the topics, tell a subscriber about this
-            if all(el in self.topics for el in sub.topics):
-                try:
-                    if sub.old_school:
-                        async with sub.lock:
-                            await asyncio.wait_for(sub.connection.write(ojs), timeout=PUBSUB_WRITE_TIMEOUT)
-                    else:
-                        async with sub.lock:
-                            await asyncio.wait_for(sub.connection.write(js), timeout=PUBSUB_WRITE_TIMEOUT)
-                except Exception:
-                    bad_subs.append(sub)
+            for topic_batch in sub.topics:
+                if all(el in self.topics for el in topic_batch):
+                    try:
+                        if sub.old_school:
+                            async with sub.lock:
+                                await asyncio.wait_for(sub.connection.write(ojs), timeout=PUBSUB_WRITE_TIMEOUT)
+                        else:
+                            async with sub.lock:
+                                await asyncio.wait_for(sub.connection.write(js), timeout=PUBSUB_WRITE_TIMEOUT)
+                    except Exception:
+                        bad_subs.append(sub)
+                    break
         return bad_subs