########################################################################
# File name: stream.py
# This file is part of: aioxmpp
#
# LICENSE
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program. If not, see
# <http://www.gnu.org/licenses/>.
#
########################################################################
"""
:mod:`~aioxmpp.stream` --- Stanza stream
########################################
The stanza stream is the layer of abstraction above the XML stream. It deals
with sending and receiving stream-level elements, mainly stanzas. It also
handles stream liveness and stream management.
It provides ways to track stanzas on their way to the remote, as far as that is
possible.
.. autoclass:: StanzaStream
Context managers
================
The following context managers can be used together with :class:`StanzaStream`
instances and the filters available on them.
.. autofunction:: iq_handler
.. autofunction:: message_handler
.. autofunction:: presence_handler
.. autofunction:: stanza_filter
Low-level stanza tracking
=========================
The following classes are used to track stanzas in the XML stream to the
server. This is independent of things like :xep:`Message Delivery Receipts
<0184>` (for which services are provided at :mod:`aioxmpp.tracking`); it
only provides tracking to the remote server and even that only if stream
management is used. Otherwise, it only provides tracking in the :mod:`aioxmpp`
internal queues.
.. autoclass:: StanzaToken
.. autoclass:: StanzaState
Filters
=======
The filters used by the :class:`StanzaStream` are implemented by the following
classes:
.. autoclass:: Filter
.. autoclass:: AppFilter
Exceptions
==========
.. autoclass:: DestructionRequested
"""
import asyncio
import contextlib
import functools
import logging
import warnings
from datetime import datetime, timedelta
from enum import Enum
from . import (
stanza,
stanza as stanza_,
errors,
custom_queue,
nonza,
callbacks,
protocol,
structs,
)
from .plugins import xep0199
from .utils import namespaces
[docs]class Filter:
"""
A filter chain for stanzas. The idea is to process a stanza through a
sequence of user- and service-definable functions.
Each function must either return the stanza it received as argument or
:data:`None`. If it returns :data:`None` the filtering aborts and the
caller of :meth:`filter` also receives :data:`None`.
Each function receives the result of the previous function for further
processing.
.. automethod:: register
.. automethod:: filter
.. automethod:: unregister
"""
class Token:
def __str__(self):
return "<{}.{} 0x{:x}>".format(
type(self).__module__,
type(self).__qualname__,
id(self))
def __init__(self):
super().__init__()
self._filter_order = []
[docs] def register(self, func, order):
"""
Register a function `func` as filter in the chain. `order` must be a
value which will be used to order the registered functions relative to
each other.
Functions with the same order are sorted in the order of their
addition, with the function which was added earliest first.
Remember that all values passed to `order` which are registered at the
same time in the same :class:`Filter` need to be at least partially
orderable with respect to each other.
Return an opaque token which is needed to unregister a function.
"""
token = self.Token()
self._filter_order.append((order, token, func))
self._filter_order.sort(key=lambda x: x[0])
return token
[docs] def filter(self, stanza_obj):
"""
Pass the given `stanza_obj` through the filter chain and return the
result of the chain. See :class:`Filter` for details on how the value
is passed through the registered functions.
"""
for _, _, func in self._filter_order:
stanza_obj = func(stanza_obj)
if stanza_obj is None:
return None
return stanza_obj
[docs] def unregister(self, token_to_remove):
"""
Unregister a function from the filter chain using the token returned by
:meth:`register`.
"""
for i, (_, token, _) in enumerate(self._filter_order):
if token == token_to_remove:
break
else:
raise ValueError("unregistered token: {!r}".format(
token_to_remove))
del self._filter_order[i]
[docs]class AppFilter(Filter):
"""
A specialized :class:`Filter` version. The only difference is in the
handling of the `order` argument to :meth:`register`:
.. automethod:: register
"""
[docs] def register(self, func, order=0):
"""
This method works exactly like :meth:`Filter.register`, but `order` has
a default value of ``0``.
"""
return super().register(func, order)
class PingEventType(Enum):
SEND_OPPORTUNISTIC = 0
SEND_NOW = 1
TIMEOUT = 2
[docs]class DestructionRequested(ConnectionError):
"""
Subclass of :class:`ConnectionError` indicating that the destruction of the
stream was requested by the user, directly or indirectly.
"""
[docs]class StanzaState(Enum):
"""
The various states an outgoing stanza can have.
.. attribute:: ACTIVE
The stanza has just been enqueued for sending and has not been taken
care of by the StanzaStream yet.
.. attribute:: SENT
The stanza has been sent over a stream with Stream Management enabled,
but not acked by the remote yet.
.. attribute:: ACKED
The stanza has been sent over a stream with Stream Management enabled
and has been acked by the remote.
This is a final state.
.. attribute:: SENT_WITHOUT_SM
The stanza has been sent over a stream without Stream Management enabled
or has been sent over a stream with Stream Management enabled, but for
which resumption has failed before the stanza has been acked.
This is a final state.
.. attribute:: ABORTED
The stanza has been retracted before it left the active queue.
This is a final state.
.. attribute:: DROPPED
The stanza has been dropped by one of the filters configured in the
:class:`StanzaStream`.
This is a final state.
.. attribute:: DISCONNECTED
The stream has been stopped (without SM) or closed before the stanza was
sent.
This is a final state.
"""
ACTIVE = 0
SENT = 1
ACKED = 2
SENT_WITHOUT_SM = 3
ABORTED = 4
DROPPED = 5
DISCONNECTED = 6
class StanzaErrorAwareListener:
def __init__(self, forward_to):
self._forward_to = forward_to
def data(self, stanza_obj):
if stanza_obj.type_.is_error:
return self._forward_to.error(
stanza_obj.error.to_exception()
)
return self._forward_to.data(stanza_obj)
def error(self, exc):
return self._forward_to.error(exc)
def is_valid(self):
return self._forward_to.is_valid()
[docs]class StanzaToken:
"""
A token to follow the processing of a `stanza`.
`on_state_change` may be a function which will be called with the token and
the new :class:`StanzaState` whenever the state of the token changes.
.. versionadded:: 0.8
Stanza tokens are :term:`awaitable`.
.. describe:: await token
.. describe:: yield from token
Wait until the stanza is either sent or failed to sent.
.. warning::
This only works with Python 3.5 or newer.
:raises ConnectionError: if the stanza enters
:attr:`~.StanzaState.DISCONNECTED` state.
:raises RuntimeError: if the stanza enters :attr:`~.StanzaState.ABORTED`
or :attr:`~.StanzaState.DROPPED` state.
:return: :data:`None`
If a coroutine awaiting a token is cancelled, the token is aborted. Use
:func:`asyncio.shield` to prevent this.
.. warning::
This is no guarantee that the recipient received the stanza. Without
stream management, it can not even guarenteed that the server has
seen the stanza.
This is primarily useful as a synchronisation primitive between the
sending of a stanza and another stream operation, such as closing the
stream.
.. autoattribute:: state
.. automethod:: abort
"""
__slots__ = ("stanza", "_state", "on_state_change", "_sent_future")
def __init__(self, stanza, *, on_state_change=None):
self.stanza = stanza
self._state = StanzaState.ACTIVE
self._sent_future = None
self.on_state_change = on_state_change
@property
[docs] def state(self):
"""
The current :class:`StanzaState` of the token. Tokens are created with
:attr:`StanzaState.ACTIVE`.
"""
return self._state
def _update_future(self):
if self._sent_future.done():
return
if self._state == StanzaState.DISCONNECTED:
self._sent_future.set_exception(ConnectionError("disconnected"))
elif self._state == StanzaState.DROPPED:
self._sent_future.set_exception(
RuntimeError("stanza dropped by filter")
)
elif self._state == StanzaState.ABORTED:
self._sent_future.set_exception(RuntimeError("stanza aborted"))
elif (self._state == StanzaState.SENT_WITHOUT_SM or
self._state == StanzaState.ACKED):
self._sent_future.set_result(None)
def _set_state(self, new_state):
self._state = new_state
if self.on_state_change is not None:
self.on_state_change(self, new_state)
if self._sent_future is not None:
self._update_future()
[docs] def abort(self):
"""
Abort the stanza. Attempting to call this when the stanza is in any
non-:class:`~StanzaState.ACTIVE`, non-:class:`~StanzaState.ABORTED`
state results in a :class:`RuntimeError`.
When a stanza is aborted, it will reside in the active queue of the
stream, not will be sent and instead discarded silently.
"""
if (self._state != StanzaState.ACTIVE and
self._state != StanzaState.ABORTED):
raise RuntimeError("cannot abort stanza (already sent)")
self._set_state(StanzaState.ABORTED)
def __repr__(self):
return "<StanzaToken id=0x{:016x}>".format(id(self))
@asyncio.coroutine
def __await__(self):
if self._sent_future is None:
self._sent_future = asyncio.Future()
self._update_future()
try:
yield from asyncio.shield(self._sent_future)
except asyncio.CancelledError:
if self._state == StanzaState.ACTIVE:
self.abort()
raise
__iter__ = __await__
[docs]class StanzaStream:
"""
A stanza stream. This is the next layer of abstraction above the XMPP XML
stream, which mostly deals with stanzas (but also with certain other
stream-level elements, such as :xep:`0198` Stream Management Request/Acks).
It is independent from a specific :class:`~aioxmpp.protocol.XMLStream`
instance. A :class:`StanzaStream` can be started with one XML stream,
stopped later and then resumed with another XML stream. The user of the
:class:`StanzaStream` has to make sure that the XML streams are compatible,
identity-wise (use the same JID).
`local_jid` may be the **bare** sender JID associated with the stanza
stream. This is required for compatibility with ejabberd. If it is omitted,
communication with ejabberd instances may not work.
`loop` may be used to explicitly specify the :class:`asyncio.BaseEventLoop`
to use, otherwise the current event loop is used.
`base_logger` can be used to explicitly specify a :class:`logging.Logger`
instance to fork off the logger from. The :class:`StanzaStream` will use a
child logger of `base_logger` called ``StanzaStream``.
.. versionchanged:: 0.4
The `local_jid` argument was added.
The stanza stream takes care of ensuring stream liveness. For that, pings
are sent in a periodic interval. If stream management is enabled, stream
management ack requests are used as pings, otherwise :xep:`0199` pings are
used.
The general idea of pinging is, to save computing power, to send pings only
when other stanzas are also about to be sent, if possible. The time window
for waiting for other stanzas is defined by
:attr:`ping_opportunistic_interval`. The general time which the
:class:`StanzaStream` waits between the reception of the previous ping and
contemplating the sending of the next ping is controlled by
:attr:`ping_interval`. See the attributes descriptions for details:
.. attribute:: ping_interval = timedelta(seconds=15)
A :class:`datetime.timedelta` instance which controls the time between a
ping response and starting the next ping. When this time elapses,
opportunistic mode is engaged for the time defined by
:attr:`ping_opportunistic_interval`.
.. attribute:: ping_opportunistic_interval = timedelta(seconds=15)
This is the time interval after :attr:`ping_interval`. During that
interval, :class:`StanzaStream` waits for other stanzas to be sent. If a
stanza gets send during that interval, the ping is fired. Otherwise, the
ping is fired after the interval.
After a ping has been sent, the response must arrive in a time of
:attr:`ping_interval` for the stream to be considered alive. If the
response fails to arrive within that interval, the stream fails (see
:attr:`on_failure`).
Starting/Stopping the stream:
.. automethod:: start
.. automethod:: stop
.. automethod:: wait_stop
.. automethod:: close
.. autoattribute:: running
.. automethod:: flush_incoming
Sending stanzas:
.. automethod:: send
.. automethod:: enqueue
.. method:: enqueue_stanza
Alias of :meth:`enqueue`.
.. deprecated:: 0.8
This alias is deprecated and will be removed in 1.0.
.. automethod:: send_and_wait_for_sent
.. automethod:: send_iq_and_wait_for_reply
Receiving stanzas:
.. automethod:: register_iq_request_coro
.. automethod:: unregister_iq_request_coro
.. automethod:: register_iq_response_future
.. automethod:: register_iq_response_callback
.. automethod:: unregister_iq_response
.. automethod:: register_message_callback
.. automethod:: unregister_message_callback
.. automethod:: register_presence_callback
.. automethod:: unregister_presence_callback
Inbound stanza filters allow to hook into the stanza processing by
replacing, modifying or otherwise processing stanza contents *before* the
above callbacks are invoked. With inbound stanza filters, there are no
restrictions as to what processing may take place on a stanza, as no one
but the stream may have references to its contents. See below for a
guideline on when to use stanza filters.
.. warning::
Raising an exception from within a stanza filter kills the stream.
Note that if a filter function drops an incoming stanza (by returning
:data:`None`), it **must** ensure that the client still behaves RFC
compliant.
.. attribute:: app_inbound_presence_filter
This is a :class:`AppFilter` based filter chain on inbound presence
stanzas. It can be used to attach application-specific filters.
.. attribute:: service_inbound_presence_filter
This is another filter chain for inbound presence stanzas. It runs
*before* the :attr:`app_inbound_presence_filter` chain and all functions
registered there must have :class:`service.Service` *classes* as `order`
value (see :meth:`Filter.register`).
This filter chain is intended to be used by library services, such as a
:xep:`115` implementation which may start a :xep:`30` lookup at the
target entity to resolve the capability hash or prime the :xep:`30`
cache with the service information obtained by interpreting the
:xep:`115` hash value.
.. attribute:: app_inbound_message_filter
This is a :class:`AppFilter` based filter chain on inbound message
stanzas. It can be used to attach application-specific filters.
.. attribute:: service_inbound_message_filter
This is the analogon of :attr:`service_inbound_presence_filter` for
:attr:`app_inbound_message_filter`.
Outbound stanza filters work similar to inbound stanza filters, but due to
their location in the processing chain and possible interactions with
senders of stanzas, there are some things to consider:
* Per convention, a outbound stanza filter **must not** modify any child
elements which are already present in the stanza when it receives the
stanza.
It may however add new child elements or remove existing child elements,
as well as copying and *then* modifying existing child elements.
* If the stanza filter replaces the stanza, it is responsible for making
sure that the new stanza has appropriate
:attr:`~.stanza.StanzaBase.from_`, :attr:`~.stanza.StanzaBase.to` and
:attr:`~.stanza.StanzaBase.id` values. There are no checks to enforce
this, because errorr handling at this point is peculiar. The stanzas will
be sent as-is.
* Similar to inbound filters, it is the responsibility of the filters that
if stanzas are dropped, the client still behaves RFC-compliant.
Now that you have been warned, here are the attributes for accessing the
outbound filter chains. These otherwise work exactly like their inbound
counterparts, but service filters run *after* application filters on
outbound processing.
.. attribute:: app_outbound_presence_filter
This is a :class:`AppFilter` based filter chain on outbound presence
stanzas. It can be used to attach application-specific filters.
Before using this attribute, make sure that you have read the notes
above.
.. attribute:: service_outbound_presence_filter
This is the analogon of :attr:`service_inbound_presence_filter`, but for
outbound presence. It runs *after* the
:meth:`app_outbound_presence_filter`.
Before using this attribute, make sure that you have read the notes
above.
.. attribute:: app_outbound_message_filter
This is a :class:`AppFilter` based filter chain on inbound message
stanzas. It can be used to attach application-specific filters.
Before using this attribute, make sure that you have read the notes
above.
.. attribute:: service_outbound_messages_filter
This is the analogon of :attr:`service_outbound_presence_filter`, but
for outbound messages.
Before using this attribute, make sure that you have read the notes
above.
When to use stanza filters? In general, applications will rarely need
them. However, services may make profitable use of them, and it is a
convenient way for them to inspect incoming or outgoing stanzas without
having to take up the registration slots (remember that
:meth:`register_message_callback` et. al. only allow *one* callback per
designator).
In general, whenever you do something which *supplements* the use of the
stanza with respect to the RFC but does not fulfill the orignial intent of
the stanza, it is advisable to use a filter instead of a callback on the
actual stanza.
Vice versa, if you were to develop a service which manages presence
subscriptions, it would be more correct to use
:meth:`register_presence_callback`; this prevents other services which try
to do the same from conflicting with you. You would then provide callbacks
to the application to let it learn about presence subscriptions.
Using stream management:
.. automethod:: start_sm
.. automethod:: resume_sm
.. automethod:: stop_sm
.. autoattribute:: sm_enabled
Stream management state inspection:
.. autoattribute:: sm_outbound_base
.. autoattribute:: sm_inbound_ctr
.. autoattribute:: sm_unacked_list
.. autoattribute:: sm_id
.. autoattribute:: sm_max
.. autoattribute:: sm_location
.. autoattribute:: sm_resumable
Miscellaneous:
.. autoattribute:: local_jid
Signals:
.. signal:: on_failure(exc)
Emits when the stream has failed, i.e. entered stopped state without
request by the user.
:param exc: The exception which caused the stream to fail.
:type exc: :class:`Exception`
A failure occurs whenever the main task of the :class:`StanzaStream`
(the one started by :meth:`start`) terminates with an exception.
Examples are :class:`ConnectionError` as raised upon a ping timeout and
any exceptions which may be raised by the
:meth:`aioxmpp.protocol.XMLStream.send_xso` method.
Before :meth:`on_failure` is emitted, the :class:`~.protocol.XMLStream`
is :meth:`~.protocol.XMLStream.abort`\ -ed if SM is enabled and
:meth:`~.protocol.XMLStream.close`\ -ed if SM is not enabled.
.. versionchanged:: 0.6
The closing behaviour was added.
.. signal:: on_stream_destroyed(reason)
The stream has been stopped in a manner which means that all state must
be discarded.
:param reason: The exception which caused the stream to be destroyeds
:type reason: :class:`Exception`
When this signal is emitted, others have or will most likely see
unavailable presence from the XMPP resource associated with the stream,
and stanzas sent in the mean time are not guaranteed to be received.
`reason` may be a :class:`DestructionRequested` instance to indicate
that the destruction was requested by the user, in some way.
There is no guarantee (it is not even likely) that it is possible to
send stanzas over the stream at the time this signal is emitted.
.. versionchanged:: 0.8
The `reason` argument was added.
.. signal:: on_stream_established()
When a stream is newly established, this signal is fired. This happens
whenever a non-SM stream is started and whenever a stream which
previously had SM disabled is started with SM enabled.
"""
_ALLOW_ENUM_COERCION = True
on_failure = callbacks.Signal()
on_stream_destroyed = callbacks.Signal()
on_stream_established = callbacks.Signal()
def __init__(self,
local_jid=None,
*,
loop=None,
base_logger=logging.getLogger("aioxmpp")):
super().__init__()
self._loop = loop or asyncio.get_event_loop()
self._logger = base_logger.getChild("StanzaStream")
self._task = None
self._local_jid = local_jid
self._active_queue = custom_queue.AsyncDeque(loop=self._loop)
self._incoming_queue = custom_queue.AsyncDeque(loop=self._loop)
self._iq_response_map = callbacks.TagDispatcher()
self._iq_request_map = {}
# list of running IQ request coroutines: used to cancel them when the
# stream is destroyed
self._iq_request_tasks = []
self._message_map = {}
self._presence_map = {}
self._ping_send_opportunistic = False
self._next_ping_event_at = None
self._next_ping_event_type = None
self._xmlstream_exception = None
self._established = False
self._closed = False
self.ping_interval = timedelta(seconds=15)
self.ping_opportunistic_interval = timedelta(seconds=15)
self._sm_enabled = False
self._broker_lock = asyncio.Lock(loop=loop)
self.app_inbound_presence_filter = AppFilter()
self.service_inbound_presence_filter = Filter()
self.app_inbound_message_filter = AppFilter()
self.service_inbound_message_filter = Filter()
self.app_outbound_presence_filter = AppFilter()
self.service_outbound_presence_filter = Filter()
self.app_outbound_message_filter = AppFilter()
self.service_outbound_message_filter = Filter()
@property
def local_jid(self):
"""
The `local_jid` argument to the constructor.
.. warning::
Changing this arbitrarily while the stream is running may have
unintended side effects.
"""
return self._local_jid
@local_jid.setter
[docs] def local_jid(self, value):
self._local_jid = value
def _coerce_enum(self, value, enum_class):
if not isinstance(value, enum_class):
if self._ALLOW_ENUM_COERCION:
warnings.warn(
"passing a non-enum value as type_ is deprecated and will "
"be invalid as of aioxmpp 1.0",
DeprecationWarning,
stacklevel=3)
return enum_class(value)
else:
raise TypeError("type_ must be {}, got {!r}".format(
enum_class.__name__,
value
))
return value
def _done_handler(self, task):
"""
Called when the main task (:meth:`_run`, :attr:`_task`) returns.
"""
try:
task.result()
except asyncio.CancelledError:
# normal termination
pass
except Exception as err:
try:
if self._sm_enabled:
self._xmlstream.abort()
else:
self._xmlstream.close()
except Exception:
pass
self.on_failure(err)
self._logger.exception("broker task failed")
def _xmlstream_failed(self, exc):
self._xmlstream_exception = exc
self.stop()
def _destroy_stream_state(self, exc):
"""
Destroy all state which does not make sense to keep after a disconnect
(without stream management).
"""
self._logger.debug("destroying stream state (exc=%r)", exc)
self._iq_response_map.close_all(exc)
for task in self._iq_request_tasks:
# we don’t need to remove, that’s handled by their
# add_done_callback
task.cancel()
while not self._active_queue.empty():
token = self._active_queue.get_nowait()
token._set_state(StanzaState.DISCONNECTED)
if self._established:
self.on_stream_destroyed(exc)
self._established = False
def _iq_request_coro_done(self, request, task):
"""
Called when an IQ request handler coroutine returns. `request` holds
the IQ request which triggered the excecution of the coroutine and
`task` is the :class:`asyncio.Task` which tracks the running coroutine.
Compose a response and send that response.
"""
self._iq_request_tasks.remove(task)
try:
payload = task.result()
except errors.XMPPError as err:
response = request.make_reply(type_=structs.IQType.ERROR)
response.error = stanza.Error.from_exception(err)
except Exception:
response = request.make_reply(type_=structs.IQType.ERROR)
response.error = stanza.Error(
condition=(namespaces.stanzas, "undefined-condition"),
type_=structs.ErrorType.CANCEL,
)
self._logger.exception("IQ request coroutine failed")
else:
response = request.make_reply(type_=structs.IQType.RESULT)
response.payload = payload
self.enqueue(response)
def _process_incoming_iq(self, stanza_obj):
"""
Process an incoming IQ stanza `stanza_obj`. Calls the response handler,
spawns a request handler coroutine or drops the stanza while logging a
warning if no handler can be found.
"""
self._logger.debug("incoming iq: %r", stanza_obj)
if stanza_obj.type_.is_response:
# iq response
self._logger.debug("iq is response")
keys = [(stanza_obj.from_, stanza_obj.id_)]
if self._local_jid is not None:
# needed for some servers
if keys[0][0] == self._local_jid:
keys.append((None, keys[0][1]))
elif keys[0][0] is None:
keys.append((self._local_jid, keys[0][1]))
for key in keys:
try:
self._iq_response_map.unicast(key, stanza_obj)
self._logger.debug("iq response delivered to key %r", key)
break
except KeyError:
pass
else:
self._logger.warning(
"unexpected IQ response: from=%r, id=%r",
*key)
else:
# iq request
self._logger.debug("iq is request")
key = (stanza_obj.type_, type(stanza_obj.payload))
try:
coro = self._iq_request_map[key]
except KeyError:
self._logger.warning(
"unhandleable IQ request: from=%r, type_=%r, payload=%r",
stanza_obj.from_,
stanza_obj.type_,
stanza_obj.payload
)
response = stanza_obj.make_reply(type_=structs.IQType.ERROR)
response.error = stanza.Error(
condition=(namespaces.stanzas,
"feature-not-implemented"),
)
self.enqueue(response)
return
task = asyncio.async(coro(stanza_obj))
task.add_done_callback(
functools.partial(
self._iq_request_coro_done,
stanza_obj))
self._iq_request_tasks.append(task)
self._logger.debug("started task to handle request: %r", task)
def _process_incoming_message(self, stanza_obj):
"""
Process an incoming message stanza `stanza_obj`.
"""
self._logger.debug("incoming messgage: %r", stanza_obj)
stanza_obj = self.service_inbound_message_filter.filter(stanza_obj)
if stanza_obj is None:
self._logger.debug("incoming message dropped by service "
"filter chain")
return
stanza_obj = self.app_inbound_message_filter.filter(stanza_obj)
if stanza_obj is None:
self._logger.debug("incoming message dropped by application "
"filter chain")
return
# XXX: this should be fixed better, to avoid the ambiguity between bare
# JID wildcarding and stanzas originating from bare JIDs
# also, I don’t like how we handle from_=None now
if stanza_obj.from_ is None:
stanza_obj.from_ = self._local_jid
keys = [(stanza_obj.type_, stanza_obj.from_),
(stanza_obj.type_, stanza_obj.from_.bare()),
(None, stanza_obj.from_),
(None, stanza_obj.from_.bare()),
(stanza_obj.type_, None),
(None, None)]
for key in keys:
try:
cb = self._message_map[key]
except KeyError:
continue
self._logger.debug("dispatching message using key %r to %r",
key, cb)
self._loop.call_soon(cb, stanza_obj)
break
else:
self._logger.warning(
"unsolicited message dropped: from=%r, type=%r, id=%r",
stanza_obj.from_,
stanza_obj.type_,
stanza_obj.id_
)
def _process_incoming_presence(self, stanza_obj):
"""
Process an incoming presence stanza `stanza_obj`.
"""
self._logger.debug("incoming presence: %r", stanza_obj)
stanza_obj = self.service_inbound_presence_filter.filter(stanza_obj)
if stanza_obj is None:
self._logger.debug("incoming presence dropped by service filter"
" chain")
return
stanza_obj = self.app_inbound_presence_filter.filter(stanza_obj)
if stanza_obj is None:
self._logger.debug("incoming presence dropped by application "
"filter chain")
return
keys = [(stanza_obj.type_, stanza_obj.from_),
(stanza_obj.type_, None)]
for key in keys:
try:
cb = self._presence_map[key]
except KeyError:
continue
self._logger.debug("dispatching presence using key: %r", key)
self._loop.call_soon(cb, stanza_obj)
break
else:
self._logger.warning(
"unhandled presence dropped: from=%r, type=%r, id=%r",
stanza_obj.from_,
stanza_obj.type_,
stanza_obj.id_
)
def _process_incoming_erroneous_stanza(self, stanza_obj, exc):
self._logger.debug(
"erroneous stanza received (may be incomplete): %r",
stanza_obj
)
try:
type_ = stanza_obj.type_
except AttributeError:
# ugh, type is broken
# exit early
self._logger.debug(
"stanza has broken type, cannot properly handle"
)
return
if type_.is_response:
try:
from_ = stanza_obj.from_
id_ = stanza_obj.id_
except AttributeError:
pass
else:
if isinstance(stanza_obj, stanza.IQ):
self._logger.debug(
"erroneous stanza can be forwarded to handlers as "
"error"
)
key = (from_, id_)
try:
self._iq_response_map.unicast_error(
key,
errors.ErroneousStanza(stanza_obj)
)
except KeyError:
pass
elif isinstance(exc, stanza.UnknownIQPayload):
reply = stanza_obj.make_error(error=stanza.Error(condition=(
namespaces.stanzas,
"feature-not-implemented")
))
self.enqueue(reply)
elif isinstance(exc, stanza.PayloadParsingError):
reply = stanza_obj.make_error(error=stanza.Error(condition=(
namespaces.stanzas,
"bad-request")
))
self.enqueue(reply)
def _process_incoming(self, xmlstream, queue_entry):
"""
Dispatch to the different methods responsible for the different stanza
types or handle a non-stanza stream-level element from `stanza_obj`,
which has arrived over the given `xmlstream`.
"""
stanza_obj, exc = queue_entry
# first, handle SM stream objects
if isinstance(stanza_obj, nonza.SMAcknowledgement):
self._logger.debug("received SM ack: %r", stanza_obj)
if not self._sm_enabled:
self._logger.warning("received SM ack, but SM not enabled")
return
self.sm_ack(stanza_obj.counter)
if self._next_ping_event_type == PingEventType.TIMEOUT:
self._logger.debug("resetting ping timeout")
self._next_ping_event_type = PingEventType.SEND_OPPORTUNISTIC
self._next_ping_event_at = (datetime.utcnow() +
self.ping_interval)
return
elif isinstance(stanza_obj, nonza.SMRequest):
self._logger.debug("received SM request: %r", stanza_obj)
if not self._sm_enabled:
self._logger.warning("received SM request, but SM not enabled")
return
response = nonza.SMAcknowledgement()
response.counter = self._sm_inbound_ctr
self._logger.debug("sending SM ack: %r", response)
xmlstream.send_xso(response)
return
# raise if it is not a stanza
if not isinstance(stanza_obj, stanza.StanzaBase):
raise RuntimeError(
"unexpected stanza class: {}".format(stanza_obj))
# now handle stanzas, these always increment the SM counter
if self._sm_enabled:
self._sm_inbound_ctr += 1
# check if the stanza has errors
if exc is not None:
self._process_incoming_erroneous_stanza(stanza_obj, exc)
return
if isinstance(stanza_obj, stanza.IQ):
self._process_incoming_iq(stanza_obj)
elif isinstance(stanza_obj, stanza.Message):
self._process_incoming_message(stanza_obj)
elif isinstance(stanza_obj, stanza.Presence):
self._process_incoming_presence(stanza_obj)
[docs] def flush_incoming(self):
"""
Flush all incoming queues to the respective processing methods. The
handlers are called as usual, thus it may require at least one
iteration through the asyncio event loop before effects can be seen.
The incoming queues are empty after a call to this method.
It is legal (but pretty useless) to call this method while the stream
is :attr:`running`.
"""
while True:
try:
stanza_obj = self._incoming_queue.get_nowait()
except asyncio.QueueEmpty:
break
self._process_incoming(None, stanza_obj)
def _send_stanza(self, xmlstream, token):
"""
Send a stanza token `token` over the given `xmlstream`.
Only sends if the `token` has not been aborted (see
:meth:`StanzaToken.abort`). Sends the state of the token acoording to
:attr:`sm_enabled`.
"""
if token.state == StanzaState.ABORTED:
return
stanza_obj = token.stanza
if isinstance(stanza_obj, stanza.Presence):
stanza_obj = self.app_outbound_presence_filter.filter(
stanza_obj
)
if stanza_obj is not None:
stanza_obj = self.service_outbound_presence_filter.filter(
stanza_obj
)
elif isinstance(stanza_obj, stanza.Message):
stanza_obj = self.app_outbound_message_filter.filter(
stanza_obj
)
if stanza_obj is not None:
stanza_obj = self.service_outbound_message_filter.filter(
stanza_obj
)
if stanza_obj is None:
token._set_state(StanzaState.DROPPED)
self._logger.debug("outgoing stanza %r dropped by filter chain",
token.stanza)
return
self._logger.debug("forwarding stanza to xmlstream: %r",
stanza_obj)
xmlstream.send_xso(stanza_obj)
if self._sm_enabled:
token._set_state(StanzaState.SENT)
self._sm_unacked_list.append(token)
else:
token._set_state(StanzaState.SENT_WITHOUT_SM)
def _process_outgoing(self, xmlstream, token):
"""
Process the current outgoing stanza `token` and also any other outgoing
stanza which is currently in the active queue. After all stanzas have
been processed, use :meth:`_send_ping` to allow an opportunistic ping
to be sent.
"""
self._send_stanza(xmlstream, token)
# try to send a bulk
while True:
try:
token = self._active_queue.get_nowait()
except asyncio.QueueEmpty:
break
self._send_stanza(xmlstream, token)
self._send_ping(xmlstream)
def _recv_pong(self, stanza):
"""
Process the reception of a XEP-0199 ping reply.
"""
if not self.running:
return
if self._next_ping_event_type != PingEventType.TIMEOUT:
return
self._next_ping_event_type = PingEventType.SEND_OPPORTUNISTIC
self._next_ping_event_at = datetime.utcnow() + self.ping_interval
def _send_ping(self, xmlstream):
"""
Opportunistically send a ping over the given `xmlstream`.
If stream management is enabled, an SM request is always sent,
independent of the current ping state. Otherwise, a XEP-0199 ping is
sent if and only if we are currently in the opportunistic ping interval
(see :attr:`ping_opportunistic_interval`).
If a ping is sent, and we are currently not waiting for a pong to be
received, the ping timeout is configured.
"""
if not self._ping_send_opportunistic:
return
if self._sm_enabled:
self._logger.debug("sending SM req")
xmlstream.send_xso(nonza.SMRequest())
else:
request = stanza.IQ(type_=structs.IQType.GET)
request.payload = xep0199.Ping()
request.autoset_id()
self.register_iq_response_callback(
None,
request.id_,
self._recv_pong
)
self._logger.debug("sending XEP-0199 ping: %r", request)
xmlstream.send_xso(request)
self._ping_send_opportunistic = False
if self._next_ping_event_type != PingEventType.TIMEOUT:
self._logger.debug("configuring ping timeout")
self._next_ping_event_at = datetime.utcnow() + self.ping_interval
self._next_ping_event_type = PingEventType.TIMEOUT
def _process_ping_event(self, xmlstream):
"""
Process a ping timed event on the current `xmlstream`.
"""
if self._next_ping_event_type == PingEventType.SEND_OPPORTUNISTIC:
self._logger.debug("ping: opportunistic interval started")
self._next_ping_event_at += self.ping_opportunistic_interval
self._next_ping_event_type = PingEventType.SEND_NOW
# ping send opportunistic is always true for sm
if not self._sm_enabled:
self._ping_send_opportunistic = True
elif self._next_ping_event_type == PingEventType.SEND_NOW:
self._logger.debug("ping: requiring ping to be sent now")
self._send_ping(xmlstream)
elif self._next_ping_event_type == PingEventType.TIMEOUT:
self._logger.warning("ping: response timeout tripped")
raise ConnectionError("ping timeout")
else:
raise RuntimeError("unknown ping event type: {!r}".format(
self._next_ping_event_type))
[docs] def register_iq_response_callback(self, from_, id_, cb):
"""
Register a callback function `cb` to be called when a IQ stanza with
type ``result`` or ``error`` is recieved from the
:class:`~aioxmpp.JID` `from_` with the id `id_`.
The callback is called at most once.
.. note::
In contrast to :meth:`register_iq_response_future`, errors which
occur on a level below XMPP stanzas cannot be caught using a
callback.
If you need notification about other errors and still want to use
callbacks, use of a future with
:meth:`asyncio.Future.add_done_callback` is recommended.
"""
self._iq_response_map.add_listener(
(from_, id_),
callbacks.OneshotAsyncTagListener(cb, loop=self._loop)
)
self._logger.debug("iq response callback registered: from=%r, id=%r",
from_, id_)
[docs] def register_iq_response_future(self, from_, id_, fut):
"""
Register a future `fut` for an IQ stanza with type ``result`` or
``error`` from the :class:`~aioxmpp.JID` `from_` with the id
`id_`.
If the type of the IQ stanza is ``result``, the stanza is set as result
to the future. If the type of the IQ stanza is ``error``, the stanzas
error field is converted to an exception and set as the exception of
the future.
The future might also receive different exceptions:
* :class:`.errors.ErroneousStanza`, if the response stanza received
could not be parsed.
Note that this exception is not emitted if the ``from`` address of
the stanza is unset, because the code cannot determine whether a
sender deliberately used an erroneous address to make parsing fail
or no sender address was used. In the former case, an attacker could
use that to inject a stanza which would be taken as a stanza from the
peer server. Thus, the future will never be fulfilled in these
cases.
Also note that this exception does not derive from
:class:`.errors.XMPPError`, as it cannot provide the same
attributes. Instead, it dervies from :class:`.errors.StanzaError`,
from which :class:`.errors.XMPPError` also derives; to catch all
possible stanza errors, catching :class:`.errors.StanzaError` is
sufficient and future-proof.
* :class:`ConnectionError` if the stream is :meth:`stop`\ -ped (only if
SM is not enabled) or :meth:`close`\ -ed.
* Any :class:`Exception` which may be raised from
:meth:`~.protocol.XMLStream.send_xso`, which are generally also
:class:`ConnectionError` or at least :class:`OSError` subclasses.
"""
self._iq_response_map.add_listener(
(from_, id_),
StanzaErrorAwareListener(
callbacks.FutureListener(fut)
)
)
self._logger.debug("iq response future registered: from=%r, id=%r",
from_, id_)
[docs] def unregister_iq_response(self, from_, id_):
"""
Unregister a registered callback or future for the IQ response
identified by `from_` and `id_`. See
:meth:`register_iq_response_future` or
:meth:`register_iq_response_callback` for details on the arguments
meanings and how to register futures and callbacks respectively.
.. note::
Futures will automatically be unregistered when they are cancelled.
"""
self._iq_response_map.remove_listener((from_, id_))
self._logger.debug("iq response unregistered: from=%r, id=%r",
from_, id_)
[docs] def register_iq_request_coro(self, type_, payload_cls, coro):
"""
Register a coroutine to run when an IQ request is received.
:param type_: IQ type to react to (must be a request type).
:type type_: :class:`~aioxmpp.IQType`
:param payload_cls: Payload class to react to (subclass of :class:`~xso.XSO`)
:type payload_cls: :class:`~.XMLStreamClass`
:param coro: Coroutine to run
:raises ValueError: if there is already a coroutine registered for this
targe
:raises ValueError: if `type_` is not a request IQ type
:raises ValueError: if `type_` is not a valid
:class:`~.IQType` (and cannot be cast to a
:class:`~.IQType`)
The coroutine `coro` will be spawned whenever an IQ stanza with the
given `type_` and payload being an instance of the `payload_cls` is
received. The coroutine must return a valid value for the
:attr:`.IQ.payload` attribute. The value will be set as the
payload attribute value of an IQ response (with type
:attr:`~.IQType.RESULT`) which is generated and sent by the stream.
If the coroutine raises an exception, it will be converted to a
:class:`~.stanza.Error` object. That error object is then used as
payload for an IQ response (with type :attr:`~.IQType.ERROR`) which is
generated and sent by the stream.
If the exception is a subclass of :class:`aioxmpp.errors.XMPPError`, it
is converted to an :class:`~.stanza.Error` instance directly.
Otherwise, it is wrapped in a :class:`aioxmpp.XMPPCancelError`
with ``undefined-condition``.
For this to work, `payload_cls` *must* be registered using
:meth:`~.IQ.as_payload_class`. Otherwise, the payload will
not be recognised by the stream parser and the IQ is automatically
responded to with a ``feature-not-implemented`` error.
.. versionadded:: 0.6
If the stream is :meth:`stop`\ -ped (only if SM is not enabled) or
:meth:`close`\ ed, running IQ response coroutines are
:meth:`asyncio.Task.cancel`\ -led.
To protect against that, fork from your coroutine using
:func:`asyncio.ensure_future`.
.. versionchanged:: 0.7
The `type_` argument is now supposed to be a :class:`~.IQType`
member.
.. deprecated:: 0.7
Passing a :class:`str` as `type_` argument is deprecated and will
raise a :class:`TypeError` as of the 1.0 release. See the Changelog
for :ref:`api-changelog-0.7` for further details on how to upgrade
your code efficiently.
"""
type_ = self._coerce_enum(type_, structs.IQType)
if not type_.is_request:
raise ValueError(
"{!r} is not a request IQType".format(type_)
)
key = type_, payload_cls
if key in self._iq_request_map:
raise ValueError("only one listener is allowed per tag")
self._iq_request_map[key] = coro
self._logger.debug(
"iq request coroutine registered: type=%r, payload=%r",
type_, payload_cls)
[docs] def unregister_iq_request_coro(self, type_, payload_cls):
"""
Unregister a coroutine previously registered with
:meth:`register_iq_request_coro`.
:param type_: IQ type to react to (must be a request type).
:type type_: :class:`~structs.IQType`
:param payload_cls: Payload class to react to (subclass of :class:`~xso.XSO`)
:type payload_cls: :class:`~.XMLStreamClass`
:raises KeyError: if no coroutine has been registered for the given
``(type_, payload_cls)`` pair
:raises ValueError: if `type_` is not a valid
:class:`~.IQType` (and cannot be cast to a
:class:`~.IQType`)
The match is solely made using the `type_` and `payload_cls` arguments,
which have the same meaning as in :meth:`register_iq_request_coro`.
.. versionchanged:: 0.7
The `type_` argument is now supposed to be a :class:`~.IQType`
member.
.. deprecated:: 0.7
Passing a :class:`str` as `type_` argument is deprecated and will
raise a :class:`TypeError` as of the 1.0 release. See the Changelog
for :ref:`api-changelog-0.7` for further details on how to upgrade
your code efficiently.
"""
type_ = self._coerce_enum(type_, structs.IQType)
del self._iq_request_map[type_, payload_cls]
self._logger.debug(
"iq request coroutine unregistered: type=%r, payload=%r",
type_, payload_cls)
[docs] def register_message_callback(self, type_, from_, cb):
"""
Register a callback to be called when a message is received.
:param type_: Message type to listen for, or :data:`None` for a
wildcard match.
:type type_: :class:`~.MessageType` or :data:`None`
:param from_: Sender JID to listen for, or :data:`None` for a wildcard
match.
:type from_: :class:`~aioxmpp.JID` or :data:`None`
:param cb: Callback function to call
:raises ValueError: if another function is already registered for the
same ``(type_, from_)`` pair.
:raises ValueError: if `type_` is not a valid
:class:`~.MessageType` (and cannot be cast
to a :class:`~.MessageType`)
`cb` will be called whenever a message stanza matching the `type_` and
`from_` is received, according to the wildcarding rules below. More
specific callbacks win over less specific callbacks, and the match on
the `from_` address takes precedence over the match on the `type_`.
To be explicit, the order in which callbacks are searched for a given
``type_`` and ``from_`` of a stanza is:
* ``type_``, ``from_``
* ``type_``, ``from_.bare()``
* ``None``, ``from_``
* ``None``, ``from_.bare()``
* ``type_``, ``None``
* ``None``, ``None``
.. note::
When the server sends a stanza without from attribute, it is
replaced with the bare :attr:`local_jid`, as per :rfc:`6120`.
In the future, there might be a different way to select those
stanzas.
.. versionchanged:: 0.7
The `type_` argument is now supposed to be a
:class:`~.MessageType` member.
.. deprecated:: 0.7
Passing a :class:`str` as `type_` argument is deprecated and will
raise a :class:`TypeError` as of the 1.0 release. See the Changelog
for :ref:`api-changelog-0.7` for further details on how to upgrade
your code efficiently.
"""
if type_ is not None:
type_ = self._coerce_enum(type_, structs.MessageType)
key = type_, from_
if key in self._message_map:
raise ValueError(
"only one listener is allowed per (type_, from_) pair"
)
self._message_map[key] = cb
self._logger.debug(
"message callback registered: type=%r, from=%r",
type_, from_)
[docs] def unregister_message_callback(self, type_, from_):
"""
Unregister a callback previously registered with
:meth:`register_message_callback`.
:param type_: Message type to listen for.
:type type_: :class:`~.MessageType` or :data:`None`
:param from_: Sender JID to listen for.
:type from_: :class:`~aioxmpp.JID` or :data:`None`
:raises KeyError: if no function is currently registered for the given
``(type_, from_)`` pair.
:raises ValueError: if `type_` is not a valid
:class:`~.MessageType` (and cannot be cast
to a :class:`~.MessageType`)
The match is made on the exact pair; it is not possible to unregister
arbitrary listeners by passing :data:`None` to both arguments (i.e. the
wildcarding only applies for receiving stanzas, not for unregistering
callbacks; unregistering the super-wildcard with both arguments set to
:data:`None` is of course possible).
.. versionchanged:: 0.7
The `type_` argument is now supposed to be a
:class:`~.MessageType` member.
.. deprecated:: 0.7
Passing a :class:`str` as `type_` argument is deprecated and will
raise a :class:`TypeError` as of the 1.0 release. See the Changelog
for :ref:`api-changelog-0.7` for further details on how to upgrade
your code efficiently.
"""
if type_ is not None:
type_ = self._coerce_enum(type_, structs.MessageType)
del self._message_map[type_, from_]
self._logger.debug(
"message callback unregistered: type=%r, from=%r",
type_, from_)
[docs] def register_presence_callback(self, type_, from_, cb):
"""
Register a callback to be called when a presence stanza is received.
:param type_: Presence type to listen for.
:type type_: :class:`~.PresenceType`
:param from_: Sender JID to listen for, or :data:`None` for a wildcard
match.
:type from_: :class:`~aioxmpp.JID` or :data:`None`.
:param cb: Callback function
:raises ValueError: if another listener with the same ``(type_,
from_)`` pair is already registered
:raises ValueError: if `type_` is not a valid
:class:`~.PresenceType` (and cannot be cast
to a :class:`~.PresenceType`)
`cb` will be called whenever a presence stanza matching the `type_` is
received from the specified sender. `from_` may be :data:`None` to
indicate a wildcard. Like with :meth:`register_message_callback`, more
specific callbacks win over less specific callbacks. The fallback order
is identical, except that the ``type_=None`` entries described there do
not apply for presence stanzas and are thus omitted.
.. versionchanged:: 0.7
The `type_` argument is now supposed to be a
:class:`~.PresenceType` member.
.. deprecated:: 0.7
Passing a :class:`str` as `type_` argument is deprecated and will
raise a :class:`TypeError` as of the 1.0 release. See the Changelog
for :ref:`api-changelog-0.7` for further details on how to upgrade
your code efficiently.
"""
type_ = self._coerce_enum(type_, structs.PresenceType)
key = type_, from_
if key in self._presence_map:
raise ValueError(
"only one listener is allowed per (type_, from_) pair"
)
self._presence_map[key] = cb
self._logger.debug(
"presence callback registered: type=%r, from=%r",
type_, from_)
[docs] def unregister_presence_callback(self, type_, from_):
"""
Unregister a callback previously registered with
:meth:`register_presence_callback`.
:param type_: Presence type to listen for.
:type type_: :class:`~.PresenceType`
:param from_: Sender JID to listen for, or :data:`None` for a wildcard
match.
:type from_: :class:`~aioxmpp.JID` or :data:`None`.
:raises KeyError: if no callback is currently registered for the given
``(type_, from_)`` pair
:raises ValueError: if `type_` is not a valid
:class:`~.PresenceType` (and cannot be cast
to a :class:`~.PresenceType`)
The match is made on the exact pair; it is not possible to unregister
arbitrary listeners by passing :data:`None` to the `from_` arguments
(i.e. the wildcarding only applies for receiving stanzas, not for
unregistering callbacks; unregistering a wildcard match with `from_`
set to :data:`None` is of course possible).
.. versionchanged:: 0.7
The `type_` argument is now supposed to be a
:class:`~.PresenceType` member.
.. deprecated:: 0.7
Passing a :class:`str` as `type_` argument is deprecated and will
raise a :class:`TypeError` as of the 1.0 release. See the Changelog
for :ref:`api-changelog-0.7` for further details on how to upgrade
your code efficiently.
"""
type_ = self._coerce_enum(type_, structs.PresenceType)
del self._presence_map[type_, from_]
self._logger.debug(
"presence callback unregistered: type=%r, from=%r",
type_, from_)
def _start_prepare(self, xmlstream, receiver):
self._xmlstream_failure_token = xmlstream.on_closing.connect(
self._xmlstream_failed
)
xmlstream.stanza_parser.add_class(stanza.IQ, receiver)
xmlstream.stanza_parser.add_class(stanza.Message, receiver)
xmlstream.stanza_parser.add_class(stanza.Presence, receiver)
xmlstream.error_handler = self.recv_erroneous_stanza
if self._sm_enabled:
self._logger.debug("using SM")
xmlstream.stanza_parser.add_class(nonza.SMAcknowledgement,
receiver)
xmlstream.stanza_parser.add_class(nonza.SMRequest,
receiver)
self._xmlstream_exception = None
def _start_rollback(self, xmlstream):
xmlstream.error_handler = None
xmlstream.stanza_parser.remove_class(stanza.Presence)
xmlstream.stanza_parser.remove_class(stanza.Message)
xmlstream.stanza_parser.remove_class(stanza.IQ)
if self._sm_enabled:
xmlstream.stanza_parser.remove_class(
nonza.SMRequest)
xmlstream.stanza_parser.remove_class(
nonza.SMAcknowledgement)
xmlstream.on_closing.disconnect(
self._xmlstream_failure_token
)
def _start_commit(self, xmlstream):
if not self._established:
self.on_stream_established()
self._established = True
self._task = asyncio.async(self._run(xmlstream), loop=self._loop)
self._task.add_done_callback(self._done_handler)
self._logger.debug("broker task started as %r", self._task)
self._next_ping_event_at = datetime.utcnow() + self.ping_interval
self._next_ping_event_type = PingEventType.SEND_OPPORTUNISTIC
self._ping_send_opportunistic = self._sm_enabled
[docs] def start(self, xmlstream):
"""
Start or resume the stanza stream on the given
:class:`aioxmpp.protocol.XMLStream` `xmlstream`.
This starts the main broker task, registers stanza classes at the
`xmlstream` and reconfigures the ping state.
"""
if self.running:
raise RuntimeError("already started")
self._start_prepare(xmlstream, self.recv_stanza)
self._closed = False
self._start_commit(xmlstream)
[docs] def stop(self):
"""
Send a signal to the main broker task to terminate. You have to check
:attr:`running` and possibly wait for it to become :data:`False` ---
the task takes at least one loop through the event loop to terminate.
It is guarenteed that the task will not attempt to send stanzas over
the existing `xmlstream` after a call to :meth:`stop` has been made.
It is legal to call :meth:`stop` even if the task is already
stopped. It is a no-op in that case.
"""
if not self.running:
return
self._logger.debug("sending stop signal to task")
self._task.cancel()
@asyncio.coroutine
[docs] def wait_stop(self):
"""
Stop the stream and wait for it to stop.
See :meth:`stop` for the general stopping conditions. You can assume
that :meth:`stop` is the first thing this coroutine calls.
"""
if not self.running:
return
self.stop()
try:
yield from self._task
except asyncio.CancelledError:
pass
@asyncio.coroutine
[docs] def close(self):
"""
Close the stream and the underlying XML stream (if any is connected).
This is essentially a way of saying "I do not want to use this stream
anymore" (until the next call to :meth:`start`). If the stream is
currently running, the XML stream is closed gracefully (potentially
sending an SM ack), the worker is stopped and any Stream Management
state is cleaned up.
If an error occurs while the stream stops, the error is ignored.
After the call to :meth:`close` has started, :meth:`on_failure` will
not be emitted, even if the XML stream fails before closure has
completed.
After a call to :meth:`close`, the stream is stopped, all SM state is
discarded and calls to :meth:`enqueue_stanza` raise a
:class:`DestructionRequested` ``"close() called"``. Such a
:class:`StanzaStream` can be re-started by calling :meth:`start`.
.. versionchanged:: 0.8
Before 0.8, an error during a call to :meth:`close` would stop the
stream from closing completely, and the exception was re-raised. If
SM was enabled, the state would have been kept, allowing for
resumption and ensuring that stanzas still enqueued or
unacknowledged would get a chance to be sent.
If you want to have guarantees that all stanzas sent up to a certain
point are sent, you should be using :meth:`send_and_wait_for_sent`
with stream management.
"""
exc = DestructionRequested("close() called")
if self.running:
if self.sm_enabled:
self._xmlstream.send_xso(nonza.SMAcknowledgement(
counter=self._sm_inbound_ctr
))
yield from self._xmlstream.close_and_wait() # does not raise
yield from self.wait_stop() # may raise
self._closed = True
self._xmlstream_exception = exc
self._destroy_stream_state(self._xmlstream_exception)
if self.sm_enabled:
self.stop_sm()
@asyncio.coroutine
def _run(self, xmlstream):
self._xmlstream = xmlstream
active_fut = asyncio.async(self._active_queue.get(),
loop=self._loop)
incoming_fut = asyncio.async(self._incoming_queue.get(),
loop=self._loop)
try:
while True:
timeout = self._next_ping_event_at - datetime.utcnow()
if timeout.total_seconds() < 0:
timeout = timedelta()
done, pending = yield from asyncio.wait(
[
active_fut,
incoming_fut,
],
return_when=asyncio.FIRST_COMPLETED,
timeout=timeout.total_seconds())
with (yield from self._broker_lock):
if active_fut in done:
self._process_outgoing(xmlstream, active_fut.result())
active_fut = asyncio.async(
self._active_queue.get(),
loop=self._loop)
if incoming_fut in done:
self._process_incoming(xmlstream,
incoming_fut.result())
incoming_fut = asyncio.async(
self._incoming_queue.get(),
loop=self._loop)
timeout = self._next_ping_event_at - datetime.utcnow()
if timeout.total_seconds() <= 0:
self._process_ping_event(xmlstream)
finally:
# make sure we rescue any stanzas which possibly have already been
# caught by the calls to get()
self._logger.debug("task terminating, rescuing stanzas and "
"clearing handlers")
if incoming_fut.done() and not incoming_fut.exception():
self._incoming_queue.putleft_nowait(incoming_fut.result())
else:
incoming_fut.cancel()
if active_fut.done() and not active_fut.exception():
self._active_queue.putleft_nowait(active_fut.result())
else:
active_fut.cancel()
# we also lock shutdown, because the main race is among the SM
# variables
with (yield from self._broker_lock):
if not self.sm_enabled or not self.sm_resumable:
self._destroy_stream_state(
self._xmlstream_exception or
DestructionRequested(
"close() or stop() called and stream is not "
"resumable"
)
)
if self.sm_enabled:
self._stop_sm()
self._start_rollback(xmlstream)
if self._xmlstream_exception:
raise self._xmlstream_exception
def recv_stanza(self, stanza):
"""
Inject a `stanza` into the incoming queue.
"""
self._incoming_queue.put_nowait((stanza, None))
def recv_erroneous_stanza(self, partial_obj, exc):
self._incoming_queue.put_nowait((partial_obj, exc))
[docs] def enqueue(self, stanza, **kwargs):
"""
Put a `stanza` in the internal transmission queue and return a token to
track it.
:param stanza: Stanza to send
:type stanza: :class:`IQ`, :class:`Message` or :class:`Presence`
:param kwargs: see :class:`StanzaToken`
:return: token which tracks the stanza
:rtype: :class:`StanzaToken`
The `stanza` is enqueued in the active queue for transmission and will
be sent on the next opportunity. The relative ordering of stanzas
enqueued is always preserved.
Return a fresh :class:`StanzaToken` instance which traks the progress
of the transmission of the `stanza`. The `kwargs` are forwarded to the
:class:`StanzaToken` constructor.
This method calls :meth:`~.stanza.StanzaBase.autoset_id` on the stanza
automatically.
.. seealso::
:meth:`send`
for a more high-level way to send stanzas.
"""
if self._closed:
raise self._xmlstream_exception
stanza.validate()
token = StanzaToken(stanza, **kwargs)
self._active_queue.put_nowait(token)
stanza.autoset_id()
self._logger.debug("enqueued stanza %r with token %r",
stanza, token)
return token
enqueue_stanza = enqueue
@property
[docs] def running(self):
"""
:data:`True` if the broker task is currently running, and :data:`False`
otherwise.
"""
return self._task is not None and not self._task.done()
@asyncio.coroutine
[docs] def start_sm(self, request_resumption=True):
"""
Start stream management (version 3). This negotiates stream management
with the server.
If the server rejects the attempt to enable stream management, a
:class:`.errors.StreamNegotiationFailure` is raised. The stream is
still running in that case.
.. warning::
This method cannot and does not check whether the server advertised
support for stream management. Attempting to negotiate stream
management without server support might lead to termination of the
stream.
If an XML stream error occurs during the negotiation, the result
depends on a few factors. In any case, the stream is not running
afterwards. If the :class:`SMEnabled` response was not received before
the XML stream died, SM is also disabled and the exception which caused
the stream to die is re-raised (this is due to the implementation of
:func:`~.protocol.send_and_wait_for`). If the :class:`SMEnabled`
response was received and annonuced support for resumption, SM is
enabled. Otherwise, it is disabled. No exception is raised if
:class:`SMEnabled` was received, as this method has no way to determine
that the stream failed.
If negotiation succeeds, this coroutine initializes a new stream
management session. The stream management state attributes become
available and :attr:`sm_enabled` becomes :data:`True`.
"""
if not self.running:
raise RuntimeError("cannot start Stream Management while"
" StanzaStream is not running")
if self.sm_enabled:
raise RuntimeError("Stream Management already enabled")
with (yield from self._broker_lock):
response = yield from protocol.send_and_wait_for(
self._xmlstream,
[
nonza.SMEnable(resume=bool(request_resumption)),
],
[
nonza.SMEnabled,
nonza.SMFailed
]
)
if isinstance(response, nonza.SMFailed):
raise errors.StreamNegotiationFailure(
"Server rejected SM request")
self._sm_outbound_base = 0
self._sm_inbound_ctr = 0
self._sm_unacked_list = []
self._sm_enabled = True
self._sm_id = response.id_
self._sm_resumable = response.resume
self._sm_max = response.max_
self._sm_location = response.location
self._ping_send_opportunistic = True
self._logger.info("SM started: resumable=%s, stream id=%r",
self._sm_resumable,
self._sm_id)
# if not self._xmlstream:
# # stream died in the meantime...
# if self._xmlstream_exception:
# raise self._xmlstream_exception
self._xmlstream.stanza_parser.add_class(
nonza.SMRequest,
self.recv_stanza)
self._xmlstream.stanza_parser.add_class(
nonza.SMAcknowledgement,
self.recv_stanza)
@property
[docs] def sm_enabled(self):
"""
:data:`True` if stream management is currently enabled on the stream,
:data:`False` otherwise.
"""
return self._sm_enabled
@property
[docs] def sm_outbound_base(self):
"""
The last value of the remote stanza counter.
.. note::
Accessing this attribute when :attr:`sm_enabled` is :data:`False`
raises :class:`RuntimeError`.
"""
if not self.sm_enabled:
raise RuntimeError("Stream Management not enabled")
return self._sm_outbound_base
@property
[docs] def sm_inbound_ctr(self):
"""
The current value of the inbound stanza counter.
.. note::
Accessing this attribute when :attr:`sm_enabled` is :data:`False`
raises :class:`RuntimeError`.
"""
if not self.sm_enabled:
raise RuntimeError("Stream Management not enabled")
return self._sm_inbound_ctr
@property
[docs] def sm_unacked_list(self):
"""
A **copy** of the list of stanza tokens which have not yet been acked
by the remote party.
.. note::
Accessing this attribute when :attr:`sm_enabled` is :data:`False`
raises :class:`RuntimeError`.
Accessing this attribute is expensive, as the list is copied. In
general, access to this attribute should not be neccessary at all.
"""
if not self.sm_enabled:
raise RuntimeError("Stream Management not enabled")
return self._sm_unacked_list[:]
@property
[docs] def sm_max(self):
"""
The value of the ``max`` attribute of the
:class:`~.nonza.SMEnabled` response from the server.
.. note::
Accessing this attribute when :attr:`sm_enabled` is :data:`False`
raises :class:`RuntimeError`.
"""
if not self.sm_enabled:
raise RuntimeError("Stream Management not enabled")
return self._sm_max
@property
[docs] def sm_location(self):
"""
The value of the ``location`` attribute of the
:class:`~.nonza.SMEnabled` response from the server.
.. note::
Accessing this attribute when :attr:`sm_enabled` is :data:`False`
raises :class:`RuntimeError`.
"""
if not self.sm_enabled:
raise RuntimeError("Stream Management not enabled")
return self._sm_location
@property
[docs] def sm_id(self):
"""
The value of the ``id`` attribute of the
:class:`~.nonza.SMEnabled` response from the server.
.. note::
Accessing this attribute when :attr:`sm_enabled` is :data:`False`
raises :class:`RuntimeError`.
"""
if not self.sm_enabled:
raise RuntimeError("Stream Management not enabled")
return self._sm_id
@property
[docs] def sm_resumable(self):
"""
The value of the ``resume`` attribute of the
:class:`~.nonza.SMEnabled` response from the server.
.. note::
Accessing this attribute when :attr:`sm_enabled` is :data:`False`
raises :class:`RuntimeError`.
"""
if not self.sm_enabled:
raise RuntimeError("Stream Management not enabled")
return self._sm_resumable
def _resume_sm(self, remote_ctr):
"""
Version of :meth:`resume_sm` which can be used during slow start.
"""
self._logger.info("resuming SM stream with remote_ctr=%d", remote_ctr)
# remove any acked stanzas
self.sm_ack(remote_ctr)
# reinsert the remaining stanzas
for token in self._sm_unacked_list:
self._active_queue.putleft_nowait(token)
self._sm_unacked_list.clear()
@asyncio.coroutine
[docs] def resume_sm(self, xmlstream):
"""
Resume an SM-enabled stream using the given `xmlstream`.
If the server rejects the attempt to resume stream management, a
:class:`.errors.StreamNegotiationFailure` is raised. The stream is then
in stopped state and stream management has been stopped.
.. warning::
This method cannot and does not check whether the server advertised
support for stream management. Attempting to negotiate stream
management without server support might lead to termination of the
stream.
If the XML stream dies at any point during the negotiation, the SM
state is left unchanged. If no response has been received yet, the
exception which caused the stream to die is re-raised. The state of the
stream depends on whether the main task already noticed the dead
stream.
If negotiation succeeds, this coroutine resumes the stream management
session and initiates the retransmission of any unacked stanzas. The
stream is then in running state.
"""
if self.running:
raise RuntimeError("Cannot resume Stream Management while"
" StanzaStream is running")
self._start_prepare(xmlstream, self.recv_stanza)
try:
response = yield from protocol.send_and_wait_for(
xmlstream,
[
nonza.SMResume(previd=self.sm_id,
counter=self._sm_inbound_ctr)
],
[
nonza.SMResumed,
nonza.SMFailed
]
)
if isinstance(response, nonza.SMFailed):
xmlstream.stanza_parser.remove_class(
nonza.SMRequest)
xmlstream.stanza_parser.remove_class(
nonza.SMAcknowledgement)
self.stop_sm()
raise errors.StreamNegotiationFailure(
"Server rejected SM resumption")
self._resume_sm(response.counter)
except:
self._start_rollback(xmlstream)
raise
self._start_commit(xmlstream)
def _stop_sm(self):
"""
Version of :meth:`stop_sm` which can be called during startup.
"""
if not self.sm_enabled:
raise RuntimeError("Stream Management is not enabled")
self._logger.info("stopping SM stream")
self._sm_enabled = False
del self._sm_outbound_base
del self._sm_inbound_ctr
for token in self._sm_unacked_list:
token._set_state(StanzaState.SENT_WITHOUT_SM)
del self._sm_unacked_list
self._destroy_stream_state(ConnectionError(
"stream management disabled"
))
[docs] def stop_sm(self):
"""
Disable stream management on the stream.
Attempting to call this method while the stream is running or without
stream management enabled results in a :class:`RuntimeError`.
Any sent stanzas which have not been acked by the remote yet are put
into :attr:`StanzaState.SENT_WITHOUT_SM` state.
"""
if self.running:
raise RuntimeError("Cannot stop Stream Management while"
" StanzaStream is running")
return self._stop_sm()
def sm_ack(self, remote_ctr):
"""
Process the remote stanza counter `remote_ctr`. Any acked stanzas are
dropped from :attr:`sm_unacked_list` and put into
:attr:`StanzaState.ACKED` state and the counters are increased
accordingly.
Attempting to call this without Stream Management enabled results in a
:class:`RuntimeError`.
"""
if not self._sm_enabled:
raise RuntimeError("Stream Management is not enabled")
self._logger.debug("sm_ack(%d)", remote_ctr)
to_drop = remote_ctr - self._sm_outbound_base
if to_drop < 0:
self._logger.warning(
"remote stanza counter is *less* than before "
"(outbound_base=%d, remote_ctr=%d)",
self._sm_outbound_base,
remote_ctr)
return
acked = self._sm_unacked_list[:to_drop]
del self._sm_unacked_list[:to_drop]
self._sm_outbound_base = remote_ctr
if acked:
self._logger.debug("%d stanzas acked by remote", len(acked))
for token in acked:
token._set_state(StanzaState.ACKED)
@asyncio.coroutine
[docs] def send_iq_and_wait_for_reply(self, iq, *,
timeout=None):
"""
Send an IQ stanza `iq` and wait for the response. If `timeout` is not
:data:`None`, it must be the time in seconds for which to wait for a
response.
If the response is a ``"result"`` IQ, the value of the
:attr:`~aioxmpp.IQ.payload` attribute is returned. Otherwise,
the exception generated from the :attr:`~aioxmpp.IQ.error`
attribute is raised.
.. seealso::
:meth:`register_iq_response_future` and
:meth:`send_and_wait_for_sent` for other cases raising exceptions.
.. deprecated:: 0.8
This method will be removed in 1.0. Use :meth:`send` instead.
.. versionchanged:: 0.8
On a timeout, :class:`TimeoutError` is now raised instead of
:class:`asyncio.TimeoutError`.
"""
warnings.warn(
r"send_iq_and_wait_for_reply is deprecated and will be removed in"
r" 1.0",
DeprecationWarning,
stacklevel=1,
)
return (yield from self.send(iq, timeout=timeout))
@asyncio.coroutine
[docs] def send_and_wait_for_sent(self, stanza):
"""
Send the given `stanza` over the given :class:`StanzaStream` `stream`.
.. deprecated:: 0.8
This method will be removed in 1.0. Use :meth:`send` instead.
"""
warnings.warn(
r"send_and_wait_for_sent is deprecated and will be removed in 1.0",
DeprecationWarning,
stacklevel=1,
)
yield from self.enqueue(stanza)
@asyncio.coroutine
[docs] def send(self, stanza, *, timeout=None):
"""
Send a stanza.
:param stanza: Stanza to send
:type stanza: :class:`~.IQ`, :class:`~.Presence` or :class:`~.Message`
:param timeout: Maximum time in seconds to wait for an IQ response, or
:data:`None` to disable the timeout.
:type timeout: :class:`~numbers.Real` or :data:`None`
:raise OSError: if the underlying XML stream fails and stream
management is not disabled.
:raise aioxmpp.stream.DestructionRequested:
if the stream is closed while sending the stanza or waiting for a
response.
:raise aioxmpp.errors.XMPPError: if an error IQ response is received
:raise aioxmpp.errors.ErroneousStanza: if the IQ response could not be
parsed
:return: IQ response :attr:`~.IQ.payload` or :data:`None`
Send the stanza and wait for it to be sent. If the stanza is an IQ
request, the response is awaited and the :attr:`~.IQ.payload` of the
response is returned.
The `timeout` as well as any of the exception cases referring to a
"response" do not apply for IQ response stanzas, message stanzas or
presence stanzas sent with this method, as this method only waits for
a reply if an IQ *request* stanza is being sent.
If `stanza` is an IQ request and the response is not received within
`timeout` seconds, :class:`TimeoutError` (not
:class:`asyncio.TimeoutError`!) is raised.
.. warning::
Setting a timeout is recommended for IQ requests. If the IQ is sent
directly to the clients server for processing (i.e. if the
:attr:`~.IQ.to` attribute is :data:`None`), malformed responses
are discarded instead of raising :class:`.errors.ErroneusStanza`.
This is due to limitations in the :mod:`aioxmpp.xso` code, which are
to be fixed at some point.
.. versionadded:: 0.8
"""
stanza.autoset_id()
self._logger.debug("sending %r and waiting for it to be sent",
stanza)
if not isinstance(stanza, stanza_.IQ) or stanza.type_.is_response:
yield from self.enqueue(stanza)
return
fut = asyncio.Future()
self.register_iq_response_future(
stanza.to,
stanza.id_,
fut,
)
try:
yield from self.enqueue(stanza)
except:
fut.cancel()
raise
if not timeout:
reply = yield from fut
else:
try:
reply = yield from asyncio.wait_for(
fut,
timeout=timeout
)
except asyncio.TimeoutError:
raise TimeoutError
return reply.payload
@contextlib.contextmanager
[docs]def iq_handler(stream, type_, payload_cls, coro):
"""
Context manager to temporarily register a coroutine to handle IQ requests
on a :class:`StanzaStream`.
:param stream: Stanza stream to register the coroutine at
:type stream: :class:`StanzaStream`
:param type_: IQ type to react to (must be a request type).
:type type_: :class:`~aioxmpp.IQType`
:param payload_cls: Payload class to react to (subclass of
:class:`~xso.XSO`)
:type payload_cls: :class:`~.XMLStreamClass`
:param coro: Coroutine to register
The coroutine is registered when the context is entered and unregistered
when the context is exited. Running coroutines are not affected by exiting
the context manager.
.. versionadded:: 0.8
"""
stream.register_iq_request_coro(
type_,
payload_cls,
coro,
)
try:
yield
finally:
stream.unregister_iq_request_coro(type_, payload_cls)
@contextlib.contextmanager
[docs]def message_handler(stream, type_, from_, cb):
"""
Context manager to temporarily register a callback to handle messages on a
:class:`StanzaStream`.
:param stream: Stanza stream to register the coroutine at
:type stream: :class:`StanzaStream`
:param type_: Message type to listen for, or :data:`None` for a wildcard
match.
:type type_: :class:`~.MessageType` or :data:`None`
:param from_: Sender JID to listen for, or :data:`None` for a wildcard
match.
:type from_: :class:`~aioxmpp.JID` or :data:`None`
:param cb: Callback to register
The callback is registered when the context is entered and unregistered
when the context is exited.
.. versionadded:: 0.8
"""
stream.register_message_callback(
type_,
from_,
cb,
)
try:
yield
finally:
stream.unregister_message_callback(
type_,
from_,
)
@contextlib.contextmanager
[docs]def presence_handler(stream, type_, from_, cb):
"""
Context manager to temporarily register a callback to handle presence
stanzas on a :class:`StanzaStream`.
:param stream: Stanza stream to register the coroutine at
:type stream: :class:`StanzaStream`
:param type_: Presence type to listen for.
:type type_: :class:`~.PresenceType`
:param from_: Sender JID to listen for, or :data:`None` for a wildcard
match.
:type from_: :class:`~aioxmpp.JID` or :data:`None`.
:param cb: Callback to register
The callback is registered when the context is entered and unregistered
when the context is exited.
.. versionadded:: 0.8
"""
stream.register_presence_callback(
type_,
from_,
cb,
)
try:
yield
finally:
stream.unregister_presence_callback(
type_,
from_,
)
_Undefined = object()
@contextlib.contextmanager
[docs]def stanza_filter(filter_, func, order=_Undefined):
"""
Context manager to temporarily register a filter function on a
:class:`Filter`.
:param filter_: Filter to register the function at
:type filter_: :class:`Filter`
:param func: Filter function to register
:param order: Order parameter to pass to :meth:`.Filter.register`
The type of `order` is specific to the :class:`Filter` instance used, see
the documentation of :meth:`Filter.register` and :meth:`AppFilter.register`
respectively.
The filter function is registered when the context is entered and
unregistered when the context is exited.
.. versionadded:: 0.8
"""
if order is _Undefined:
token = filter_.register(func)
else:
token = filter_.register(func, order)
try:
yield
finally:
filter_.unregister(token)