Send discovery message periodically

Author: jonnor

msgflo/msgflo.py

@@ -63,11 +63,14 @@ def nack(self, msg):
     self._engine.nack_message(msg)
 
 
+DEFAULT_DISCOVERY_PERIOD=60
+
 # Interface for engine/transport implementations
 class Engine(object):
-    def __init__(self, broker):
+    def __init__(self, broker, discovery_period=None):
         self.broker_url = broker
         self.broker_info = urlparse.urlparse(self.broker_url)
+        self.discovery_period = discovery_period if discovery_period else DEFAULT_DISCOVERY_PERIOD
 
     def done_callback(self, done_cb):
         self._done_cb = done_cb
@@ -91,8 +94,8 @@ def __init__(self, raw):
 
 class AmqpEngine(Engine):
 
-  def __init__(self, broker):
-    Engine.__init__(self, broker)
+  def __init__(self, broker, discovery_period=None):
+    Engine.__init__(self, broker, discovery_period=discovery_period)
 
     # Connect to AMQP broker with default connection and authentication
     # FIXME: respect self.broker_url
@@ -108,14 +111,19 @@ def add_participant(self, participant):
     self.participant = participant
     self.participant._engine = self
 
-    self._send_discovery(self._channel, self.participant.definition)
-
     # Create and configure message exchange and queue
     for p in self.participant.definition['inports']:
       self._setup_queue(self.participant, self._channel, 'in', p)
     for p in self.participant.definition['outports']:
       self._setup_queue(self.participant, self._channel, 'out', p)
 
+    def send_discovery():
+      while self.participant:
+        self._send_discovery(self._channel, self.participant.definition)
+        delay = self.discovery_period/2.2
+        gevent.sleep(delay) # yields
+    gevent.Greenlet.spawn(send_discovery)
+
   def run(self):
     # Start message pump
     self._message_pump_greenlet = gevent.spawn(self._message_pump_greenthread)
@@ -249,9 +257,7 @@ def _message_pump_greenthread(self):
 
   def _on_connect(self, client, userdata, flags, rc):
       print("Connected with result code" + str(rc))
-      self._send_discovery(self.participant.definition)
-
-
+  
       # Subscribe to queues for inports
       subscriptions = [] # ("topic", QoS)
       for port in self.participant.definition['inports']:
@@ -260,6 +266,13 @@ def _on_connect(self, client, userdata, flags, rc):
         subscriptions.append((topic, 0))
       self._client.subscribe(subscriptions)
 
+      def send_discovery():
+        while self.participant:
+          delay = self.discovery_period/2.2
+          self._send_discovery(self.participant.definition)
+          gevent.sleep(delay) # yields
+      gevent.Greenlet.spawn(send_discovery)
+
   def _on_subscribe(self, client, userdata, mid, granted_qos):
       logging.debug('subscribed %s' % str(mid))