diff options
| -rw-r--r-- | slixmpp/plugins/xep_0198/stream_management.py | 40 |
1 files changed, 31 insertions, 9 deletions
diff --git a/slixmpp/plugins/xep_0198/stream_management.py b/slixmpp/plugins/xep_0198/stream_management.py index 759e82e1..2b68faec 100644 --- a/slixmpp/plugins/xep_0198/stream_management.py +++ b/slixmpp/plugins/xep_0198/stream_management.py @@ -71,7 +71,8 @@ class XEP_0198(BasePlugin): self.window_counter = self.window - self.enabled = False + self.enabled_in = False + self.enabled_out = False self.unacked_queue = collections.deque() register_stanza_plugin(StreamFeatures, stanza.StreamManagement) @@ -131,6 +132,7 @@ class XEP_0198(BasePlugin): self.xmpp.add_filter('in', self._handle_incoming) self.xmpp.add_filter('out_sync', self._handle_outgoing) + self.xmpp.add_event_handler('disconnected', self.disconnected) self.xmpp.add_event_handler('session_end', self.session_end) def plugin_end(self): @@ -139,6 +141,7 @@ class XEP_0198(BasePlugin): self.xmpp.unregister_feature('sm', self.order) self.xmpp.unregister_feature('sm', self.resume_order) + self.xmpp.del_event_handler('disconnected', self.disconnected) self.xmpp.del_event_handler('session_end', self.session_end) self.xmpp.del_filter('in', self._handle_incoming) self.xmpp.del_filter('out_sync', self._handle_outgoing) @@ -154,9 +157,19 @@ class XEP_0198(BasePlugin): self.xmpp.remove_stanza(stanza.Ack) self.xmpp.remove_stanza(stanza.RequestAck) + def disconnected(self, event): + """Reset enabled state until we can resume/reenable.""" + log.debug("disconnected, disabling SM") + self.xmpp.event('sm_disabled', event) + self.enabled_in = False + self.enabled_out = False + def session_end(self, event): """Reset stream management state.""" - self.enabled = False + log.debug("session_end, disabling SM") + self.xmpp.event('sm_disabled', event) + self.enabled_in = False + self.enabled_out = False self.unacked_queue.clear() self.sm_id = None self.handled = 0 @@ -171,6 +184,7 @@ class XEP_0198(BasePlugin): def request_ack(self, e=None): """Request an ack from the server.""" + log.debug("requesting ack") req = stanza.RequestAck(self.xmpp) self.xmpp.send_raw(str(req)) @@ -193,9 +207,7 @@ class XEP_0198(BasePlugin): enable = stanza.Enable(self.xmpp) enable['resume'] = self.allow_resume enable.send() - self.enabled = True - self.handled = 0 - self.unacked_queue.clear() + log.debug("enabling SM") waiter = Waiter('enabled_or_failed', MatchMany([ @@ -204,11 +216,11 @@ class XEP_0198(BasePlugin): self.xmpp.register_handler(waiter) result = await waiter.wait() elif self.sm_id and self.allow_resume and 'bind' not in self.xmpp.features: - self.enabled = True resume = stanza.Resume(self.xmpp) resume['h'] = self.handled resume['previd'] = self.sm_id resume.send() + log.debug("resuming SM") # Wait for a response before allowing stream feature processing # to continue. The actual result processing will be done in the @@ -231,6 +243,8 @@ class XEP_0198(BasePlugin): self.xmpp.features.add('stream_management') if stanza['id']: self.sm_id = stanza['id'] + self.enabled_in = True + self.handled = 0 self.xmpp.event('sm_enabled', stanza) def _handle_resumed(self, stanza): @@ -239,6 +253,7 @@ class XEP_0198(BasePlugin): Raises a :term:`session_resumed` event. """ self.xmpp.features.add('stream_management') + self.enabled_in = True self._handle_ack(stanza) for id, stanza in self.unacked_queue: self.xmpp.send(stanza, use_filters=False) @@ -252,7 +267,8 @@ class XEP_0198(BasePlugin): Raises an :term:`sm_failed` event. """ - self.enabled = False + self.enabled_in = False + self.enabled_out = False self.unacked_queue.clear() self.xmpp.event('sm_failed', stanza) @@ -289,7 +305,7 @@ class XEP_0198(BasePlugin): def _handle_incoming(self, stanza): """Increment the handled counter for each inbound stanza.""" - if not self.enabled: + if not self.enabled_in: return stanza if isinstance(stanza, (Message, Presence, Iq)): @@ -299,7 +315,13 @@ class XEP_0198(BasePlugin): def _handle_outgoing(self, stanza): """Store outgoing stanzas in a queue to be acked.""" - if not self.enabled: + from slixmpp.plugins.xep_0198 import stanza as st + if isinstance(stanza, (st.Enable, st.Resume)): + self.enabled_out = True + self.unacked_queue.clear() + log.debug("enabling outgoing SM: %s" % stanza) + + if not self.enabled_out: return stanza if isinstance(stanza, (Message, Presence, Iq)): |