added support of namespaces

Author: ToGoBananas

moleculer/consumer.py

@@ -1,14 +1,27 @@
 import json
-from .topics import EXCHANGES as MOLECULER_EXCHANGES
 import datetime
 from .service import request_handler, INFO_PACKET_TEMPLATE, event_handler
 
 
 class MoleculerConsumer:
 
-    def __init__(self, node_id):
+    def __init__(self, node_id, moleculer_topics, namespace=None):
+        self.namespace = namespace
+        self.moleculer_topics = moleculer_topics
         self.node_id = node_id
         self.is_node_discovered = False
+        if self.namespace is None:
+            self.info_template = 'MOL.INFO.{node_id}'
+        else:
+            self.info_template = 'MOL-{namespace}.INFO.{node_id}'
+        if self.namespace is None:
+            self.res_template = 'MOL.RES.{node_id}'
+        else:
+            self.res_template = 'MOL-{namespace}.RES.{node_id}'
+        if self.namespace is None:
+            self.pong_template = 'MOL.PONG.{node_id}'
+        else:
+            self.pong_template = 'MOL-{namespace}.PONG.{node_id}'
 
     def build_info_package(self):
         info_packet = INFO_PACKET_TEMPLATE
@@ -19,7 +32,7 @@ def build_info_package(self):
     def discover(self, channel, basic_deliver, properties, body):
         discover_packet = json.loads(body)
         sender = discover_packet['sender']
-        sender_queue = 'MOL.INFO.{node_id}'.format(node_id=sender)
+        sender_queue = self.info_template.format(node_id=sender, namespace=self.namespace)
         info_packet = self.build_info_package()  # TODO: reuse same package
         channel.basic_publish('', sender_queue, json.dumps(info_packet))
 
@@ -31,7 +44,7 @@ def info(self, channel, basic_deliver, properties, body):
                 info_packet = INFO_PACKET_TEMPLATE
                 info_packet['sender'] = self.node_id
                 info_packet['services'][0]['nodeID'] = self.node_id
-                channel.basic_publish(MOLECULER_EXCHANGES['INFO'], '',  json.dumps(info_packet))
+                channel.basic_publish(self.moleculer_topics.exchanges['INFO'], '', json.dumps(info_packet))
                 self.is_node_discovered = True
         else:
             pass  # TODO: save discovered services
@@ -43,7 +56,7 @@ def ping(self, channel, basic_deliver, properties, body):
         ping_packet = json.loads(body)
         sender_node_id, time = ping_packet['sender'], ping_packet['time']
         if sender_node_id != self.node_id:
-            sender_exchange = 'MOL.PONG.{node_id}'.format(node_id=sender_node_id)
+            sender_exchange = self.pong_template.format(node_id=sender_node_id, namespace=self.namespace)
             pong_packet = {
                 'ver': '2',
                 'sender': self.node_id,
@@ -69,7 +82,7 @@ def request(self, channel, basic_deliver, properties, body):
             'success': True,
             'data': {'result': 'Response from python node: ' + self.node_id}
         }
-        sender_exchange = 'MOL.RES.{node_id}'.format(node_id=sender)
+        sender_exchange = self.res_template.format(node_id=sender, namespace=self.namespace)
         channel.basic_publish('', sender_exchange, json.dumps(response_packet))
 
     def disconnect(self, channel, basic_deliver, properties, body):
@@ -82,7 +95,7 @@ def response(self, channel, basic_deliver, properties, body):
         #  TODO: handle responses from other services
 
     def event(self, channel, basic_deliver, properties, body):
-        print('EVENT!!!!')
+        # print('EVENT!!!!')
         event_packet = json.loads(body)
         sender, event, data = event_packet['sender'], event_packet['event'], event_packet['data']
         event_handler(sender, event, data)

moleculer/node.py

@@ -5,7 +5,7 @@
 import json
 import psutil
 
-from .topics import MoleculerTopics, EXCHANGES as MOLECULER_EXCHANGES
+from .topics import MoleculerTopics
 from pika.channel import Channel
 from pika.adapters.select_connection import SelectConnection
 from .consumer import MoleculerConsumer
@@ -29,7 +29,7 @@ class MoleculerNode(object):
     NODE_ID = 'python-node-1'
     HEARTBEAT_INTERVAL = 5
 
-    def __init__(self, amqp_url, node_id=None):
+    def __init__(self, amqp_url, node_id=None, namespace=None):
         """Setup the example publisher object, passing in the URL we will use
         to connect to RabbitMQ.
 
@@ -54,8 +54,11 @@ def __init__(self, amqp_url, node_id=None):
         if node_id is not None:
             self.NODE_ID = node_id
 
-        self.moleculer_topics = MoleculerTopics(self.NODE_ID)
-        self.consumer = MoleculerConsumer(self.NODE_ID)
+        self.namespace = namespace
+
+        self.moleculer_topics = MoleculerTopics(self.NODE_ID, namespace=self.namespace)
+        self.consumer = MoleculerConsumer(self.NODE_ID, moleculer_topics=self.moleculer_topics,
+                                          namespace=self.namespace)
 
     def connect(self):
         """This method connects to RabbitMQ, returning the connection handle.
@@ -115,7 +118,7 @@ def start_heartbeating(self):
             'sender': self.NODE_ID,
             'cpu': psutil.cpu_percent(interval=None)
         }
-        self.channel.basic_publish(MOLECULER_EXCHANGES['HEARTBEAT'], '',
+        self.channel.basic_publish(self.moleculer_topics.exchanges['HEARTBEAT'], '',
                                    json.dumps(heartbeat_packet))
         self._connection.add_timeout(self.HEARTBEAT_INTERVAL, self.start_heartbeating)
 
@@ -142,7 +145,8 @@ def subscribe_to_topics(self):
     def create_topics(self):
         queues = self.moleculer_topics.queues.items()
         action_queues, events_queues = self.moleculer_topics.action_queues, self.moleculer_topics.event_queues
-        self.expect_topics_count = len(queues) + len(MOLECULER_EXCHANGES) + len(action_queues) + len(events_queues)
+        self.expect_topics_count = len(queues) + len(self.moleculer_topics.exchanges) + len(action_queues) + len(
+            events_queues)
 
         for queue_type, queue_name in queues:
             if queue_type in ('REQUEST', 'RESPONSE'):
@@ -158,7 +162,7 @@ def create_topics(self):
         for queue_name in events_queues:
             self.setup_queue(queue_name, ttl=True, exclusive=False)
 
-        for exchange_type, exchange_name in MOLECULER_EXCHANGES.items():
+        for exchange_type, exchange_name in self.moleculer_topics.exchanges.items():
             self.setup_exchange(exchange_name)
 
         self._connection.add_timeout(0.1, self.check_topics_status)
@@ -196,7 +200,7 @@ def discover_packet(self):
             'ver': '2',
             'sender': self.NODE_ID
         }
-        self.channel.basic_publish(MOLECULER_EXCHANGES['DISCOVER'], '', json.dumps(req))
+        self.channel.basic_publish(self.moleculer_topics.exchanges['DISCOVER'], '', json.dumps(req))
 
     def on_channel_closed(self, channel, reply_code, reply_text):
         """Invoked by pika when RabbitMQ unexpectedly closes the channel.
@@ -332,7 +336,7 @@ def stop(self):
             'ver': '2',
             'sender': self.NODE_ID
         }
-        self.channel.basic_publish(MOLECULER_EXCHANGES['DISCONNECT'], '',
+        self.channel.basic_publish(self.moleculer_topics.exchanges['DISCONNECT'], '',
                                    json.dumps(disconnect_packet))
         self._stopping = True
         self.close_channel()

moleculer/topics.py

@@ -9,6 +9,11 @@
     'DISCONNECT': 'MOL.DISCONNECT'
 }
 
+REQB = 'MOL.REQB.{service_name}.{action}'
+REQB_NAMESPACE = 'MOL-{namespace}.REQB.{service_name}.{action}'
+EVENTB = 'MOL.EVENTB.{service_name}.{event}'
+EVENTB_NAMESPACE = 'MOL-{namespace}.EVENTB.{service_name}.{event}'
+
 
 class MoleculerTopics:
     EVENT_QUEUE = 'MOL.EVENT.{node_id}'
@@ -32,32 +37,54 @@ def queues(self):
             result[attr.replace('_QUEUE', '')] = getattr(self, attr)
         return result
 
-    def __init__(self, node_id):
+    def __init__(self, node_id, namespace=None):
         for queue_name in self.queue_attrs:
-            setattr(self, queue_name, getattr(MoleculerTopics, queue_name).format(node_id=node_id))
+            if namespace is None:
+                setattr(self, queue_name, getattr(MoleculerTopics, queue_name).format(node_id=node_id))
+            else:
+                queue_string = getattr(MoleculerTopics, queue_name).format(node_id=node_id)
+                queue_string = queue_string.repalce('MOL', 'MOL-' + namespace)
+                setattr(self, queue_name, queue_string)
+        self.namespace = namespace
 
     @property
     def bindings(self):
         result = {}
         for queue_type, queue_name in self.queues.items():
-            if queue_type in EXCHANGES:
-                result[queue_name] = EXCHANGES[queue_type]
+            exchanges = self.exchanges
+            if queue_type in exchanges:
+                result[queue_name] = exchanges[queue_type]
         return result
 
     @property
     def action_queues(self):
+        if self.namespace is None:
+            template = REQB
+        else:
+            template = REQB_NAMESPACE
         result = []
         for service in INFO_PACKET_TEMPLATE['services']:
             service_name = service['name']
             for action in service['actions'].keys():
-                result.append('MOL.REQB.{service_name}.{action}'.format(service_name=service_name, action=action))
+                result.append(template.format(service_name=service_name, action=action, namespace=self.namespace))
         return result
 
     @property
     def event_queues(self):
+        if self.namespace is None:
+            template = EVENTB
+        else:
+            template = EVENTB_NAMESPACE
         result = []
         for service in INFO_PACKET_TEMPLATE['services']:
             service_name = service['name']
             for event in service['events'].keys():
-                result.append('MOL.EVENTB.{service_name}.{event}'.format(service_name=service_name, event=event))
+                result.append(template.format(service_name=service_name, event=event, namespace=self.namespace))
         return result
+
+    @property
+    def exchanges(self):
+        if self.namespace is None:
+            return EXCHANGES
+        else:
+            return {x: x.replace('MOL', 'MOL-' + self.namespace) for x in EXCHANGES}