Mercurial > libervia-backend

comparison sat/plugins/plugin_xep_0060.py @ 3589:ed037c98df37

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
plugin XEP-0060: add trigger points
author Goffi <goffi@goffi.org>
date Thu, 29 Jul 2021 17:26:59 +0200
parents 2c7a52a62be3
children 1269c2d788be
comparison
equal deleted inserted replaced
3588:2c7a52a62be3 3589:ed037c98df37
584 return [item['id'] 584 return [item['id']
585 for item in iq_result.pubsub.publish.elements(pubsub.NS_PUBSUB, 'item')] 585 for item in iq_result.pubsub.publish.elements(pubsub.NS_PUBSUB, 'item')]
586 except AttributeError: 586 except AttributeError:
587 return [] 587 return []
588 588
589 async def publish(self, client, service, nodeIdentifier, items=None, options=None): 589 async def publish(
590 return await client.pubsub_client.publish( 590 self,
591 client: SatXMPPEntity,
592 service: jid.JID,
593 nodeIdentifier: str,
594 items: Optional[List[domish.Element]] = None,
595 options: Optional[dict] = None
596 ) -> List[str]:
597 published_ids = await client.pubsub_client.publish(
591 service, nodeIdentifier, items, client.pubsub_client.parent.jid, 598 service, nodeIdentifier, items, client.pubsub_client.parent.jid,
592 options=options 599 options=options
593 ) 600 )
601
602 await self.host.trigger.asyncPoint(
603 "XEP-0060_publish", client, service, nodeIdentifier, items, options,
604 published_ids
605 )
606 return published_ids
594 607
595 def _unwrapMAMMessage(self, message_elt): 608 def _unwrapMAMMessage(self, message_elt):
596 try: 609 try:
597 item_elt = reduce( 610 item_elt = reduce(
598 lambda elt, ns_name: next(elt.elements(*ns_name)), 611 lambda elt, ns_name: next(elt.elements(*ns_name)),
668 max_items = None 681 max_items = None
669 if rsm_request and item_ids: 682 if rsm_request and item_ids:
670 raise ValueError("items_id can't be used with rsm") 683 raise ValueError("items_id can't be used with rsm")
671 if extra is None: 684 if extra is None:
672 extra = {} 685 extra = {}
686 cont, ret = await self.host.trigger.asyncReturnPoint(
687 "XEP-0060_getItems", client, service, node, max_items, item_ids, sub_id,
688 rsm_request, extra
689 )
690 if not cont:
691 return ret
673 try: 692 try:
674 mam_query = extra["mam"] 693 mam_query = extra["mam"]
675 except KeyError: 694 except KeyError:
676 d = client.pubsub_client.items( 695 d = client.pubsub_client.items(
677 service = service, 696 service = service,
1095 self.subscribe(client, service, nodeIdentifier, options=options or None) 1114 self.subscribe(client, service, nodeIdentifier, options=options or None)
1096 ) 1115 )
1097 d.addCallback(lambda subscription: subscription.subscriptionIdentifier or "") 1116 d.addCallback(lambda subscription: subscription.subscriptionIdentifier or "")
1098 return d 1117 return d
1099 1118
1100 async def subscribe(self, client, service, nodeIdentifier, sub_jid=None, options=None): 1119 async def subscribe(
1120 self,
1121 client: SatXMPPEntity,
1122 service: jid.JID,
1123 nodeIdentifier: str,
1124 sub_jid: Optional[jid.JID] = None,
1125 options: Optional[dict] = None
1126 ) -> pubsub.Subscription:
1101 # TODO: reimplement a subscribtion cache, checking that we have not subscription before trying to subscribe 1127 # TODO: reimplement a subscribtion cache, checking that we have not subscription before trying to subscribe
1102 return await client.pubsub_client.subscribe( 1128 subscription = await client.pubsub_client.subscribe(
1103 service, nodeIdentifier, sub_jid or client.jid.userhostJID(), options=options 1129 service, nodeIdentifier, sub_jid or client.jid.userhostJID(), options=options
1104 ) 1130 )
1131 await self.host.trigger.asyncPoint(
1132 "XEP-0060_subscribe", client, service, nodeIdentifier, sub_jid, options,
1133 subscription
1134 )
1135 return subscription
1105 1136
1106 def _unsubscribe(self, service, nodeIdentifier, profile_key=C.PROF_KEY_NONE): 1137 def _unsubscribe(self, service, nodeIdentifier, profile_key=C.PROF_KEY_NONE):
1107 client = self.host.getClient(profile_key) 1138 client = self.host.getClient(profile_key)
1108 service = None if not service else jid.JID(service) 1139 service = None if not service else jid.JID(service)
1109 return defer.ensureDeferred(self.unsubscribe(client, service, nodeIdentifier)) 1140 return defer.ensureDeferred(self.unsubscribe(client, service, nodeIdentifier))
1110 1141
1111 async def unsubscribe( 1142 async def unsubscribe(
1112 self, 1143 self,
1113 client, 1144 client: SatXMPPEntity,
1114 service, 1145 service: jid.JID,
1115 nodeIdentifier, 1146 nodeIdentifier: str,
1116 sub_jid=None, 1147 sub_jid=None,
1117 subscriptionIdentifier=None, 1148 subscriptionIdentifier=None,
1118 sender=None, 1149 sender=None,
1119 ): 1150 ):
1120 return await client.pubsub_client.unsubscribe( 1151 await client.pubsub_client.unsubscribe(
1121 service, 1152 service,
1122 nodeIdentifier, 1153 nodeIdentifier,
1123 sub_jid or client.jid.userhostJID(), 1154 sub_jid or client.jid.userhostJID(),
1124 subscriptionIdentifier, 1155 subscriptionIdentifier,
1125 sender, 1156 sender,
1157 )
1158 await self.host.trigger.asyncPoint(
1159 "XEP-0060_unsubscribe", client, service, nodeIdentifier, sub_jid,
1160 subscriptionIdentifier, sender
1126 ) 1161 )
1127 1162
1128 def _subscriptions(self, service, nodeIdentifier="", profile_key=C.PROF_KEY_NONE): 1163 def _subscriptions(self, service, nodeIdentifier="", profile_key=C.PROF_KEY_NONE):
1129 client = self.host.getClient(profile_key) 1164 client = self.host.getClient(profile_key)
1130 service = None if not service else jid.JID(service) 1165 service = None if not service else jid.JID(service)