3 from sleekxmpp.exceptions import IqError, IqTimeout
6 class OMFClient(sleekxmpp.ClientXMPP):
7 def __init__(self, jid, password):
8 sleekxmpp.ClientXMPP.__init__(self, jid, password)
10 self._registered = False
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
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)
22 self._logger = logging.getLogger("nepi.testbeds.omf")
28 def start(self, event):
31 self._server = "pubsub.%s" % self.boundjid.domain
33 def register(self, iq):
35 self._logger.info("%s already registered!" % self.boundjid)
40 resp['register']['username'] = self.boundjid.user
41 resp['register']['password'] = self.password
45 self._logger.info("Account created for %s!" % self.boundjid)
46 self._registered = True
48 self._logger.error("Could not register account: %s" %
49 e.iq['error']['text'])
51 self._logger.error("No response from server.")
55 self.plugin['xep_0077'].cancel_registration(
56 ifrom=self.boundjid.full)
57 self._logger.info("Account unregistered for %s!" % self.boundjid)
59 self._logger.error("Could not unregister account: %s" %
60 e.iq['error']['text'])
62 self._logger.error("No response from server.")
66 result = self['xep_0060'].get_nodes(self._server)
67 for item in result['disco_items']['items']:
68 self._logger.info(' - %s' % str(item))
71 error = traceback.format_exc()
72 self._logger.error('Could not retrieve node list.\ntraceback:\n%s', error)
74 def subscriptions(self):
76 result = self['xep_0060'].get_subscriptions(self._server)
78 for node in result['node']:
79 self._logger.info(' - %s' % str(node))
82 error = traceback.format_exc()
83 self._logger.error('Could not retrieve subscriptions.\ntraceback:\n%s', error)
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)
95 self['xep_0060'].create_node(self._server, node, config = config)
97 error = traceback.format_exc()
98 self._logger.error('Could not create node: %s\ntraceback:\n%s' % (node, error))
100 def delete(self, node):
102 self['xep_0060'].delete_node(self._server, node)
103 self._logger.info('Deleted node: %s' % node)
105 error = traceback.format_exc()
106 self._logger.error('Could not delete node: %s\ntraceback:\n%s' % (node, error))
108 def publish(self, data, node):
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)
114 error = traceback.format_exc()
115 self._logger.error('Could not publish to: %s\ntraceback:\n%s' \
116 % (self.boundjid, error))
120 result = self['xep_0060'].get_item(self._server, self.boundjid,
122 for item in result['pubsub']['items']['substanzas']:
123 self._logger.info('Retrieved item %s: %s' % (item['id'],
124 tostring(item['payload'])))
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))
130 def retract(self, data):
132 result = self['xep_0060'].retract(self._server, self.boundjid, data)
133 self._logger.info('Retracted item %s from node %s' % (data, self.boundjid))
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))
141 result = self['xep_0060'].purge(self._server, self.boundjid)
142 self._logger.info('Purged all items from node %s' % self.boundjid)
144 error = traceback.format_exc()
145 self._logger.error('Could not purge items from node %s\ntraceback:\n%s' \
146 % (self.boundjid, error))
148 def subscribe(self, node):
150 result = self['xep_0060'].subscribe(self._server, node)
151 self._logger.info('Subscribed %s to node %s' \
152 % (self.boundjid.bare, self.boundjid))
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))
158 def unsubscribe(self, node):
160 result = self['xep_0060'].unsubscribe(self._server, node)
161 self._logger.info('Unsubscribed %s from node %s' % (self.boundjid.bare, node))
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))
167 def handle_omf_message(self, iq):
168 for i in iq['pubsub_event']['items']:
169 self._logger.debug(i)
171 #<item xmlns="http://jabber.org/protocol/pubsub#event" id="dFbv6WRlMuKghJ0"><omf-message xmlns="http://jabber.org/protocol/pubsub"><LOGGING id="'omf-payload'"><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>