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