libervia-backend: libervia/backend/plugins/plugin_xep

comparison libervia/backend/plugins/plugin_xep_0198.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	4b842c1fb686
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 # Max number of stanza to send before requesting ack
 MAX_STANZA_ACK_R = 5
 # Max number of seconds before requesting ack
 MAX_DELAY_ACK_R = 30
 MAX_COUNTER = 2**32
-RESUME_MAX = 5*60
+RESUME_MAX = 5 * 60
 # if we don't have an answer to ACK REQUEST after this delay, connection is aborted
 ACK_TIMEOUT = 35
 class ProfileSessionData(object):
 @enabled.setter
 def enabled(self, enabled):
 if enabled:
 if self._enabled:
-raise exceptions.InternalError(
+raise exceptions.InternalError("Stream Management can't be enabled twice")
-"Stream Management can't be enabled twice")
 self._enabled = True
 callback, kw = self.callback_data
 self.timer = task.LoopingCall(callback, **kw)
 self.timer.start(MAX_DELAY_ACK_R, now=False)
 else:
 # FIXME: location is not handled yet
 def __init__(self, host):
 log.info(_("Plugin Stream Management initialization"))
 self.host = host
-host.register_namespace('sm', NS_SM)
+host.register_namespace("sm", NS_SM)
 host.trigger.add("stream_hooks", self.add_hooks)
 host.trigger.add("xml_init", self._xml_init_trigger)
 host.trigger.add("disconnecting", self._disconnecting_trigger)
 host.trigger.add("disconnected", self._disconnected_trigger)
 try:
-self._ack_timeout = int(host.memory.config_get("", "ack_timeout", ACK_TIMEOUT))
+self._ack_timeout = int(
+host.memory.config_get("", "ack_timeout", ACK_TIMEOUT)
+)
 except ValueError:
 log.error(_("Invalid ack_timeout value, please check your configuration"))
 self._ack_timeout = ACK_TIMEOUT
 if not self._ack_timeout:
 log.info(_("Ack timeout disabled"))
 else:
-log.info(_("Ack timeout set to {timeout}s").format(
+log.info(_("Ack timeout set to {timeout}s").format(timeout=self._ack_timeout))
-timeout=self._ack_timeout))
 def profile_connecting(self, client):
-client._xep_0198_session = ProfileSessionData(callback=self.check_acks,
+client._xep_0198_session = ProfileSessionData(
-client=client)
+callback=self.check_acks, client=client
+)
 def get_handler(self, client):
 return XEP_0198_handler(self)
 def add_hooks(self, client, receive_hooks, send_hooks):
 send_hooks.append(partial(self.on_send, client=client))
 return True
 def _xml_init_trigger(self, client):
 """Enable or resume a stream mangement"""
-if not (NS_SM, 'sm') in client.xmlstream.features:
+if not (NS_SM, "sm") in client.xmlstream.features:
-log.warning(_(
+log.warning(
-"Your server doesn't support stream management ({namespace}), this is "
+_(
-"used to improve connection problems detection (like network outages). "
+"Your server doesn't support stream management ({namespace}), this is "
-"Please ask your server administrator to enable this feature.".format(
+"used to improve connection problems detection (like network outages). "
-namespace=NS_SM)))
+"Please ask your server administrator to enable this feature.".format(
+namespace=NS_SM
+)
+)
+)
 return True
 session = client._xep_0198_session
 # a disconnect timer from a previous disconnection may still be active
 try:
 disconnect_timer.cancel()
 del session.disconnect_timer
 if session.resume_enabled:
 # we are resuming a session
-resume_elt = domish.Element((NS_SM, 'resume'))
+resume_elt = domish.Element((NS_SM, "resume"))
-resume_elt['h'] = str(session.in_counter)
+resume_elt["h"] = str(session.in_counter)
-resume_elt['previd'] = session.session_id
+resume_elt["previd"] = session.session_id
 client.send(resume_elt)
 session.resuming = True
 # session.enabled will be set on <resumed/> reception
 return False
 else:
 # we start a new session
 assert session.out_counter == 0
-enable_elt = domish.Element((NS_SM, 'enable'))
+enable_elt = domish.Element((NS_SM, "enable"))
-enable_elt['resume'] = 'true'
+enable_elt["resume"] = "true"
 client.send(enable_elt)
 session.enabled = True
 return True
 def _disconnecting_trigger(self, client):
 return True
 session = client._xep_0198_session
 session.enabled = False
 if session.resume_enabled:
 session.disconnected_time = time.time()
-session.disconnect_timer = reactor.callLater(session.session_max,
+session.disconnect_timer = reactor.callLater(
-client.disconnect_profile,
+session.session_max, client.disconnect_profile, reason
-reason)
+)
 # disconnect_profile must not be called at this point
 # because session can be resumed
 return False
 else:
 return True
 # log.debug("check_acks (in_counter={}, out_counter={}, buf len={}, buf idx={})"
 #     .format(session.in_counter, session.out_counter, len(session.buffer),
 #             session.buffer_idx))
 if session.ack_requested or not session.buffer:
 return
-if (session.out_counter - session.buffer_idx >= MAX_STANZA_ACK_R
+if (
-or time.time() - session.last_ack_r >= MAX_DELAY_ACK_R):
+session.out_counter - session.buffer_idx >= MAX_STANZA_ACK_R
+or time.time() - session.last_ack_r >= MAX_DELAY_ACK_R
+):
 self.request_ack(client)
 session.ack_requested = True
 session.last_ack_r = time.time()
 def update_buffer(self, session, server_acked):
 log.error(
 "error while cleaning buffer, invalid index (buffer is empty):\n"
 "diff = {diff}\n"
 "server_acked = {server_acked}\n"
 "buffer_idx = {buffer_id}".format(
-diff=diff, server_acked=server_acked,
+diff=diff, server_acked=server_acked, buffer_id=session.buffer_idx
-buffer_id=session.buffer_idx))
+)
+)
 session.buffer_idx += diff
 def replay_buffer(self, client, buffer_, discard_results=False):
 """Resend all stanza in buffer
 try:
 stanza = buffer_.pop()
 except IndexError:
 break
 else:
-if ((discard_results
+if (
-and stanza.name == 'iq'
+discard_results
-and stanza.getAttribute('type') == 'result')):
+and stanza.name == "iq"
+and stanza.getAttribute("type") == "result"
+):
 continue
 client.send(stanza)
 def send_ack(self, client):
 """Send an answer element with current IN counter"""
-a_elt = domish.Element((NS_SM, 'a'))
+a_elt = domish.Element((NS_SM, "a"))
-a_elt['h'] = str(client._xep_0198_session.in_counter)
+a_elt["h"] = str(client._xep_0198_session.in_counter)
 client.send(a_elt)
 def request_ack(self, client):
 """Send a request element"""
 session = client._xep_0198_session
-r_elt = domish.Element((NS_SM, 'r'))
+r_elt = domish.Element((NS_SM, "r"))
 client.send(r_elt)
 if session.req_timer is not None:
 raise exceptions.InternalError("req_timer should not be set")
 if self._ack_timeout:
-session.req_timer = reactor.callLater(self._ack_timeout, self.on_ack_time_out,
+session.req_timer = reactor.callLater(
-client)
+self._ack_timeout, self.on_ack_time_out, client
+)
 def _connectionFailed(self, failure_, connector):
 normal_host, normal_port = connector.normal_location
 del connector.normal_location
-log.warning(_(
+log.warning(
-"Connection failed using location given by server (host: {host}, port: "
+_(
-"{port}), switching to normal host and port (host: {normal_host}, port: "
+"Connection failed using location given by server (host: {host}, port: "
-"{normal_port})".format(host=connector.host, port=connector.port,
+"{port}), switching to normal host and port (host: {normal_host}, port: "
-normal_host=normal_host, normal_port=normal_port)))
+"{normal_port})".format(
+host=connector.host,
+port=connector.port,
+normal_host=normal_host,
+normal_port=normal_port,
+)
+)
+)
 connector.host, connector.port = normal_host, normal_port
 connector.connectionFailed = connector.connectionFailed_ori
 del connector.connectionFailed_ori
 return connector.connectionFailed(failure_)
 def on_enabled(self, enabled_elt, client):
 session = client._xep_0198_session
 session.in_counter = 0
 # we check that resuming is possible and that we have a session id
-resume = C.bool(enabled_elt.getAttribute('resume'))
+resume = C.bool(enabled_elt.getAttribute("resume"))
-session_id = enabled_elt.getAttribute('id')
+session_id = enabled_elt.getAttribute("id")
 if not session_id:
 log.warning(_('Incorrect <enabled/> element received, no "id" attribute'))
 if not resume or not session_id:
-log.warning(_(
+log.warning(
-"You're server doesn't support session resuming with stream management, "
+_(
-"please contact your server administrator to enable it"))
+"You're server doesn't support session resuming with stream management, "
+"please contact your server administrator to enable it"
+)
+)
 return
 session.session_id = session_id
 # XXX: we disable resource binding, which must not be done
 #      when we resume the session.
 client.factory.authenticator.res_binding = False
 # location, in case server want resuming session to be elsewhere
 try:
-location = enabled_elt['location']
+location = enabled_elt["location"]
 except KeyError:
 pass
 else:
 # TODO: handle IPv6 here (in brackets, cf. XEP)
 try:
-domain, port = location.split(':', 1)
+domain, port = location.split(":", 1)
 port = int(port)
 except ValueError:
-log.warning(_("Invalid location received: {location}")
+log.warning(
-.format(location=location))
+_("Invalid location received: {location}").format(location=location)
+)
 else:
 session.location = (domain, port)
 # we monkey patch connector to use the new location
 connector = client.xmlstream.transport.connector
 connector.normal_location = connector.host, connector.port
 connector.host = domain
 connector.port = port
 connector.connectionFailed_ori = connector.connectionFailed
-connector.connectionFailed = partial(self._connectionFailed,
+connector.connectionFailed = partial(
-connector=connector)
+self._connectionFailed, connector=connector
+)
 # resuming time
 try:
-max_s = int(enabled_elt['max'])
+max_s = int(enabled_elt["max"])
 except (ValueError, KeyError) as e:
 if isinstance(e, ValueError):
 log.warning(_('Invalid "max" attribute'))
 max_s = RESUME_MAX
-log.info(_("Using default session max value ({max_s} s).".format(
+log.info(
-max_s=max_s)))
+_("Using default session max value ({max_s} s).".format(max_s=max_s))
+)
 log.info(_("Stream Management enabled"))
 else:
-log.info(_(
+log.info(
-"Stream Management enabled, with a resumption time of {res_m:.2f} min"
+_(
-.format(res_m = max_s/60)))
+"Stream Management enabled, with a resumption time of {res_m:.2f} min".format(
+res_m=max_s / 60
+)
+)
+)
 session.session_max = max_s
 def on_resumed(self, enabled_elt, client):
 session = client._xep_0198_session
 assert not session.enabled
 del session.resuming
-server_acked = int(enabled_elt['h'])
+server_acked = int(enabled_elt["h"])
 self.update_buffer(session, server_acked)
 resend_count = len(session.buffer)
 # we resend all stanza which have not been received properly
 self.replay_buffer(client, session.buffer)
 # now we can continue the session
 session.enabled = True
 d_time = time.time() - session.disconnected_time
-log.info(_("Stream session resumed (disconnected for {d_time} s, {count} "
+log.info(
-"stanza(s) resent)").format(d_time=int(d_time), count=resend_count))
+_(
+"Stream session resumed (disconnected for {d_time} s, {count} "
+"stanza(s) resent)"
+).format(d_time=int(d_time), count=resend_count)
+)
 def on_failed(self, failed_elt, client):
 session = client._xep_0198_session
 condition_elt = failed_elt.firstChildElement()
 buffer_ = session.get_buffer_copy()
 del session.resuming
 except AttributeError:
 # stream management can't be started at all
 msg = _("Can't use stream management")
 if condition_elt is None:
-log.error(msg + '.')
+log.error(msg + ".")
 else:
-log.error(_("{msg}: {reason}").format(
+log.error(_("{msg}: {reason}").format(msg=msg, reason=condition_elt.name))
-msg=msg, reason=condition_elt.name))
 else:
 # only stream resumption failed, we can try full session init
 # XXX: we try to start full session init from this point, with many
 #      variables/attributes already initialised with a potentially different
 #      jid. This is experimental and may not be safe. It may be more
 #      secured to abord the connection and restart everything with a fresh
 #      client.
 msg = _("stream resumption not possible, restarting full session")
 if condition_elt is None:
-log.warning('{msg}.'.format(msg=msg))
+log.warning("{msg}.".format(msg=msg))
 else:
-log.warning("{msg}: {reason}".format(
+log.warning("{msg}: {reason}".format(msg=msg, reason=condition_elt.name))
-msg=msg, reason=condition_elt.name))
 # stream resumption failed, but we still can do normal stream management
 # we restore attributes as if the session was new, and init stream
 # we keep everything initialized, and only do binding, roster request
 # and initial presence sending.
 if client.conn_deferred.called:
 client.conn_deferred = defer.Deferred()
 else:
 log.error("conn_deferred should be called at this point")
-plg_0045 = self.host.plugins.get('XEP-0045')
+plg_0045 = self.host.plugins.get("XEP-0045")
-plg_0313 = self.host.plugins.get('XEP-0313')
+plg_0313 = self.host.plugins.get("XEP-0313")
 # FIXME: we should call all loaded plugins with generic callbacks
 #        (e.g. prepareResume and resume), so a hot resuming can be done
 #        properly for all plugins.
 if plg_0045 is not None:
-# we have to remove joined rooms
+# we have to remove joined rooms
 muc_join_args = plg_0045.pop_rooms(client)
 # we need to recreate roster
 client.handlers.remove(client.roster)
 client.roster = client.roster.__class__(self.host)
 client.roster.setHandlerParent(client)
 # bind init is not done when resuming is possible, so we have to do it now
 bind_init = jabber_client.BindInitializer(client.xmlstream)
 bind_init.required = True
 d = bind_init.start()
 # we set the jid, which may have changed
-d.addCallback(lambda __: setattr(client.factory.authenticator, "jid", client.jid))
+d.addCallback(
+lambda __: setattr(client.factory.authenticator, "jid", client.jid)
+)
 # we call the trigger who will send the <enable/> element
 d.addCallback(lambda __: self._xml_init_trigger(client))
 # then we have to re-request the roster, as changes may have occured
 d.addCallback(lambda __: client.roster.request_roster())
 # we add got_roster to be sure to have roster before sending initial presence
 # initial presence must be sent manually
 d.addCallback(lambda __: client.presence.available())
 if plg_0045 is not None:
 # we re-join MUC rooms
 muc_d_list = defer.DeferredList(
-[defer.ensureDeferred(plg_0045.join(*args))
+[defer.ensureDeferred(plg_0045.join(*args)) for args in muc_join_args]
-for args in muc_join_args]
 )
 d.addCallback(lambda __: muc_d_list)
 # at the end we replay the buffer, as those stanzas have probably not
 # been received
-d.addCallback(lambda __: self.replay_buffer(client, buffer_,
+d.addCallback(
-discard_results=True))
+lambda __: self.replay_buffer(client, buffer_, discard_results=True)
+)
 def on_receive(self, element, client):
 if not client.is_component:
 session = client._xep_0198_session
 if session.enabled and element.name.lower() in C.STANZA_NAMES:
 session.in_counter += 1 % MAX_COUNTER
 def on_send(self, obj, client):
 if not client.is_component:
 session = client._xep_0198_session
-if (session.enabled
+if (
+session.enabled
 and domish.IElement.providedBy(obj)
-and obj.name.lower() in C.STANZA_NAMES):
+and obj.name.lower() in C.STANZA_NAMES
+):
 session.out_counter += 1 % MAX_COUNTER
 session.buffer.appendleft(obj)
 self.check_acks(client)
 def on_ack_request(self, r_elt, client):
 log.error("req_timer should be set")
 else:
 session.req_timer.cancel()
 session.req_timer = None
 try:
-server_acked = int(a_elt['h'])
+server_acked = int(a_elt["h"])
 except ValueError:
-log.warning(_("Server returned invalid ack element, disabling stream "
+log.warning(
-"management: {xml}").format(xml=a_elt))
+_(
+"Server returned invalid ack element, disabling stream "
+"management: {xml}"
+).format(xml=a_elt)
+)
 session.enabled = False
 return
 if server_acked > session.out_counter:
-log.error(_("Server acked more stanzas than we have sent, disabling stream "
+log.error(
-"management."))
+_(
+"Server acked more stanzas than we have sent, disabling stream "
+"management."
+)
+)
 session.reset()
 return
 self.update_buffer(session, server_acked)
 self.check_acks(client)

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_xep_0198.py @ 4270:0d7bb4df2343