tools/pharos-dashboard/notification_framework/notification.py

   1 import json
   2 import re
   3
   4 import pika
   5
   6
   7 class Notification(object):
   8     """
   9     This class can be used by the dashboard and the labs to exchange notifications about booking
  10     events and pod status. It utilizes rabbitmq to communicate.
  11
  12     Notifications are associated to an event and to a topic.
  13     Events are:
  14     [ 'booking_start', 'booking_stop', 'pod_status' ]
  15     The topic is usually a POD name, ie:
  16     'Intel POD 2'
  17     """
  18
  19     def __init__(self, dashboard_url, verbose=False):
  20         self.rabbitmq_broker = dashboard_url
  21         self.verbose = verbose
  22         self._registry = {}
  23
  24         self.connection = pika.BlockingConnection(pika.ConnectionParameters(
  25             host=self.rabbitmq_broker))
  26         self.channel = self.connection.channel()
  27
  28         self.channel.exchange_declare(exchange='notifications', type='topic')
  29
  30         self.result = self.channel.queue_declare(exclusive=True)
  31         self.queue_name = self.result.method.queue
  32
  33     def register(self, function, event, regex):
  34         """
  35         Registers a function to be called for the specified event.
  36         :param function: the function to register
  37         :param event: the event type
  38         :param regex: a regex to specify for wich topics the function will be called. Some
  39         possible Expressions can be:
  40         'Intel POD 2' : Intel POD 2
  41         'Intel POD .*' : All Intel Pods
  42         '.*' : All Topics
  43         """
  44
  45         if event not in self._registry:
  46             self._registry[event] = [(function, regex)]
  47         else:
  48             self._registry[event].append((function, regex))
  49
  50     def receive(self):
  51         """
  52         Start receiving notifications. This is a blocking operation, if a notification is received,
  53         the registered functions will be called.
  54         """
  55         if self.verbose:
  56             print('Start receiving Notifications. Keys: ', self._registry.keys())
  57         self._receive_message(self._registry.keys())
  58
  59     def send(self, event, topic, content):
  60         """
  61         Send an event notification.
  62         :param event: the event type
  63         :param topic: the pod name
  64         :param content: a JSON-serializable dictionary
  65         """
  66         message = {
  67             'event': event,
  68             'topic': topic,
  69             'content': content
  70         }
  71         self._send_message(message)
  72
  73     def _send_message(self, event):
  74         routing_key = event['type']
  75         message = json.dumps(event)
  76         self.channel.basic_publish(exchange='notifications',
  77                                    routing_key=routing_key,
  78                                    body=message,
  79                                    properties=pika.BasicProperties(
  80                                        content_type='application/json'
  81                                    ))
  82         if self.verbose:
  83             print(" [x] Sent %r:%r" % (routing_key, message))
  84
  85     def _receive_message(self, binding_keys):
  86         for key in binding_keys:
  87             self.channel.queue_bind(exchange='notifications',
  88                                     queue=self.queue_name,
  89                                     routing_key=key)
  90         self.channel.basic_consume(self._message_callback,
  91                                    queue=self.queue_name,
  92                                    no_ack=True)
  93         self.channel.start_consuming()
  94
  95     def _message_callback(self, ch, method, properties, body):
  96         if self.verbose:
  97             print(" [x] Got %r:%r" % (method.routing_key, body))
  98         if method.routing_key not in self._registry:
  99             return
 100         for func, regex in self._registry[method.routing_key]:
 101             message = json.loads(body.decode())
 102             match = re.match(regex, message['topic'])
 103             if match:
 104                 func(body)