added python client for moleculer

Author: ToGoBananas

client_example.py

@@ -0,0 +1,20 @@
+from moleculer.client import MoleculerClient
+import threading
+import time
+
+
+def main():
+    client = MoleculerClient('amqp://guest:guest@localhost:5672/%2F?connection_attempts=3&heartbeat_interval=3600')
+    t1 = threading.Thread(target=client.run)  # run loop in separate thread
+    t1.start()
+
+    # wait for initialization
+    time.sleep(5)  # TODO: extract 'ready' event from thread
+    print(client.emit('event_test'))
+
+    time.sleep(3)
+    print(client.broadcast('event_test'))
+
+
+if __name__ == '__main__':
+    main()

moleculer/client.py

@@ -1,15 +1,105 @@
-import pika
+from .node import MoleculerNode, LOGGER
+from pika.channel import Channel
+import json
 
 
-class MoleculerClient:
+class MoleculerClient(MoleculerNode):
 
-    def __init__(self, username='guest', password='guest', host='localhost'):
-        url = 'amqp://{username}:{password}@{host}:5672/%2F?connection_attempts=3&heartbeat_interval=3600'.format(
-            username=username, password=password, host=host
-        )
-        connection = pika.BlockingConnection(pika.URLParameters(url))
-        self.channel = connection.channel()
+    def __init__(self, amqp_url):
+        super().__init__(amqp_url, node_id='PYTHON-CLIENT')
+        self.network = NetworkInfo()
 
-    def publish_event(self, node_id=None, balanced=False, event_name=None):
-        if balanced:
-            pass
+    def on_channel_open(self, channel):
+        LOGGER.info('Channel opened')
+        self.channel: Channel = channel
+        self.add_on_channel_close_callback()
+        info_queue = 'MOL.INFO.{node_id}'.format(node_id=self.NODE_ID)
+        self.setup_queue(info_queue)
+        self.channel.queue_bind(self.on_bindok, info_queue, 'MOL.INFO')
+        self.channel.basic_consume(self.process_info_packages, info_queue)
+        self.discover_packet()
+
+    def process_info_packages(self, unused_channel, basic_deliver, properties, body):
+        info_packet = json.loads(body)
+        self.network.add_node(info_packet)
+
+    def emit(self, event_name, data=None):
+        candidates = self.get_emit_candidates(event_name)
+        if len(candidates) == 0:
+            return {'error': 'This event not registered.'}
+        else:
+            if data is None:
+                data = {}
+            event_package = MoleculerClient.build_event('PYTHON-CLIENT', event_name, data)
+            for service_name in candidates:
+                queue_name = 'MOL.EVENTB.{service}.{event}'.format(service=service_name, event=event_name)
+                self.channel.basic_publish('', queue_name, event_package)
+
+    def broadcast(self, event_name, data=None):
+        candidates = self.get_broadcast_candidates(event_name)
+        if len(candidates) == 0:
+            return {'error': 'This event not registered.'}
+        else:
+            if data is None:
+                data = {}
+            event_package = MoleculerClient.build_event('PYTHON-CLIENT', event_name, data)
+            for node_id in candidates:
+                queue_name = 'MOL.EVENT.{node_id}'.format(node_id=node_id)
+                self.channel.basic_publish('', queue_name, event_package)
+
+    def get_emit_candidates(self, event_name):
+        service_names = set()
+        for node_id, node_info in self.network.NODES.items():
+            if event_name in node_info['events']:
+                service_name = node_info['service_name']
+                service_names.add(service_name)
+        return service_names
+
+    def get_broadcast_candidates(self, event_name):
+        candidates = []
+        for node_id, node_info in self.network.NODES.items():
+            if event_name in node_info['events']:
+                candidates.append(node_id)
+        return candidates
+
+    @staticmethod
+    def build_event(sender_node_id, event_name, payload):
+        event = {
+            'ver': '2',
+            'sender': sender_node_id,
+            'event': event_name,
+            'data': payload,
+            'groups': []
+        }
+        return json.dumps(event)
+
+
+class NetworkInfo:
+    NODES = {}
+
+    def add_node(self, info_packet: dict):
+        node_id = info_packet['sender']
+        if node_id not in self.NODES.keys():
+            self.NODES[node_id] = {
+                'actions': {},
+                'events': []
+            }
+            node = self.NODES[node_id]
+            for service in info_packet['services']:
+                service_name = service['name']
+                is_service_node = bool(service_name == '$node')
+                for action_name, action_spec in service['actions'].items():
+                    if is_service_node:
+                        queue_name = 'MOL.REQB.{action}'.format(action=action_name)
+                    else:
+                        queue_name = 'MOL.REQB.{service_name}.{action}'.format(service_name=service_name,
+                                                                               action=action_name)
+                    node['actions'][action_name] = queue_name
+
+                for event_name in service['events'].keys():
+                    node['events'].append(event_name)
+            else:
+                node['service_name'] = service_name
+
+    def disconnect_node(self, node_id):
+        del self.NODES[node_id]

moleculer/consumer.py

@@ -10,8 +10,18 @@ def __init__(self, node_id):
         self.node_id = node_id
         self.is_node_discovered = False
 
-    def discover(self, unused_channel, basic_deliver, properties, body):
-        pass
+    def build_info_package(self):
+        info_packet = INFO_PACKET_TEMPLATE
+        info_packet['sender'] = self.node_id
+        info_packet['services'][0]['nodeID'] = self.node_id
+        return info_packet
+
+    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)
+        info_packet = self.build_info_package()  # TODO: reuse same package
+        channel.basic_publish('', sender_queue, json.dumps(info_packet))
 
     def info(self, channel, basic_deliver, properties, body):
         if not self.is_node_discovered:
@@ -67,10 +77,12 @@ def disconnect(self, channel, basic_deliver, properties, body):
         pass
 
     def response(self, channel, basic_deliver, properties, body):
-        channel.basic_ack(basic_deliver.delivery_tag)
+        # channel.basic_ack(basic_deliver.delivery_tag)
+        pass
         #  TODO: handle responses from other services
 
     def event(self, channel, basic_deliver, properties, body):
+        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

@@ -37,7 +37,7 @@ def __init__(self, amqp_url, node_id=None):
 
         """
         self._connection = None
-        self._channel: Channel = None
+        self.channel: Channel = None
 
         self._deliveries = None
         self._acked = None
@@ -87,7 +87,7 @@ def on_connection_closed(self, connection, reply_code, reply_text):
         :param str reply_text: The server provided reply_text if given
 
         """
-        self._channel = None
+        self.channel = None
         if self._stopping:
             self._connection.ioloop.stop()
         else:
@@ -101,7 +101,7 @@ def open_channel(self):
 
     def on_channel_open(self, channel):
         LOGGER.info('Channel opened')
-        self._channel: Channel = channel
+        self.channel: Channel = channel
         # self._channel.basic_qos(prefetch_count=1)  # TODO: figure out why prefetch must be disabled to make it work
         # self._channel.confirm_delivery()  # Enabled delivery confirmations
         self.add_on_channel_close_callback()
@@ -115,25 +115,25 @@ def start_heartbeating(self):
             'sender': self.NODE_ID,
             'cpu': psutil.cpu_percent(interval=None)
         }
-        self._channel.basic_publish(MOLECULER_EXCHANGES['HEARTBEAT'], '',
-                                    json.dumps(heartbeat_packet))
+        self.channel.basic_publish(MOLECULER_EXCHANGES['HEARTBEAT'], '',
+                                   json.dumps(heartbeat_packet))
         self._connection.add_timeout(self.HEARTBEAT_INTERVAL, self.start_heartbeating)
 
     def subscribe_to_topics(self):
         if self.is_bindings_ready:
             self.add_on_cancel_callback()
             # for queue in self.moleculer_topics.queues.values():
-            self._channel.basic_consume(self.consumer.discover, self.moleculer_topics.queues['DISCOVER'])
-            self._channel.basic_consume(self.consumer.info, self.moleculer_topics.queues['INFO'])
-            self._channel.basic_consume(self.consumer.ping, self.moleculer_topics.queues['PING'])
-            self._channel.basic_consume(self.consumer.request, self.moleculer_topics.queues['REQUEST'])
-            self._channel.basic_consume(self.consumer.response, self.moleculer_topics.queues['RESPONSE'])
-            self._channel.basic_consume(self.consumer.event, self.moleculer_topics.queues['EVENT'])
+            self.channel.basic_consume(self.consumer.discover, self.moleculer_topics.queues['DISCOVER'])
+            self.channel.basic_consume(self.consumer.info, self.moleculer_topics.queues['INFO'])
+            self.channel.basic_consume(self.consumer.ping, self.moleculer_topics.queues['PING'])
+            self.channel.basic_consume(self.consumer.request, self.moleculer_topics.queues['REQUEST'])
+            self.channel.basic_consume(self.consumer.response, self.moleculer_topics.queues['RESPONSE'])
+            self.channel.basic_consume(self.consumer.event, self.moleculer_topics.queues['EVENT'], no_ack=True)
 
             for queue_name in self.moleculer_topics.action_queues:
-                self._channel.basic_consume(self.consumer.request, queue_name)
+                self.channel.basic_consume(self.consumer.request, queue_name)
             for queue_name in self.moleculer_topics.event_queues:
-                self._channel.basic_consume(self.consumer.event, queue_name, no_ack=True)
+                self.channel.basic_consume(self.consumer.event, queue_name, no_ack=True)
 
             self._connection.add_timeout(0.5, self.discover_packet)
         else:
@@ -181,22 +181,22 @@ def is_bindings_ready(self):
     def bind_queues_to_exchanges(self):
         self.expect_bindings_count = len(self.moleculer_topics.bindings)
         for queue_name, fanout_name in self.moleculer_topics.bindings.items():
-            self._channel.queue_bind(self.on_bindok, queue_name, fanout_name)
+            self.channel.queue_bind(self.on_bindok, queue_name, fanout_name)
 
     def add_on_channel_close_callback(self):
         """This method tells pika to call the on_channel_closed method if
         RabbitMQ unexpectedly closes the channel.
 
         """
         LOGGER.info('Adding channel close callback')
-        self._channel.add_on_close_callback(self.on_channel_closed)
+        self.channel.add_on_close_callback(self.on_channel_closed)
 
     def discover_packet(self):
         req = {
             'ver': '2',
             'sender': self.NODE_ID
         }
-        self._channel.basic_publish(MOLECULER_EXCHANGES['DISCOVER'], '',  json.dumps(req))
+        self.channel.basic_publish(MOLECULER_EXCHANGES['DISCOVER'], '', json.dumps(req))
 
     def on_channel_closed(self, channel, reply_code, reply_text):
         """Invoked by pika when RabbitMQ unexpectedly closes the channel.
@@ -211,7 +211,7 @@ def on_channel_closed(self, channel, reply_code, reply_text):
 
         """
         LOGGER.warning('Channel was closed: (%s) %s', reply_code, reply_text)
-        self._channel = None
+        self.channel = None
         if not self._stopping:
             self._connection.close()
 
@@ -224,9 +224,9 @@ def setup_exchange(self, exchange_name):
 
         """
         LOGGER.info('Declaring exchange %s', exchange_name)
-        self._channel.exchange_declare(self.on_exchange_declareok,
-                                       exchange_name,
-                                       self.EXCHANGE_TYPE, durable=True)
+        self.channel.exchange_declare(self.on_exchange_declareok,
+                                      exchange_name,
+                                      self.EXCHANGE_TYPE, durable=True)
 
     def on_exchange_declareok(self, unused_frame):
         """Invoked by pika when RabbitMQ has finished the Exchange.Declare RPC
@@ -252,8 +252,8 @@ def setup_queue(self, queue_name, ttl=True, exclusive=False, durable=False):
         arguments = {}
         if ttl:
             arguments['x-message-ttl'] = 5000  # eventTimeToLive: https://github.com/ice-services/moleculer/pull/72
-        self._channel.queue_declare(self.on_queue_declareok, queue_name,
-                                    exclusive=exclusive, durable=durable, arguments=arguments)
+        self.channel.queue_declare(self.on_queue_declareok, queue_name,
+                                   exclusive=exclusive, durable=durable, arguments=arguments)
 
     def on_queue_declareok(self, method_frame):
         """Method invoked by pika when the Queue.Declare RPC call made in
@@ -282,7 +282,7 @@ def add_on_cancel_callback(self):
 
         """
         LOGGER.info('Adding consumer cancellation callback')
-        self._channel.add_on_cancel_callback(self.on_consumer_cancelled)
+        self.channel.add_on_cancel_callback(self.on_consumer_cancelled)
 
     def on_consumer_cancelled(self, method_frame):
         """Invoked by pika when RabbitMQ sends a Basic.Cancel for a consumer
@@ -293,8 +293,8 @@ def on_consumer_cancelled(self, method_frame):
         """
         LOGGER.info('Consumer was cancelled remotely, shutting down: %r',
                     method_frame)
-        if self._channel:
-            self._channel.close()
+        if self.channel:
+            self.channel.close()
 
     def run(self):
         """Run the service code by connecting and then starting the IOLoop.
@@ -332,8 +332,8 @@ def stop(self):
             'ver': '2',
             'sender': self.NODE_ID
         }
-        self._channel.basic_publish(MOLECULER_EXCHANGES['DISCONNECT'], '',
-                                    json.dumps(disconnect_packet))
+        self.channel.basic_publish(MOLECULER_EXCHANGES['DISCONNECT'], '',
+                                   json.dumps(disconnect_packet))
         self._stopping = True
         self.close_channel()
         self.close_connection()
@@ -343,9 +343,9 @@ def close_channel(self):
         the Channel.Close RPC command.
 
         """
-        if self._channel is not None:
+        if self.channel is not None:
             LOGGER.info('Closing the channel')
-            self._channel.close()
+            self.channel.close()
 
     def close_connection(self):
         """This method closes the connection to RabbitMQ."""