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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
########################################################################
# File name: utils.py
# This file is part of: aioxmpp
#
# LICENSE
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program. If not, see
# <http://www.gnu.org/licenses/>.
#
########################################################################
from abc import ABCMeta, abstractproperty
from . import xso as chatstates_xso
class ChatStateStrategy(metaclass=ABCMeta):
@abstractproperty
def sending(self):
"""
Return whether to send chat state notifications.
"""
raise NotImplementedError # pragma: no cover
def reset(self):
"""
Reset the strategy (called after a reconnect).
"""
pass
def no_reply(self):
"""
Called when the replies did not include a chat state.
"""
pass
class DoNotEmit(ChatStateStrategy):
"""
Chat state strategy: Do not emit chat state notifications.
"""
@property
def sending(self):
return False
class DiscoverSupport(ChatStateStrategy):
"""
Chat state strategy: Discover support for chat state notifications
as per section 5.1 of :xep:`0085`.
"""
def __init__(self):
self.state = True
def reset(self):
self.state = True
def no_reply(self):
self.state = False
@property
def sending(self):
return self.state
class AlwaysEmit(ChatStateStrategy):
"""
Chat state strategy: Always emit chat state notifications.
"""
@property
def sending(self):
return True
class ChatStateManager:
"""
Manage the state of our chat state.
:param strategy: the strategy used to decide whether to send
notifications (defaults to :class:`DiscoverSupport`)
:type strategy: a subclass of :class:`ChatStateStrategy`
.. automethod:: handle
Methods to pass in protocol level information:
.. automethod:: no_reply
.. automethod:: reset
"""
def __init__(self, strategy=None):
self._state = chatstates_xso.ChatState.ACTIVE
if strategy is None:
strategy = DiscoverSupport()
self._strategy = strategy
def handle(self, state, message=False):
"""
Handle a state update.
:param state: the new chat state
:type state: :class:`~aioxmpp.chatstates.ChatState`
:param message: pass true to indicate that we handle the
:data:`ACTIVE` state that is implied by
sending a content message.
:type message: :class:`bool`
:returns: whether a standalone notification must be sent for
this state update, respective if a chat state
notification must be included with the message.
:raises ValueError: if `message` is true and a state other
than :data:`ACTIVE` is passed.
"""
if message:
if state != chatstates_xso.ChatState.ACTIVE:
raise ValueError(
"Only the state ACTIVE can be sent with messages."
)
elif self._state == state:
return False
self._state = state
return self._strategy.sending
def no_reply(self):
"""
Call this method if the peer did not include a chat state
notification.
"""
self._strategy.no_reply()
def reset(self):
"""
Call this method on connection reset.
"""
self._strategy.reset()
|
