Mercurial > libervia-backend
annotate sat/plugins/plugin_xep_0198.py @ 3606:25d3677701e7
doc: developer documentation explaining storage and pubsub cache
author | Goffi <goffi@goffi.org> |
---|---|
date | Thu, 29 Jul 2021 22:51:01 +0200 |
parents | 888109774673 |
children | 32d714a8ea51 |
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( |
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
457 [plg_0045.join(*args) for args in muc_join_args]) |
c8c68a3b0a79
plugins XEP-0045, XEP-0198: rejoin MUC rooms while a hot reconnection is done:
Goffi <goffi@goffi.org>
parents:
2865
diff
changeset
|
458 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
|
459 # 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
|
460 # been received |
93da7c6f8e0c
plugin XEP-0198: retrieve missing messages + send buffered ones on hot reconnection:
Goffi <goffi@goffi.org>
parents:
3008
diff
changeset
|
461 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
|
462 discard_results=True)) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
463 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
464 def onReceive(self, element, client): |
2796
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
465 if not client.is_component: |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
466 session = client._xep_0198_session |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
467 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
|
468 session.in_counter += 1 % MAX_COUNTER |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
469 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
470 def onSend(self, obj, client): |
2796
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
471 if not client.is_component: |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
472 session = client._xep_0198_session |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
473 if (session.enabled |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
474 and domish.IElement.providedBy(obj) |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
475 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
|
476 session.out_counter += 1 % MAX_COUNTER |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
477 session.buffer.appendleft(obj) |
fdc53c8a5439
plugin XEP-0198: don't use hooks on components
Goffi <goffi@goffi.org>
parents:
2795
diff
changeset
|
478 self.checkAcks(client) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
479 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
480 def onAckRequest(self, r_elt, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
481 self.sendAck(client) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
482 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
483 def onAckAnswer(self, a_elt, client): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
484 session = client._xep_0198_session |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
485 session.ack_requested = False |
2727
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
486 if self._ack_timeout: |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
487 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
|
488 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
|
489 else: |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
490 session.req_timer.cancel() |
59ac9284dee8
plugin XEP-0198: allow to customise/disable ack_timeout:
Goffi <goffi@goffi.org>
parents:
2725
diff
changeset
|
491 session.req_timer = None |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
492 try: |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
493 server_acked = int(a_elt['h']) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
494 except ValueError: |
3028 | 495 log.warning(_("Server returned invalid ack element, disabling stream " |
496 "management: {xml}").format(xml=a_elt)) | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
497 session.enabled = False |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
498 return |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
499 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
500 if server_acked > session.out_counter: |
3028 | 501 log.error(_("Server acked more stanzas than we have sent, disabling stream " |
502 "management.")) | |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
503 session.reset() |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
504 return |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
505 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
506 self.updateBuffer(session, server_acked) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
507 self.checkAcks(client) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
508 |
2725
d0466af33483
plugin XEP-0198: abort connection if ack is not received after a timeout:
Goffi <goffi@goffi.org>
parents:
2691
diff
changeset
|
509 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
|
510 """Called when a requested ACK has not been received in time""" |
3028 | 511 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
|
512 try: |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
513 xmlstream = client.xmlstream |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
514 except AttributeError: |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
515 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
|
516 else: |
3133
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
517 transport = xmlstream.transport |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
518 if transport is None: |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
519 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
|
520 else: |
87b8808ac49d
plugin XEP-0198: don't crash when xmlstream has been closed
Goffi <goffi@goffi.org>
parents:
3028
diff
changeset
|
521 transport.abortConnection() |
2795
25639611c303
plugin XEP-0198: fixed typo resulting in a crash on resume
Goffi <goffi@goffi.org>
parents:
2729
diff
changeset
|
522 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
|
523 |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
524 |
3028 | 525 @implementer(iwokkel.IDisco) |
2691
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
526 class XEP_0198_handler(xmlstream.XMPPHandler): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
527 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
528 def __init__(self, plugin_parent): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
529 self.plugin_parent = plugin_parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
530 self.host = plugin_parent.host |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
531 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
532 def connectionInitialized(self): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
533 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
534 SM_ENABLED, self.plugin_parent.onEnabled, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
535 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
536 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
537 SM_RESUMED, self.plugin_parent.onResumed, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
538 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
539 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
540 SM_FAILED, self.plugin_parent.onFailed, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
541 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
542 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
543 SM_R_REQUEST, self.plugin_parent.onAckRequest, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
544 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
545 self.xmlstream.addObserver( |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
546 SM_A_REQUEST, self.plugin_parent.onAckAnswer, client=self.parent |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
547 ) |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
548 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
549 def getDiscoInfo(self, requestor, target, nodeIdentifier=""): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
550 return [disco.DiscoFeature(NS_SM)] |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
551 |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
552 def getDiscoItems(self, requestor, target, nodeIdentifier=""): |
1ecceac3df96
plugin XEP-0198: Stream Management implementation:
Goffi <goffi@goffi.org>
parents:
diff
changeset
|
553 return [] |