src/nepi/resources/omf/omf_client.py

   1 import logging
   2 import sleekxmpp
   3 from sleekxmpp.exceptions import IqError, IqTimeout
   4 import traceback
   5 import xml.etree.ElementTree as ET
   6
   7 import neco
   8
   9 # inherit from BaseXmpp and XMLStream classes
  10 class OMFClient(sleekxmpp.ClientXMPP):
  11     """
  12     .. class:: Class Args :
  13
  14         :param jid: Jabber Id (= Xmpp Slice + Date)
  15         :type jid: Str
  16         :param password: Jabber Password (= Xmpp Password)
  17         :type password: Str
  18
  19     .. note::
  20
  21        This class is an XMPP Client with customized method
  22
  23     """
  24
  25     def __init__(self, jid, password):
  26         """
  27
  28         :param jid: Jabber Id (= Xmpp Slice + Date)
  29         :type jid: Str
  30         :param password: Jabber Password (= Xmpp Password)
  31         :type password: Str
  32
  33
  34         """
  35         sleekxmpp.ClientXMPP.__init__(self, jid, password)
  36         self._ready = False
  37         self._registered = False
  38         self._server = None
  39
  40         self.register_plugin('xep_0077') # In-band registration
  41         self.register_plugin('xep_0030')
  42         self.register_plugin('xep_0059')
  43         self.register_plugin('xep_0060') # PubSub
  44
  45         self.add_event_handler("session_start", self.start)
  46         self.add_event_handler("register", self.register)
  47         self.add_event_handler("pubsub_publish", self.handle_omf_message)
  48
  49         self._logger = logging.getLogger("neco.omf.xmppClient")
  50         self._logger.setLevel(neco.LOGLEVEL)
  51
  52     @property
  53     def ready(self):
  54         """ Check if the client is ready
  55
  56         """
  57         return self._ready
  58
  59     def start(self, event):
  60         """ Send presence to the Xmppp Server. This function is called directly by the sleekXmpp library
  61
  62         """
  63         self.send_presence()
  64         self._ready = True
  65         self._server = "pubsub.%s" % self.boundjid.domain
  66
  67     def register(self, iq):
  68         """  Register to the Xmppp Server. This function is called directly by the sleekXmpp library
  69
  70         """
  71         if self._registered:
  72             self._logger.info(" %s already registered!" % self.boundjid)
  73             return
  74
  75         resp = self.Iq()
  76         resp['type'] = 'set'
  77         resp['register']['username'] = self.boundjid.user
  78         resp['register']['password'] = self.password
  79
  80         try:
  81             resp.send(now=True)
  82             self._logger.info(" Account created for %s!" % self.boundjid)
  83             self._registered = True
  84         except IqError as e:
  85             self._logger.error(" Could not register account: %s" %
  86                     e.iq['error']['text'])
  87         except IqTimeout:
  88             self._logger.error(" No response from server.")
  89
  90     def unregister(self):
  91         """  Unregister from the Xmppp Server.
  92
  93         """
  94         try:
  95             self.plugin['xep_0077'].cancel_registration(
  96                 ifrom=self.boundjid.full)
  97             self._logger.info(" Account unregistered for %s!" % self.boundjid)
  98         except IqError as e:
  99             self._logger.error(" Could not unregister account: %s" %
 100                     e.iq['error']['text'])
 101         except IqTimeout:
 102             self._logger.error(" No response from server.")
 103
 104     def nodes(self):
 105         """  Get all the nodes of the Xmppp Server.
 106
 107         """
 108         try:
 109             result = self['xep_0060'].get_nodes(self._server)
 110             for item in result['disco_items']['items']:
 111                 self._logger.info(' - %s' % str(item))
 112             return result
 113         except:
 114             error = traceback.format_exc()
 115             self._logger.error(' Could not retrieve node list.\ntraceback:\n%s', error)
 116
 117     def subscriptions(self):
 118         """  Get all the subscriptions of the Xmppp Server.
 119
 120         """
 121         try:
 122             result = self['xep_0060'].get_subscriptions(self._server)
 123                 #self.boundjid.full)
 124             for node in result['node']:
 125                 self._logger.info(' - %s' % str(node))
 126             return result
 127         except:
 128             error = traceback.format_exc()
 129             self._logger.error(' Could not retrieve subscriptions.\ntraceback:\n%s', error)
 130
 131     def create(self, node):
 132         """  Create the topic corresponding to the node
 133
 134         :param node: Name of the topic, corresponding to the node (ex : omf.plexus.wlab17)
 135         :type node: str
 136
 137         """
 138         self._logger.debug(" Create Topic : " + node)
 139
 140         config = self['xep_0004'].makeForm('submit')
 141         config.add_field(var='pubsub#node_type', value='leaf')
 142         config.add_field(var='pubsub#notify_retract', value='0')
 143         config.add_field(var='pubsub#publish_model', value='open')
 144         config.add_field(var='pubsub#persist_items', value='1')
 145         config.add_field(var='pubsub#max_items', value='1')
 146         config.add_field(var='pubsub#title', value=node)
 147
 148         try:
 149             self['xep_0060'].create_node(self._server, node, config = config)
 150         except:
 151             error = traceback.format_exc()
 152             self._logger.error(' Could not create topic: %s\ntraceback:\n%s' % (node, error))
 153
 154     def delete(self, node):
 155         """  Delete the topic corresponding to the node
 156
 157         :param node: Name of the topic, corresponding to the node (ex : omf.plexus.wlab17)
 158         :type node: str
 159
 160         """
 161         # To check if the queue are well empty at the end
 162         #print " length of the queue : " + str(self.send_queue.qsize())
 163         #print " length of the queue : " + str(self.event_queue.qsize())
 164         try:
 165             self['xep_0060'].delete_node(self._server, node)
 166             self._logger.info(' Deleted node: %s' % node)
 167         except:
 168             error = traceback.format_exc()
 169             self._logger.error(' Could not delete topic: %s\ntraceback:\n%s' % (node, error))
 170
 171     def publish(self, data, node):
 172         """  Publish the data to the corresponding topic
 173
 174         :param data: Data that will be published
 175         :type data: str
 176         :param node: Name of the topic
 177         :type node: str
 178
 179         """
 180
 181         self._logger.debug(" Publish to Topic : " + node)
 182         try:
 183             result = self['xep_0060'].publish(self._server,node,payload=data)
 184             # id = result['pubsub']['publish']['item']['id']
 185             # print('Published at item id: %s' % id)
 186         except:
 187             error = traceback.format_exc()
 188             self._logger.error(' Could not publish to: %s\ntraceback:\n%s' \
 189                     % (node, error))
 190
 191     def get(self, data):
 192         """  Get the item
 193
 194         :param data: data from which the items will be get back
 195         :type data: str
 196
 197
 198         """
 199         try:
 200             result = self['xep_0060'].get_item(self._server, self.boundjid,
 201                 data)
 202             for item in result['pubsub']['items']['substanzas']:
 203                 self._logger.info('Retrieved item %s: %s' % (item['id'],
 204                     tostring(item['payload'])))
 205         except:
 206             error = traceback.format_exc()
 207             self._logger.error(' Could not retrieve item %s from topic %s\ntraceback:\n%s' \
 208                     % (data, self.boundjid, error))
 209
 210     def retract(self, data):
 211         """  Retract the item
 212
 213         :param data: data from which the item will be retracted
 214         :type data: str
 215
 216         """
 217         try:
 218             result = self['xep_0060'].retract(self._server, self.boundjid, data)
 219             self._logger.info(' Retracted item %s from topic %s' % (data, self.boundjid))
 220         except:
 221             error = traceback.format_exc()
 222             self._logger.error(' Could not retract item %s from topic %s\ntraceback:\n%s' \
 223                     % (data, self.boundjid, error))
 224
 225     def purge(self):
 226         """  Purge the information in the server
 227
 228         """
 229         try:
 230             result = self['xep_0060'].purge(self._server, self.boundjid)
 231             self._logger.info(' Purged all items from topic %s' % self.boundjid)
 232         except:
 233             error = traceback.format_exc()
 234             self._logger.error(' Could not purge items from topic %s\ntraceback:\n%s' \
 235                     % (self.boundjid, error))
 236
 237     def subscribe(self, node):
 238         """ Subscribe to a topic
 239
 240         :param node: Name of the topic
 241         :type node: str
 242
 243         """
 244         try:
 245             result = self['xep_0060'].subscribe(self._server, node)
 246             #self._logger.debug('Subscribed %s to node %s' \
 247                     #% (self.boundjid.bare, node))
 248             self._logger.info(' Subscribed %s to topic %s' \
 249                     % (self.boundjid.user, node))
 250         except:
 251             error = traceback.format_exc()
 252             self._logger.error(' Could not subscribe %s to topic %s\ntraceback:\n%s' \
 253                     % (self.boundjid.bare, node, error))
 254
 255     def unsubscribe(self, node):
 256         """ Unsubscribe to a topic
 257
 258         :param node: Name of the topic
 259         :type node: str
 260
 261         """
 262         try:
 263             result = self['xep_0060'].unsubscribe(self._server, node)
 264             self._logger.info(' Unsubscribed %s from topic %s' % (self.boundjid.bare, node))
 265         except:
 266             error = traceback.format_exc()
 267             self._logger.error(' Could not unsubscribe %s from topic %s\ntraceback:\n%s' \
 268                     % (self.boundjid.bare, node, error))
 269
 270     def _check_for_tag(self, root, namespaces, tag):
 271         """  Check if an element markup is in the ElementTree
 272
 273         :param root: Root of the tree
 274         :type root: ElementTree Element
 275         :param namespaces: Namespaces of the element
 276         :type namespaces: str
 277         :param tag: Tag that will search in the tree
 278         :type tag: str
 279
 280         """
 281         for element in root.iter(namespaces+tag):
 282             if element.text:
 283                 return element
 284             else :
 285                 return None
 286
 287     def _check_output(self, root, namespaces):
 288         """ Check the significative element in the answer and display it
 289
 290         :param root: Root of the tree
 291         :type root: ElementTree Element
 292         :param namespaces: Namespaces of the tree
 293         :type namespaces: str
 294
 295         """
 296         fields = ["TARGET", "REASON", "PATH", "APPID", "VALUE"]
 297         response = ""
 298         for elt in fields:
 299             msg = self._check_for_tag(root, namespaces, elt)
 300             if msg is not None:
 301                 response = response + " " + msg.text + " :"
 302         deb = self._check_for_tag(root, namespaces, "MESSAGE")
 303         if deb is not None:
 304             self._logger.debug(response + " " + deb.text)
 305         else :
 306             self._logger.info(response)
 307
 308     def handle_omf_message(self, iq):
 309         """ Handle published/received message
 310
 311         :param iq: Stanzas that is currently published/received
 312         :type iq: Iq Stanza
 313
 314         """
 315         namespaces = "{http://jabber.org/protocol/pubsub}"
 316         for i in iq['pubsub_event']['items']:
 317             root = ET.fromstring(str(i))
 318             self._check_output(root, namespaces)
 319
 320