src/nepi/testbeds/omf/omf_client.py

   1 import logging
   2 import sleekxmpp
   3 from sleekxmpp.exceptions import IqError, IqTimeout
   4 import traceback
   5
   6 class OMFClient(sleekxmpp.ClientXMPP):
   7     def __init__(self, jid, password):
   8         sleekxmpp.ClientXMPP.__init__(self, jid, password)
   9         self._ready = False
  10         self._registered = False
  11         self._server = None
  12
  13         self.register_plugin('xep_0077') # In-band registration
  14         self.register_plugin('xep_0030')
  15         self.register_plugin('xep_0059')
  16         self.register_plugin('xep_0060') # PubSub
  17
  18         self.add_event_handler("session_start", self.start)
  19         self.add_event_handler("register", self.register)
  20         self.add_event_handler("pubsub_publish", self.handle_omf_message)
  21
  22         self._logger = logging.getLogger("nepi.testbeds.omf")
  23
  24     @property
  25     def ready(self):
  26         return self._ready
  27
  28     def start(self, event):
  29         self.send_presence()
  30         self._ready = True
  31         self._server = "pubsub.%s" % self.boundjid.domain
  32
  33     def register(self, iq):
  34         if self._registered:
  35             self._logger.info("%s already registered!" % self.boundjid)
  36             return
  37
  38         resp = self.Iq()
  39         resp['type'] = 'set'
  40         resp['register']['username'] = self.boundjid.user
  41         resp['register']['password'] = self.password
  42
  43         try:
  44             resp.send(now=True)
  45             self._logger.info("Account created for %s!" % self.boundjid)
  46             self._registered = True
  47         except IqError as e:
  48             self._logger.error("Could not register account: %s" %
  49                     e.iq['error']['text'])
  50         except IqTimeout:
  51             self._logger.error("No response from server.")
  52
  53     def unregister(self):
  54         try:
  55             self.plugin['xep_0077'].cancel_registration(
  56                 ifrom=self.boundjid.full)
  57             self._logger.info("Account unregistered for %s!" % self.boundjid)
  58         except IqError as e:
  59             self._logger.error("Could not unregister account: %s" %
  60                     e.iq['error']['text'])
  61         except IqTimeout:
  62             self._logger.error("No response from server.")
  63
  64     def nodes(self):
  65         try:
  66             result = self['xep_0060'].get_nodes(self._server)
  67             for item in result['disco_items']['items']:
  68                 self._logger.info(' - %s' % str(item))
  69             return result
  70         except:
  71             error = traceback.format_exc()
  72             self._logger.error('Could not retrieve node list.\ntraceback:\n%s', error)
  73
  74     def subscriptions(self):
  75         try:
  76             result = self['xep_0060'].get_subscriptions(self._server)
  77                 #self.boundjid.full)
  78             for node in result['node']:
  79                 self._logger.info(' - %s' % str(node))
  80             return result
  81         except:
  82             error = traceback.format_exc()
  83             self._logger.error('Could not retrieve subscriptions.\ntraceback:\n%s', error)
  84
  85     def create(self, node):
  86         config = self['xep_0004'].makeForm('submit')
  87         config.add_field(var='pubsub#node_type', value='leaf')
  88         config.add_field(var='pubsub#notify_retract', value='0')
  89         config.add_field(var='pubsub#publish_model', value='open')
  90         config.add_field(var='pubsub#persist_items', value='1')
  91         config.add_field(var='pubsub#max_items', value='1')
  92         config.add_field(var='pubsub#title', value=node)
  93
  94         try:
  95             self['xep_0060'].create_node(self._server, node, config = config)
  96         except:
  97             error = traceback.format_exc()
  98             self._logger.error('Could not create node: %s\ntraceback:\n%s' % (node, error))
  99
 100     def delete(self, node):
 101         try:
 102             self['xep_0060'].delete_node(self._server, node)
 103             self._logger.info('Deleted node: %s' % node)
 104         except:
 105             error = traceback.format_exc()
 106             self._logger.error('Could not delete node: %s\ntraceback:\n%s' % (node, error))
 107
 108     def publish(self, data, node):
 109         try:
 110             result = self['xep_0060'].publish(self._server,node,payload=data)
 111             # id = result['pubsub']['publish']['item']['id']
 112             # print('Published at item id: %s' % id)
 113         except:
 114             error = traceback.format_exc()
 115             self._logger.error('Could not publish to: %s\ntraceback:\n%s' \
 116                     % (self.boundjid, error))
 117
 118     def get(self, data):
 119         try:
 120             result = self['xep_0060'].get_item(self._server, self.boundjid,
 121                 data)
 122             for item in result['pubsub']['items']['substanzas']:
 123                 self._logger.info('Retrieved item %s: %s' % (item['id'],
 124                     tostring(item['payload'])))
 125         except:
 126             error = traceback.format_exc()
 127             self._logger.error('Could not retrieve item %s from node %s\ntraceback:\n%s' \
 128                     % (data, self.boundjid, error))
 129
 130     def retract(self, data):
 131         try:
 132             result = self['xep_0060'].retract(self._server, self.boundjid, data)
 133             self._logger.info('Retracted item %s from node %s' % (data, self.boundjid))
 134         except:
 135             error = traceback.format_exc()
 136             self._logger.error('Could not retract item %s from node %s\ntraceback:\n%s' \
 137                     % (data, self.boundjid, error))
 138
 139     def purge(self):
 140         try:
 141             result = self['xep_0060'].purge(self._server, self.boundjid)
 142             self._logger.info('Purged all items from node %s' % self.boundjid)
 143         except:
 144             error = traceback.format_exc()
 145             self._logger.error('Could not purge items from node %s\ntraceback:\n%s' \
 146                     % (self.boundjid, error))
 147
 148     def subscribe(self, node):
 149         try:
 150             result = self['xep_0060'].subscribe(self._server, node)
 151             self._logger.info('Subscribed %s to node %s' \
 152                     % (self.boundjid.bare, self.boundjid))
 153         except:
 154             error = traceback.format_exc()
 155             self._logger.error('Could not subscribe %s to node %s\ntraceback:\n%s' \
 156                     % (self.boundjid.bare, node, error))
 157
 158     def unsubscribe(self, node):
 159         try:
 160             result = self['xep_0060'].unsubscribe(self._server, node)
 161             self._logger.info('Unsubscribed %s from node %s' % (self.boundjid.bare, node))
 162         except:
 163             error = traceback.format_exc()
 164             self._logger.error('Could not unsubscribe %s from node %s\ntraceback:\n%s' \
 165                     % (self.boundjid.bare, node, error))
 166
 167     def handle_omf_message(self, iq):
 168         for i in iq['pubsub_event']['items']:
 169             self._logger.debug(i)
 170
 171             #<item xmlns="http://jabber.org/protocol/pubsub#event" id="dFbv6WRlMuKghJ0"><omf-message xmlns="http://jabber.org/protocol/pubsub"><LOGGING id="&apos;omf-payload&apos;"><LEVEL>2</LEVEL><SLICEID>default_slice</SLICEID><LOGGER>nodeHandler::NodeHandler</LOGGER><EXPID>default_slice-2012-09-28t16.22.17+02.00</EXPID><LEVEL_NAME>INFO</LEVEL_NAME><DATA>OMF Experiment Controller 5.4 (git 529a626)</DATA></LOGGING></omf-message></item>
 172
 173
 174