Mercurial > libervia-backend
annotate sat/plugins/plugin_xep_0198.py @ 4002:5245b675f7ad
plugin XEP-0313: don't wait for MAM to be retrieved in connection workflow:
MAM retrieval can be long, and can be done after connection, message just need to be
sorted when being inserted (i.e. frontends must do insort).
To avoid blocking connection for too long and result in bad UX and timeout risk, one2one
MAM message are not retrieved in background.
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 10 Mar 2023 17:22:45 +0100 |
parents | 32d714a8ea51 |
children | 524856bd7b19 |
rev | line source |
---|---|
3028 | 1 #!/usr/bin/env python3 |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
2 |
3133
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
3 # SàT plugin for managing Stream-Management |
3479 | 4 # Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
5 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
6 # This program is free software: you can redistribute it and/or modify |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
7 # it under the terms of the GNU Affero General Public License as published by |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
8 # the Free Software Foundation, either version 3 of the License, or |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
9 # (at your option) any later version. |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
10 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
11 # This program is distributed in the hope that it will be useful, |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
14 # GNU Affero General Public License for more details. |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
15 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
16 # You should have received a copy of the GNU Affero General Public License |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
17 # along with this program. If not, see <http://www.gnu.org/licenses/>. |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
18 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
19 from sat.core.i18n import _ |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
20 from sat.core.constants import Const as C |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
21 from sat.core import exceptions |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
22 from sat.core.log import getLogger |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
23 from twisted.words.protocols.jabber import client as jabber_client |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
24 from twisted.words.protocols.jabber import xmlstream |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
25 from twisted.words.xish import domish |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
26 from twisted.internet import defer |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
27 from twisted.internet import task, reactor |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
28 from functools import partial |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
29 from wokkel import disco, iwokkel |
3028 | 30 from zope.interface import implementer |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
31 import collections |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
32 import time |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
33 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
34 log = getLogger(__name__) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
35 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
36 PLUGIN_INFO = { |
3028 | 37 C.PI_NAME: "Stream Management", |
38 C.PI_IMPORT_NAME: "XEP-0198", | |
39 C.PI_TYPE: "XEP", | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
40 C.PI_MODES: C.PLUG_MODE_BOTH, |
3028 | 41 C.PI_PROTOCOLS: ["XEP-0198"], |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
42 C.PI_DEPENDENCIES: [], |
3028 | 43 C.PI_RECOMMENDATIONS: ["XEP-0045", "XEP-0313"], |
44 C.PI_MAIN: "XEP_0198", | |
45 C.PI_HANDLER: "yes", | |
46 C.PI_DESCRIPTION: _("""Implementation of Stream Management"""), | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
47 } |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
48 |
3028 | 49 NS_SM = "urn:xmpp:sm:3" |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
50 SM_ENABLED = '/enabled[@xmlns="' + NS_SM + '"]' |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
51 SM_RESUMED = '/resumed[@xmlns="' + NS_SM + '"]' |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
52 SM_FAILED = '/failed[@xmlns="' + NS_SM + '"]' |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
53 SM_R_REQUEST = '/r[@xmlns="' + NS_SM + '"]' |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
54 SM_A_REQUEST = '/a[@xmlns="' + NS_SM + '"]' |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
55 SM_H_REQUEST = '/h[@xmlns="' + NS_SM + '"]' |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
56 # Max number of stanza to send before requesting ack |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
57 MAX_STANZA_ACK_R = 5 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
58 # Max number of seconds before requesting ack |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
59 MAX_DELAY_ACK_R = 30 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
60 MAX_COUNTER = 2**32 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
61 RESUME_MAX = 5*60 |
2725
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
62 # if we don't have an answer to ACK REQUEST after this delay, connection is aborted |
2729
edd230651138
plugin XEP-0198: rised ACK_TIMEOUT + set req_time to None when timing out to be ready after resuming
Goffi <goffi@goffi.org>
parents:
2727
diff
changeset
|
63 ACK_TIMEOUT = 35 |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
64 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
65 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
66 class ProfileSessionData(object): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
67 out_counter = 0 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
68 in_counter = 0 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
69 session_id = None |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
70 location = None |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
71 session_max = None |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
72 # True when an ack answer is expected |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
73 ack_requested = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
74 last_ack_r = 0 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
75 disconnected_time = None |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
76 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
77 def __init__(self, callback, **kw): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
78 self.buffer = collections.deque() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
79 self.buffer_idx = 0 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
80 self._enabled = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
81 self.timer = None |
2725
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
82 # time used when doing a ack request |
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
83 # when it times out, connection is aborted |
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
84 self.req_timer = None |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
85 self.callback_data = (callback, kw) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
86 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
87 @property |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
88 def enabled(self): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
89 return self._enabled |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
90 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
91 @enabled.setter |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
92 def enabled(self, enabled): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
93 if enabled: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
94 if self._enabled: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
95 raise exceptions.InternalError( |
3028 | 96 "Stream Management can't be enabled twice") |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
97 self._enabled = True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
98 callback, kw = self.callback_data |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
99 self.timer = task.LoopingCall(callback, **kw) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
100 self.timer.start(MAX_DELAY_ACK_R, now=False) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
101 else: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
102 self._enabled = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
103 if self.timer is not None: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
104 self.timer.stop() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
105 self.timer = None |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
106 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
107 @property |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
108 def resume_enabled(self): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
109 return self.session_id is not None |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
110 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
111 def reset(self): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
112 self.enabled = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
113 self.buffer.clear() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
114 self.buffer_idx = 0 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
115 self.in_counter = self.out_counter = 0 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
116 self.session_id = self.location = None |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
117 self.ack_requested = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
118 self.last_ack_r = 0 |
2865
9213c6dff48d
plugin XEP-0198: reset req_timer on session.reset():
Goffi <goffi@goffi.org>
parents:
2796
diff
changeset
|
119 if self.req_timer is not None: |
9213c6dff48d
plugin XEP-0198: reset req_timer on session.reset():
Goffi <goffi@goffi.org>
parents:
2796
diff
changeset
|
120 if self.req_timer.active(): |
3028 | 121 log.error("req_timer has been called/cancelled but not reset") |
2865
9213c6dff48d
plugin XEP-0198: reset req_timer on session.reset():
Goffi <goffi@goffi.org>
parents:
2796
diff
changeset
|
122 else: |
9213c6dff48d
plugin XEP-0198: reset req_timer on session.reset():
Goffi <goffi@goffi.org>
parents:
2796
diff
changeset
|
123 self.req_timer.cancel() |
9213c6dff48d
plugin XEP-0198: reset req_timer on session.reset():
Goffi <goffi@goffi.org>
parents:
2796
diff
changeset
|
124 self.req_timer = None |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
125 |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
126 def getBufferCopy(self): |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
127 return list(self.buffer) |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
128 |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
129 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
130 class XEP_0198(object): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
131 # FIXME: location is not handled yet |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
132 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
133 def __init__(self, host): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
134 log.info(_("Plugin Stream Management initialization")) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
135 self.host = host |
3028 | 136 host.registerNamespace('sm', NS_SM) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
137 host.trigger.add("stream_hooks", self.addHooks) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
138 host.trigger.add("xml_init", self._XMLInitTrigger) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
139 host.trigger.add("disconnecting", self._disconnectingTrigger) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
140 host.trigger.add("disconnected", self._disconnectedTrigger) |
2727
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
141 try: |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
142 self._ack_timeout = int(host.memory.getConfig("", "ack_timeout", ACK_TIMEOUT)) |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
143 except ValueError: |
3028 | 144 log.error(_("Invalid ack_timeout value, please check your configuration")) |
2727
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
145 self._ack_timeout = ACK_TIMEOUT |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
146 if not self._ack_timeout: |
3028 | 147 log.info(_("Ack timeout disabled")) |
2727
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
148 else: |
3028 | 149 log.info(_("Ack timeout set to {timeout}s").format( |
2727
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
150 timeout=self._ack_timeout)) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
151 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
152 def profileConnecting(self, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
153 client._xep_0198_session = ProfileSessionData(callback=self.checkAcks, |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
154 client=client) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
155 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
156 def getHandler(self, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
157 return XEP_0198_handler(self) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
158 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
159 def addHooks(self, client, receive_hooks, send_hooks): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
160 """Add hooks to handle in/out stanzas counters""" |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
161 receive_hooks.append(partial(self.onReceive, client=client)) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
162 send_hooks.append(partial(self.onSend, client=client)) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
163 return True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
164 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
165 def _XMLInitTrigger(self, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
166 """Enable or resume a stream mangement""" |
3028 | 167 if not (NS_SM, 'sm') in client.xmlstream.features: |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
168 log.warning(_( |
3028 | 169 "Your server doesn't support stream management ({namespace}), this is " |
170 "used to improve connection problems detection (like network outages). " | |
171 "Please ask your server administrator to enable this feature.".format( | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
172 namespace=NS_SM))) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
173 return True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
174 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
175 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
176 # a disconnect timer from a previous disconnection may still be active |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
177 try: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
178 disconnect_timer = session.disconnect_timer |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
179 except AttributeError: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
180 pass |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
181 else: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
182 if disconnect_timer.active(): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
183 disconnect_timer.cancel() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
184 del session.disconnect_timer |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
185 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
186 if session.resume_enabled: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
187 # we are resuming a session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
188 resume_elt = domish.Element((NS_SM, 'resume')) |
3028 | 189 resume_elt['h'] = str(session.in_counter) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
190 resume_elt['previd'] = session.session_id |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
191 client.send(resume_elt) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
192 session.resuming = True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
193 # session.enabled will be set on <resumed/> reception |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
194 return False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
195 else: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
196 # we start a new session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
197 assert session.out_counter == 0 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
198 enable_elt = domish.Element((NS_SM, 'enable')) |
3028 | 199 enable_elt['resume'] = 'true' |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
200 client.send(enable_elt) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
201 session.enabled = True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
202 return True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
203 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
204 def _disconnectingTrigger(self, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
205 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
206 if session.enabled: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
207 self.sendAck(client) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
208 # This is a requested disconnection, so we can reset the session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
209 # to disable resuming and close normally the stream |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
210 session.reset() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
211 return True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
212 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
213 def _disconnectedTrigger(self, client, reason): |
2796
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
214 if client.is_component: |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
215 return True |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
216 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
217 session.enabled = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
218 if session.resume_enabled: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
219 session.disconnected_time = time.time() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
220 session.disconnect_timer = reactor.callLater(session.session_max, |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
221 client.disconnectProfile, |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
222 reason) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
223 # disconnectProfile must not be called at this point |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
224 # because session can be resumed |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
225 return False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
226 else: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
227 return True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
228 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
229 def checkAcks(self, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
230 """Request ack if needed""" |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
231 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
232 # log.debug("checkAcks (in_counter={}, out_counter={}, buf len={}, buf idx={})" |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
233 # .format(session.in_counter, session.out_counter, len(session.buffer), |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
234 # session.buffer_idx)) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
235 if session.ack_requested or not session.buffer: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
236 return |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
237 if (session.out_counter - session.buffer_idx >= MAX_STANZA_ACK_R |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
238 or time.time() - session.last_ack_r >= MAX_DELAY_ACK_R): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
239 self.requestAck(client) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
240 session.ack_requested = True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
241 session.last_ack_r = time.time() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
242 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
243 def updateBuffer(self, session, server_acked): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
244 """Update buffer and buffer_index""" |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
245 if server_acked > session.buffer_idx: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
246 diff = server_acked - session.buffer_idx |
3019
c9f03b1eb64d
plugin XEP-0198: catch and log an error if buffer is unexpectedly empty
Goffi <goffi@goffi.org>
parents:
3011
diff
changeset
|
247 try: |
3028 | 248 for i in range(diff): |
3019
c9f03b1eb64d
plugin XEP-0198: catch and log an error if buffer is unexpectedly empty
Goffi <goffi@goffi.org>
parents:
3011
diff
changeset
|
249 session.buffer.pop() |
c9f03b1eb64d
plugin XEP-0198: catch and log an error if buffer is unexpectedly empty
Goffi <goffi@goffi.org>
parents:
3011
diff
changeset
|
250 except IndexError: |
c9f03b1eb64d
plugin XEP-0198: catch and log an error if buffer is unexpectedly empty
Goffi <goffi@goffi.org>
parents:
3011
diff
changeset
|
251 log.error( |
3028 | 252 "error while cleaning buffer, invalid index (buffer is empty):\n" |
253 "diff = {diff}\n" | |
254 "server_acked = {server_acked}\n" | |
255 "buffer_idx = {buffer_id}".format( | |
3019
c9f03b1eb64d
plugin XEP-0198: catch and log an error if buffer is unexpectedly empty
Goffi <goffi@goffi.org>
parents:
3011
diff
changeset
|
256 diff=diff, server_acked=server_acked, |
c9f03b1eb64d
plugin XEP-0198: catch and log an error if buffer is unexpectedly empty
Goffi <goffi@goffi.org>
parents:
3011
diff
changeset
|
257 buffer_id=session.buffer_idx)) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
258 session.buffer_idx += diff |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
259 |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
260 def replayBuffer(self, client, buffer_, discard_results=False): |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
261 """Resend all stanza in buffer |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
262 |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
263 @param buffer_(collection.deque, list): buffer to replay |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
264 the buffer will be cleared by this method |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
265 @param discard_results(bool): if True, don't replay IQ result stanzas |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
266 """ |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
267 while True: |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
268 try: |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
269 stanza = buffer_.pop() |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
270 except IndexError: |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
271 break |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
272 else: |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
273 if ((discard_results |
3028 | 274 and stanza.name == 'iq' |
275 and stanza.getAttribute('type') == 'result')): | |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
276 continue |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
277 client.send(stanza) |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
278 |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
279 def sendAck(self, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
280 """Send an answer element with current IN counter""" |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
281 a_elt = domish.Element((NS_SM, 'a')) |
3028 | 282 a_elt['h'] = str(client._xep_0198_session.in_counter) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
283 client.send(a_elt) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
284 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
285 def requestAck(self, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
286 """Send a request element""" |
2725
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
287 session = client._xep_0198_session |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
288 r_elt = domish.Element((NS_SM, 'r')) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
289 client.send(r_elt) |
2725
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
290 if session.req_timer is not None: |
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
291 raise exceptions.InternalError("req_timer should not be set") |
2727
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
292 if self._ack_timeout: |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
293 session.req_timer = reactor.callLater(self._ack_timeout, self.onAckTimeOut, |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
294 client) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
295 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
296 def _connectionFailed(self, failure_, connector): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
297 normal_host, normal_port = connector.normal_location |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
298 del connector.normal_location |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
299 log.warning(_( |
3028 | 300 "Connection failed using location given by server (host: {host}, port: " |
301 "{port}), switching to normal host and port (host: {normal_host}, port: " | |
302 "{normal_port})".format(host=connector.host, port=connector.port, | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
303 normal_host=normal_host, normal_port=normal_port))) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
304 connector.host, connector.port = normal_host, normal_port |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
305 connector.connectionFailed = connector.connectionFailed_ori |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
306 del connector.connectionFailed_ori |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
307 return connector.connectionFailed(failure_) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
308 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
309 def onEnabled(self, enabled_elt, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
310 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
311 session.in_counter = 0 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
312 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
313 # we check that resuming is possible and that we have a session id |
3028 | 314 resume = C.bool(enabled_elt.getAttribute('resume')) |
315 session_id = enabled_elt.getAttribute('id') | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
316 if not session_id: |
3028 | 317 log.warning(_('Incorrect <enabled/> element received, no "id" attribute')) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
318 if not resume or not session_id: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
319 log.warning(_( |
3028 | 320 "You're server doesn't support session resuming with stream management, " |
321 "please contact your server administrator to enable it")) | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
322 return |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
323 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
324 session.session_id = session_id |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
325 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
326 # XXX: we disable resource binding, which must not be done |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
327 # when we resume the session. |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
328 client.factory.authenticator.res_binding = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
329 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
330 # location, in case server want resuming session to be elsewhere |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
331 try: |
3028 | 332 location = enabled_elt['location'] |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
333 except KeyError: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
334 pass |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
335 else: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
336 # TODO: handle IPv6 here (in brackets, cf. XEP) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
337 try: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
338 domain, port = location.split(':', 1) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
339 port = int(port) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
340 except ValueError: |
3028 | 341 log.warning(_("Invalid location received: {location}") |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
342 .format(location=location)) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
343 else: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
344 session.location = (domain, port) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
345 # we monkey patch connector to use the new location |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
346 connector = client.xmlstream.transport.connector |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
347 connector.normal_location = connector.host, connector.port |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
348 connector.host = domain |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
349 connector.port = port |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
350 connector.connectionFailed_ori = connector.connectionFailed |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
351 connector.connectionFailed = partial(self._connectionFailed, |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
352 connector=connector) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
353 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
354 # resuming time |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
355 try: |
3028 | 356 max_s = int(enabled_elt['max']) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
357 except (ValueError, KeyError) as e: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
358 if isinstance(e, ValueError): |
3028 | 359 log.warning(_('Invalid "max" attribute')) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
360 max_s = RESUME_MAX |
3028 | 361 log.info(_("Using default session max value ({max_s} s).".format( |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
362 max_s=max_s))) |
3028 | 363 log.info(_("Stream Management enabled")) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
364 else: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
365 log.info(_( |
3139
ca61807f724c
plugin XEP-0198: minor log formatting
Goffi <goffi@goffi.org>
parents:
3133
diff
changeset
|
366 "Stream Management enabled, with a resumption time of {res_m:.2f} min" |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
367 .format(res_m = max_s/60))) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
368 session.session_max = max_s |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
369 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
370 def onResumed(self, enabled_elt, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
371 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
372 assert not session.enabled |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
373 del session.resuming |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
374 server_acked = int(enabled_elt['h']) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
375 self.updateBuffer(session, server_acked) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
376 resend_count = len(session.buffer) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
377 # we resend all stanza which have not been received properly |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
378 self.replayBuffer(client, session.buffer) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
379 # now we can continue the session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
380 session.enabled = True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
381 d_time = time.time() - session.disconnected_time |
3028 | 382 log.info(_("Stream session resumed (disconnected for {d_time} s, {count} " |
383 "stanza(s) resent)").format(d_time=int(d_time), count=resend_count)) | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
384 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
385 def onFailed(self, failed_elt, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
386 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
387 condition_elt = failed_elt.firstChildElement() |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
388 buffer_ = session.getBufferCopy() |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
389 session.reset() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
390 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
391 try: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
392 del session.resuming |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
393 except AttributeError: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
394 # stream management can't be started at all |
3028 | 395 msg = _("Can't use stream management") |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
396 if condition_elt is None: |
3028 | 397 log.error(msg + '.') |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
398 else: |
3028 | 399 log.error(_("{msg}: {reason}").format( |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
400 msg=msg, reason=condition_elt.name)) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
401 else: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
402 # only stream resumption failed, we can try full session init |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
403 # XXX: we try to start full session init from this point, with many |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
404 # variables/attributes already initialised with a potentially different |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
405 # jid. This is experimental and may not be safe. It may be more |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
406 # secured to abord the connection and restart everything with a fresh |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
407 # client. |
3028 | 408 msg = _("stream resumption not possible, restarting full session") |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
409 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
410 if condition_elt is None: |
3028 | 411 log.warning('{msg}.'.format(msg=msg)) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
412 else: |
3028 | 413 log.warning("{msg}: {reason}".format( |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
414 msg=msg, reason=condition_elt.name)) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
415 # stream resumption failed, but we still can do normal stream management |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
416 # we restore attributes as if the session was new, and init stream |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
417 # we keep everything initialized, and only do binding, roster request |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
418 # and initial presence sending. |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
419 if client.conn_deferred.called: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
420 client.conn_deferred = defer.Deferred() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
421 else: |
3028 | 422 log.error("conn_deferred should be called at this point") |
423 plg_0045 = self.host.plugins.get('XEP-0045') | |
424 plg_0313 = self.host.plugins.get('XEP-0313') | |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
425 |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
426 # FIXME: we should call all loaded plugins with generic callbacks |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
427 # (e.g. prepareResume and resume), so a hot resuming can be done |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
428 # properly for all plugins. |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
429 |
3008
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
430 if plg_0045 is not None: |
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
431 # we have to remove joined rooms |
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
432 muc_join_args = plg_0045.popRooms(client) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
433 # we need to recreate roster |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
434 client.handlers.remove(client.roster) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
435 client.roster = client.roster.__class__(self.host) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
436 client.roster.setHandlerParent(client) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
437 # bind init is not done when resuming is possible, so we have to do it now |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
438 bind_init = jabber_client.BindInitializer(client.xmlstream) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
439 bind_init.required = True |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
440 d = bind_init.start() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
441 # we set the jid, which may have changed |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
442 d.addCallback(lambda __: setattr(client.factory.authenticator, "jid", client.jid)) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
443 # we call the trigger who will send the <enable/> element |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
444 d.addCallback(lambda __: self._XMLInitTrigger(client)) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
445 # then we have to re-request the roster, as changes may have occured |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
446 d.addCallback(lambda __: client.roster.requestRoster()) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
447 # we add got_roster to be sure to have roster before sending initial presence |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
448 d.addCallback(lambda __: client.roster.got_roster) |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
449 if plg_0313 is not None: |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
450 # we retrieve one2one MAM archives |
3541
888109774673
core: various changes and fixes to work with new storage and D-Bus bridge:
Goffi <goffi@goffi.org>
parents:
3479
diff
changeset
|
451 d.addCallback(lambda __: defer.ensureDeferred(plg_0313.resume(client))) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
452 # initial presence must be sent manually |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
453 d.addCallback(lambda __: client.presence.available()) |
3008
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
454 if plg_0045 is not None: |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
455 # we re-join MUC rooms |
3008
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
456 muc_d_list = defer.DeferredList( |
4001
32d714a8ea51
plugin XEP-0045: dot not wait for MAM retrieval to be completed:
Goffi <goffi@goffi.org>
parents:
3541
diff
changeset
|
457 [defer.ensureDeferred(plg_0045.join(*args)) |
32d714a8ea51
plugin XEP-0045: dot not wait for MAM retrieval to be completed:
Goffi <goffi@goffi.org>
parents:
3541
diff
changeset
|
458 for args in muc_join_args] |
32d714a8ea51
plugin XEP-0045: dot not wait for MAM retrieval to be completed:
Goffi <goffi@goffi.org>
parents:
3541
diff
changeset
|
459 ) |
3008
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
460 d.addCallback(lambda __: muc_d_list) |
3011
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
461 # at the end we replay the buffer, as those stanzas have probably not |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
462 # been received |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
463 d.addCallback(lambda __: self.replayBuffer(client, buffer_, |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
464 discard_results=True)) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
465 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
466 def onReceive(self, element, client): |
2796
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
467 if not client.is_component: |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
468 session = client._xep_0198_session |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
469 if session.enabled and element.name.lower() in C.STANZA_NAMES: |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
470 session.in_counter += 1 % MAX_COUNTER |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
471 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
472 def onSend(self, obj, client): |
2796
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
473 if not client.is_component: |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
474 session = client._xep_0198_session |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
475 if (session.enabled |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
476 and domish.IElement.providedBy(obj) |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
477 and obj.name.lower() in C.STANZA_NAMES): |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
478 session.out_counter += 1 % MAX_COUNTER |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
479 session.buffer.appendleft(obj) |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
480 self.checkAcks(client) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
481 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
482 def onAckRequest(self, r_elt, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
483 self.sendAck(client) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
484 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
485 def onAckAnswer(self, a_elt, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
486 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
487 session.ack_requested = False |
2727
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
488 if self._ack_timeout: |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
489 if session.req_timer is None: |
3008
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
490 log.error("req_timer should be set") |
2727
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
491 else: |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
492 session.req_timer.cancel() |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
493 session.req_timer = None |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
494 try: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
495 server_acked = int(a_elt['h']) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
496 except ValueError: |
3028 | 497 log.warning(_("Server returned invalid ack element, disabling stream " |
498 "management: {xml}").format(xml=a_elt)) | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
499 session.enabled = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
500 return |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
501 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
502 if server_acked > session.out_counter: |
3028 | 503 log.error(_("Server acked more stanzas than we have sent, disabling stream " |
504 "management.")) | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
505 session.reset() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
506 return |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
507 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
508 self.updateBuffer(session, server_acked) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
509 self.checkAcks(client) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
510 |
2725
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
511 def onAckTimeOut(self, client): |
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
512 """Called when a requested ACK has not been received in time""" |
3028 | 513 log.info(_("Ack was not received in time, aborting connection")) |
3133
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
514 try: |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
515 xmlstream = client.xmlstream |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
516 except AttributeError: |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
517 log.warning("xmlstream has already been terminated") |
3020
af9d71303605
plugin XEP-0198: don't crash if transport is None in onAckTimeOut
Goffi <goffi@goffi.org>
parents:
3019
diff
changeset
|
518 else: |
3133
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
519 transport = xmlstream.transport |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
520 if transport is None: |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
521 log.warning("transport was already removed") |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
522 else: |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
523 transport.abortConnection() |
2795
25639611c303
plugin XEP-0198: fixed typo resulting in a crash on resume
Goffi <goffi@goffi.org>
parents:
2729
diff
changeset
|
524 client._xep_0198_session.req_timer = None |
2725
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
525 |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
526 |
3028 | 527 @implementer(iwokkel.IDisco) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
528 class XEP_0198_handler(xmlstream.XMPPHandler): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
529 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
530 def __init__(self, plugin_parent): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
531 self.plugin_parent = plugin_parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
532 self.host = plugin_parent.host |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
533 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
534 def connectionInitialized(self): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
535 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
536 SM_ENABLED, self.plugin_parent.onEnabled, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
537 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
538 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
539 SM_RESUMED, self.plugin_parent.onResumed, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
540 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
541 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
542 SM_FAILED, self.plugin_parent.onFailed, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
543 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
544 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
545 SM_R_REQUEST, self.plugin_parent.onAckRequest, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
546 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
547 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
548 SM_A_REQUEST, self.plugin_parent.onAckAnswer, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
549 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
550 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
551 def getDiscoInfo(self, requestor, target, nodeIdentifier=""): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
552 return [disco.DiscoFeature(NS_SM)] |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
553 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
554 def getDiscoItems(self, requestor, target, nodeIdentifier=""): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
555 return [] |