src/nepi/resources/omf/omf_client.py

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