1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
"""
Slixmpp: The Slick XMPP Library
Copyright (C) 2011 Nathanael C. Fritz, Lance J.T. Stout
This file is part of Slixmpp.
See the file LICENSE for copying permission.
"""
import logging
from typing import Optional, Callable
from slixmpp import asyncio, JID
from slixmpp.xmlstream import register_stanza_plugin, ElementBase
from slixmpp.plugins.base import BasePlugin, register_plugin
from slixmpp.plugins.xep_0004.stanza import Form
log = logging.getLogger(__name__)
class XEP_0163(BasePlugin):
"""
XEP-0163: Personal Eventing Protocol (PEP)
"""
name = 'xep_0163'
description = 'XEP-0163: Personal Eventing Protocol (PEP)'
dependencies = {'xep_0030', 'xep_0060', 'xep_0115'}
def register_pep(self, name, stanza):
"""
Setup and configure events and stanza registration for
the given PEP stanza:
- Add disco feature for the PEP content.
- Register disco interest in the PEP content.
- Map events from the PEP content's namespace to the given name.
:param str name: The event name prefix to use for PEP events.
:param stanza: The stanza class for the PEP content.
"""
pubsub_stanza = self.xmpp['xep_0060'].stanza
register_stanza_plugin(pubsub_stanza.EventItem, stanza)
self.add_interest(stanza.namespace)
self.xmpp['xep_0030'].add_feature(stanza.namespace)
self.xmpp['xep_0060'].map_node_event(stanza.namespace, name)
def add_interest(self, namespace: str, jid: Optional[JID] = None):
"""
Mark an interest in a PEP subscription by including a disco
feature with the '+notify' extension.
:param namespace: The base namespace to register as an interest, such
as 'http://jabber.org/protocol/tune'. This may also
be a list of such namespaces.
:param jid: Optionally specify the JID.
"""
if not isinstance(namespace, set) and not isinstance(namespace, list):
namespace = [namespace]
for ns in namespace:
self.xmpp['xep_0030'].add_feature('%s+notify' % ns,
jid=jid)
asyncio.ensure_future(
self.xmpp['xep_0115'].update_caps(jid),
loop=self.xmpp.loop,
)
def remove_interest(self, namespace: str, jid: Optional[JID] = None):
"""
Mark an interest in a PEP subscription by including a disco
feature with the '+notify' extension.
:param namespace: The base namespace to remove as an interest, such
as 'http://jabber.org/protocol/tune'. This may also
be a list of such namespaces.
:param jid: Optionally specify the JID.
"""
if not isinstance(namespace, (set, list)):
namespace = [namespace]
for ns in namespace:
self.xmpp['xep_0030'].del_feature(jid=jid,
feature='%s+notify' % namespace)
asyncio.ensure_future(
self.xmpp['xep_0115'].update_caps(jid),
loop=self.xmpp.loop,
)
def publish(self, stanza: ElementBase, node: Optional[str] = None,
id: Optional[str] = None,
options: Optional[Form] = None,
ifrom: Optional[JID] = None,
timeout_callback: Optional[Callable] = None,
callback: Optional[Callable] = None,
timeout: Optional[int] = None):
"""
Publish a PEP update.
This is just a (very) thin wrapper around the XEP-0060 publish()
method to set the defaults expected by PEP.
:param stanza: The PEP update stanza to publish.
:param node: The node to publish the item to. If not specified,
the stanza's namespace will be used.
:param id: Optionally specify the ID of the item.
:param options: A form of publish options.
"""
if node is None:
node = stanza.namespace
if id is None:
id = 'current'
return self.xmpp['xep_0060'].publish(ifrom, node, id=id,
payload=stanza.xml,
options=options, ifrom=ifrom,
callback=callback,
timeout=timeout,
timeout_callback=timeout_callback)
register_plugin(XEP_0163)
|
