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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
|
# Copyright (C) 2018 Philipp Hörist <philipp AT hoerist.com>
#
# This file is part of nbxmpp.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; If not, see <http://www.gnu.org/licenses/>.
import time
import logging
from nbxmpp.protocol import NS_STREAM_MGMT
from nbxmpp.protocol import NS_DELAY2
from nbxmpp.simplexml import Node
from nbxmpp.const import StreamState
from nbxmpp.util import LogAdapter
log = logging.getLogger('nbxmpp.smacks')
class Smacks:
"""
This is Smacks is the Stream Management class. It takes care of requesting
and sending acks. Also, it keeps track of the unhandled outgoing stanzas.
The dispatcher has to be able to access this class to increment the
number of handled stanzas
"""
def __init__(self, client):
self._client = client
self._out_h = 0 # Outgoing stanzas handled
self._in_h = 0 # Incoming stanzas handled
self._acked_h = 0 # Last acked stanza
self._uqueue = [] # Unhandled stanzas queue
self._old_uqueue = [] # Unhandled stanzas queue of the last session
# Max number of stanzas in queue before making a request
self.max_queue = 0
self._sm_supported = False
self.enabled = False # If SM is enabled
self._enable_sent = False # If we sent 'enable'
self.resumed = False # If the session was resumed
self.resume_in_progress = False
self.resume_supported = False # Does the session support resume
self._session_id = None
self._location = None
self._log = LogAdapter(log, {'context': client.log_context})
self.register_handlers()
@property
def sm_supported(self):
return self._sm_supported
@sm_supported.setter
def sm_supported(self, value):
self._log.info('Server supports detected: %s', value)
self._sm_supported = value
def delegate(self, stanza):
if stanza.getNamespace() != NS_STREAM_MGMT:
return
if stanza.getName() == 'resumed':
self._on_resumed(stanza)
elif stanza.getName() == 'failed':
self._on_failed(None, stanza, None)
def register_handlers(self):
self._client.register_handler(
'enabled', self._on_enabled, xmlns=NS_STREAM_MGMT)
self._client.register_handler(
'failed', self._on_failed, xmlns=NS_STREAM_MGMT)
self._client.register_handler(
'r', self._send_ack, xmlns=NS_STREAM_MGMT)
self._client.register_handler(
'a', self._on_ack, xmlns=NS_STREAM_MGMT)
def send_enable(self):
if not self.sm_supported:
return
enable = Node(NS_STREAM_MGMT + ' enable', attrs={'resume': 'true'})
self._client.send_nonza(enable, now=False)
self._log.debug('Send enable')
self._enable_sent = True
def _on_enabled(self, _con, stanza, _properties):
if self.enabled:
self._log.error('Received "enabled", but SM is already enabled')
return
resume = stanza.getAttr('resume')
if resume in ('true', 'True', '1'):
self.resume_supported = True
self._session_id = stanza.getAttr('id')
self._location = stanza.getAttr('location')
self.enabled = True
self._log.info('Received enabled, location: %s, resume supported: %s, '
'session-id: %s', self._location, resume, self._session_id)
def count_incoming(self, name):
if not self.enabled:
# Dont count while we didnt receive 'enabled'
return
if name in ('a', 'r', 'resumed', 'enabled'):
return
self._log.debug('IN, %s', name)
self._in_h += 1
def save_in_queue(self, stanza):
if not self._enable_sent and not self.resumed:
# We did not yet sent 'enable' so the server
# will not count our stanzas
return
if (stanza.getName() == 'message' and
stanza.getType() in ('chat', 'groupchat')):
timestamp = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
attrs = {'stamp': timestamp}
if stanza.getType() != 'groupchat':
# Dont leak our JID to Groupchats
attrs['from'] = stanza.getAttr('from')
stanza.addChild('delay', namespace=NS_DELAY2, attrs=attrs)
self._uqueue.append(stanza)
self._log.debug('OUT, %s', stanza.getName())
self._out_h += 1
if len(self._uqueue) > self.max_queue:
self._request_ack()
# Send an ack after 100 unacked messages
if (self._in_h - self._acked_h) > 100:
self._send_ack()
def _resend_queue(self):
"""
Resends unsent stanzas when a new session is established.
This way there won't be any lost outgoing messages even on failed
smacks resumes (but message duplicates are possible).
"""
if not self._old_uqueue:
return
self._log.info('Resend %s stanzas', len(self._old_uqueue))
for stanza in self._old_uqueue:
# Use dispatcher so we increment the counter
self._client.send_stanza(stanza)
self._old_uqueue = []
def resume_request(self):
if self._session_id is None:
self._log.error('Attempted to resume without a valid session id')
return
# Save old messages in an extra "queue" to avoid race conditions
# and to make it possible to replay stanzas even when resuming fails
# Add messages here (instead of overwriting) so that repeated
# connection errors don't delete unacked stanzas
# (uqueue should be empty in this case anyways)
self._old_uqueue += self._uqueue
self._uqueue = []
resume = Node(NS_STREAM_MGMT + ' resume',
attrs={'h': self._in_h, 'previd': self._session_id})
self._acked_h = self._in_h
self.resume_in_progress = True
self._client.send_nonza(resume, now=False)
def _on_resumed(self, stanza):
"""
Checks if the number of stanzas sent are the same as the
number of stanzas received by the server. Resends stanzas not received
by the server in the last session.
"""
self._log.info('Session resumption succeeded, session-id: %s',
self._session_id)
self._validate_ack(stanza, self._old_uqueue)
# Set our out h to the h we received
self._out_h = int(stanza.getAttr('h'))
self.enabled = True
self.resumed = True
self.resume_in_progress = False
self._client.set_state(StreamState.RESUME_SUCCESSFUL)
self._resend_queue()
def _send_ack(self, *args):
ack = Node(NS_STREAM_MGMT + ' a', attrs={'h': self._in_h})
self._acked_h = self._in_h
self._log.debug('Send ack, h: %s', self._in_h)
self._client.send_nonza(ack, now=False)
def close_session(self):
# We end the connection deliberately
# Reset the state -> no resume
self._log.info('Close session')
self._reset_state()
def _request_ack(self):
request = Node(NS_STREAM_MGMT + ' r')
self._log.debug('Request ack')
self._client.send_nonza(request, now=False)
def _on_ack(self, _stream, stanza, _properties):
if not self.enabled:
return
self._log.debug('Ack received, h: %s', stanza.getAttr('h'))
self._validate_ack(stanza, self._uqueue)
def _validate_ack(self, stanza, queue):
"""
Checks if the number of stanzas sent are the same as the
number of stanzas received by the server. Pops stanzas that were
handled by the server from the queue.
"""
count_server = stanza.getAttr('h')
if count_server is None:
self._log.error('Server did not send h attribute')
return
count_server = int(count_server)
diff = self._out_h - count_server
queue_size = len(queue)
if diff < 0:
self._log.error('Mismatch detected, our h: %d, server h: %d, queue: %d',
self._out_h, count_server, queue_size)
# Don't accumulate all messages in this case
# (they would otherwise all be resent on the next reconnect)
queue = []
elif queue_size < diff:
self._log.error('Mismatch detected, our h: %d, server h: %d, queue: %d',
self._out_h, count_server, queue_size)
else:
self._log.debug('Validate ack, our h: %d, server h: %d, queue: %d',
self._out_h, count_server, queue_size)
self._log.debug('Removing %d stanzas from queue', queue_size - diff)
while len(queue) > diff:
queue.pop(0)
def _on_failed(self, _stream, stanza, _properties):
'''
This can be called after 'enable' and 'resume'
'''
self._log.info('Negotiation failed')
error_text = stanza.getTagData('text')
if error_text is not None:
self._log.info(error_text)
if stanza.getTag('item-not-found') is not None:
self._log.info('Session timed out, last server h: %s',
stanza.getAttr('h'))
self._validate_ack(stanza, self._old_uqueue)
else:
for tag in stanza.getChildren():
if tag.getName() != 'text':
self._log.info(tag.getName())
if self.resume_in_progress:
# Reset state before sending Bind, because otherwise stanza
# will be counted and ack will be requested.
# _reset_state() also resets resume_in_progress
self._reset_state()
self._client.set_state(StreamState.RESUME_FAILED)
self._reset_state()
def _reset_state(self):
# Reset all values to default
self._out_h = 0
self._in_h = 0
self._acked_h = 0
self._uqueue = []
self._old_uqueue = []
self.enabled = False
self._enable_sent = False
self.resumed = False
self.resume_in_progress = False
self.resume_supported = False
self._session_id = None
self._location = None
self._enable_sent = False
|