plugins groupblog, xep-0060: first pass of simplification
This commit is contained in:
Goffi
parent
59d21b93d1
commit
9a3fd0c2b3
1 changed files with 133 additions and 139 deletions
|
|
@ -58,46 +58,21 @@ class XEP_0060(object):
|
|||
log.info(_(u"PubSub plugin initialization"))
|
||||
self.host = host
|
||||
self.managedNodes = []
|
||||
self.clients = {}
|
||||
self.node_cache = Sessions(timeout=60, resettable_timeout=False)
|
||||
|
||||
def getHandler(self, profile):
|
||||
self.clients[profile] = SatPubSubClient(self.host, self)
|
||||
return self.clients[profile]
|
||||
|
||||
def profileDisconnected(self, profile):
|
||||
try:
|
||||
del self.clients[profile]
|
||||
except KeyError:
|
||||
pass
|
||||
client = self.host.getClient(profile)
|
||||
client.pubsub_client = SatPubSubClient(self.host, self)
|
||||
return client.pubsub_client
|
||||
|
||||
def addManagedNode(self, node_name, callback):
|
||||
"""Add a handler for a namespace
|
||||
|
||||
@param namespace: NS of the handler (will appear in disco info)
|
||||
@param callback: method to call when the handler is found
|
||||
@param profile: profile which manage this handler"""
|
||||
self.managedNodes.append((node_name, callback))
|
||||
|
||||
def __getClientNProfile(self, profile_key, action='do pusbsub'):
|
||||
"""Return a tuple of (client, profile)
|
||||
raise error when the profile doesn't exists
|
||||
@param profile_key: as usual :)
|
||||
@param action: text of action to show in case of error"""
|
||||
profile = self.host.memory.getProfileName(profile_key)
|
||||
if not profile:
|
||||
err_mess = _('Trying to %(action)s with an unknown profile key [%(profile_key)s]') % {
|
||||
'action': action,
|
||||
'profile_key': profile_key}
|
||||
log.error(err_mess)
|
||||
raise Exception(err_mess)
|
||||
try:
|
||||
client = self.clients[profile]
|
||||
except KeyError:
|
||||
err_mess = _('INTERNAL ERROR: no handler for required profile')
|
||||
log.error(err_mess)
|
||||
raise Exception(err_mess)
|
||||
return profile, client
|
||||
|
||||
def _getDeferredNodeCache(self, session_id, init, profile):
|
||||
"""Manage a node cache with deferred initialisation and concurrent access.
|
||||
|
||||
|
|
@ -146,7 +121,7 @@ class XEP_0060(object):
|
|||
@param nodeIdentifier (str): the parent node name (leave empty to retrieve all subscriptions)
|
||||
@param filter_ (str): filter the result according to the given subscription type:
|
||||
- None: do not filter
|
||||
- 'pending': subscription has been approved yet by the node owner
|
||||
- 'pending': subscription has not been approved yet by the node owner
|
||||
- 'unconfigured': subscription options have not been configured yet
|
||||
- 'subscribed': subscription is complete
|
||||
@param profile (str): %(doc_profile)s
|
||||
|
|
@ -158,8 +133,8 @@ class XEP_0060(object):
|
|||
return self._getDeferredNodeCache(session_id, d, profile)
|
||||
|
||||
def publish(self, service, nodeIdentifier, items=None, profile_key=C.PROF_KEY_NONE):
|
||||
profile, client = self.__getClientNProfile(profile_key, 'publish item')
|
||||
return client.publish(service, nodeIdentifier, items, client.parent.jid)
|
||||
client = self.host.getClient(profile_key)
|
||||
return client.pubsub_client.publish(service, nodeIdentifier, items, client.pubsub_client.parent.jid)
|
||||
|
||||
def getItems(self, service, node, max_items=None, item_ids=None, sub_id=None, rsm=None, profile_key=C.PROF_KEY_NONE):
|
||||
"""Retrieve pubsub items from a node.
|
||||
|
|
@ -175,10 +150,10 @@ class XEP_0060(object):
|
|||
- list of items
|
||||
- RSM response data
|
||||
"""
|
||||
profile, client = self.__getClientNProfile(profile_key, 'get items')
|
||||
client = self.host.getClient(profile_key)
|
||||
ext_data = {'id': unicode(uuid.uuid4()), 'rsm': rsm} if rsm else None
|
||||
d = client.items(service, node, max_items, item_ids, sub_id, client.parent.jid, ext_data)
|
||||
d.addCallback(lambda items: (items, client.getRSMResponse(ext_data['id']) if rsm else {}))
|
||||
d = client.pubsub_client.items(service, node, max_items, item_ids, sub_id, client.pubsub_client.parent.jid, ext_data)
|
||||
d.addCallback(lambda items: (items, client.pubsub_client.getRSMResponse(ext_data['id']) if rsm else {}))
|
||||
return d
|
||||
|
||||
@defer.inlineCallbacks
|
||||
|
|
@ -197,39 +172,39 @@ class XEP_0060(object):
|
|||
- list of items
|
||||
- RSM response data
|
||||
"""
|
||||
profile, client = self.__getClientNProfile(profile_key, 'get items')
|
||||
found_nodes = yield self.listNodes(service, profile=profile)
|
||||
client = self.host.getClient(profile_key)
|
||||
found_nodes = yield self.listNodes(service, profile=client.profile)
|
||||
d_dict = {}
|
||||
for publisher, node in data.items():
|
||||
if node not in found_nodes:
|
||||
log.debug(u"Skip the items retrieval for [{node}]: node doesn't exist".format(node=node))
|
||||
continue # avoid pubsub "item-not-found" error
|
||||
d_dict[publisher] = self.getItems(service, node, max_items, None, sub_id, rsm, profile)
|
||||
d_dict[publisher] = self.getItems(service, node, max_items, None, sub_id, rsm, client.profile)
|
||||
defer.returnValue(d_dict)
|
||||
|
||||
def getOptions(self, service, nodeIdentifier, subscriber, subscriptionIdentifier=None, profile_key=C.PROF_KEY_NONE):
|
||||
profile, client = self.__getClientNProfile(profile_key, 'get options')
|
||||
return client.getOptions(service, nodeIdentifier, subscriber, subscriptionIdentifier)
|
||||
client = self.host.getClient(profile_key)
|
||||
return client.pubsub_client.getOptions(service, nodeIdentifier, subscriber, subscriptionIdentifier)
|
||||
|
||||
def setOptions(self, service, nodeIdentifier, subscriber, options, subscriptionIdentifier=None, profile_key=C.PROF_KEY_NONE):
|
||||
profile, client = self.__getClientNProfile(profile_key, 'set options')
|
||||
return client.setOptions(service, nodeIdentifier, subscriber, options, subscriptionIdentifier)
|
||||
client = self.host.getClient(profile_key)
|
||||
return client.pubsub_client.setOptions(service, nodeIdentifier, subscriber, options, subscriptionIdentifier)
|
||||
|
||||
def createNode(self, service, nodeIdentifier, options, profile_key=C.PROF_KEY_NONE):
|
||||
profile, client = self.__getClientNProfile(profile_key, 'create node')
|
||||
return client.createNode(service, nodeIdentifier, options)
|
||||
client = self.host.getClient(profile_key)
|
||||
return client.pubsub_client.createNode(service, nodeIdentifier, options)
|
||||
|
||||
def deleteNode(self, service, nodeIdentifier, profile_key=C.PROF_KEY_NONE):
|
||||
profile, client = self.__getClientNProfile(profile_key, 'delete node')
|
||||
return client.deleteNode(service, nodeIdentifier)
|
||||
client = self.host.getClient(profile_key)
|
||||
return client.pubsub_client.deleteNode(service, nodeIdentifier)
|
||||
|
||||
def retractItems(self, service, nodeIdentifier, itemIdentifiers, profile_key=C.PROF_KEY_NONE):
|
||||
profile, client = self.__getClientNProfile(profile_key, 'retract items')
|
||||
return client.retractItems(service, nodeIdentifier, itemIdentifiers)
|
||||
client = self.host.getClient(profile_key)
|
||||
return client.pubsub_client.retractItems(service, nodeIdentifier, itemIdentifiers)
|
||||
|
||||
def subscribe(self, service, nodeIdentifier, sub_jid=None, options=None, profile_key=C.PROF_KEY_NONE):
|
||||
profile, client = self.__getClientNProfile(profile_key, 'subscribe node')
|
||||
return client.subscribe(service, nodeIdentifier, sub_jid or client.parent.jid.userhostJID(), options=options)
|
||||
client = self.host.getClient(profile_key)
|
||||
return client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.pubsub_client.parent.jid.userhostJID(), options=options)
|
||||
|
||||
@defer.inlineCallbacks
|
||||
def subscribeToMany(self, service, nodeIdentifiers, sub_jid=None, options=None, profile_key=C.PROF_KEY_NONE):
|
||||
|
|
@ -242,20 +217,20 @@ class XEP_0060(object):
|
|||
@param profile_key (str): %(doc_profile_key)s
|
||||
@return: list of Deferred instances.
|
||||
"""
|
||||
profile, client = self.__getClientNProfile(profile_key, 'subscribe nodes')
|
||||
found_nodes = yield self.listNodes(service, profile=profile)
|
||||
subscribed_nodes = yield self.listSubscribedNodes(service, profile=profile)
|
||||
client = self.host.getClient(profile_key)
|
||||
found_nodes = yield self.listNodes(service, profile=client.profile)
|
||||
subscribed_nodes = yield self.listSubscribedNodes(service, profile=client.profile)
|
||||
d_list = []
|
||||
for nodeIdentifier in (set(nodeIdentifiers) - set(subscribed_nodes)):
|
||||
if nodeIdentifier not in found_nodes:
|
||||
log.debug(u"Skip the subscription to [{node}]: node doesn't exist".format(node=nodeIdentifier))
|
||||
continue # avoid sat-pubsub "SubscriptionExists" error
|
||||
d_list.append(client.subscribe(service, nodeIdentifier, sub_jid or client.parent.jid.userhostJID(), options=options))
|
||||
d_list.append(client.pubsub_client.subscribe(service, nodeIdentifier, sub_jid or client.pubsub_client.parent.jid.userhostJID(), options=options))
|
||||
defer.returnValue(d_list)
|
||||
|
||||
def subscriptions(self, service, nodeIdentifier='', profile_key=C.PROF_KEY_NONE):
|
||||
profile, client = self.__getClientNProfile(profile_key, 'retrieve subscriptions')
|
||||
return client.subscriptions(service, nodeIdentifier)
|
||||
client = self.host.getClient(profile_key)
|
||||
return client.pubsub_client.subscriptions(service, nodeIdentifier)
|
||||
|
||||
|
||||
class SatPubSubClient(rsm.PubSubClient):
|
||||
|
|
|
|||
Loading…
Reference in a new issue