added event packet handling +small updates to meet transport options

Author: ToGoBananas

moleculer/client.py

@@ -106,6 +106,7 @@ def open_channel(self):
     def on_channel_open(self, channel):
         LOGGER.info('Channel opened')
         self._channel: Channel = channel
+        self._channel.basic_qos(prefetch_count=1)
         # self._channel.confirm_delivery()  # Enabled delivery confirmations
         self.add_on_channel_close_callback()
         self.create_topics()
@@ -236,7 +237,7 @@ def setup_queue(self, queue_name, ttl=True):
         LOGGER.info('Declaring queue %s', queue_name)
         arguments = {}
         if ttl:
-            arguments['x-message-ttl'] = 3000
+            arguments['x-message-ttl'] = 5000  # eventTimeToLive: https://github.com/ice-services/moleculer/pull/72
         self._channel.queue_declare(self.on_queue_declareok, queue_name, arguments=arguments)
 
     def on_queue_declareok(self, method_frame):

moleculer/consumer.py

@@ -1,7 +1,7 @@
 import json
 from .topics import EXCHANGES as MOLECULER_EXCHANGES
 import datetime
-from .service import request_handler, INFO_PACKET_TEMPLATE
+from .service import request_handler, INFO_PACKET_TEMPLATE, event_handler
 
 
 class MoleculerConsumer:
@@ -17,7 +17,7 @@ def info(self, channel, basic_deliver, properties, body):
         if not self.is_node_discovered:
             info_packet = json.loads(body)
             sender = info_packet['sender']
-            if sender != self.node_id:
+            if sender != self.node_id:  # TODO: send info package anyway
                 info_packet = INFO_PACKET_TEMPLATE
                 info_packet['sender'] = self.node_id
                 info_packet['services'][0]['nodeID'] = self.node_id
@@ -63,8 +63,14 @@ def request(self, channel, basic_deliver, properties, body):
         channel.basic_publish('', sender_exchange, json.dumps(response_packet))
 
     def disconnect(self, channel, basic_deliver, properties, body):
+        # TODO: remove disconnected services DISCOVERED list
         pass
 
     def response(self, channel, basic_deliver, properties, body):
         channel.basic_ack(basic_deliver.delivery_tag)
         #  TODO: handle responses from other services
+
+    def event(self, channel, basic_deliver, properties, body):
+        event_packet = json.loads(body)
+        sender, event, data = event_packet['sender'], event_packet['event'], event_packet['data']
+        event_handler(sender, event, data)

moleculer/service.py

@@ -27,6 +27,10 @@ def request_handler(action: str, params: dict) -> bool:
     return True
 
 
+def event_handler(sender_node_id: str, event: str, payload: dict):
+    pass
+
+
 INFO_PACKET_TEMPLATE = {
     'ver': '2',
     'sender': None,