Merge branch 'async-interal-api-break-everything' into 'master'

Make the internal "api" async

See merge request poezio/slixmpp!128

docs/api/api.rst

@@ -0,0 +1,88 @@
+.. _internal-api:
+
+Internal "API"
+==============
+
+Slixmpp has a generic API registry that can be used by its plugins to allow
+access control, redefinition of behaviour, without having to inherit from the
+plugin or do more dark magic.
+
+The idea is that each api call can be replaced, most of them use a form
+of in-memory storage that can be, for example, replaced with database
+or file-based storaged.
+
+
+Each plugin is assigned an API proxy bound to itself, but only a few make use
+of it.
+
+See also :ref:`api-simple-tuto`.
+
+Description of a generic API call
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. code-block:: python
+
+    def get_toto(jid, node, ifrom, args):
+        return 'toto'
+
+    self.xmpp.plugin['xep_XXXX'].api.register(handler, 'get_toto')
+
+Each API call will receive 4 parameters (which can be ``None`` if data
+is not relevant to the operation), which are ``jid`` (``Optional[JID]``),
+``node`` (``Optional[str]``), ``ifrom`` (``Optional[JID]``), and ``args``
+(``Any``).
+
+- ``jid``, if relevant, represents the JID targeted by that operation
+- ``node``, if relevant is an arbitrary string, but was thought for, e.g.,
+  a pubsub or disco node.
+- ``ifrom``, if relevant, is the JID the event is coming from.
+- ``args`` is the event-specific data passed on by the plugin, often a dict
+  of arguments (can be None as well).
+
+.. note::
+    Since 1.8.0, API calls can be coroutines.
+
+
+Handler hierarchy
+~~~~~~~~~~~~~~~~~
+
+The ``self.api.register()`` signature is as follows:
+
+.. code-block:: python
+
+    def register(handler, op, jid=None, node=None, dedfault=False):
+        pass
+
+As you can see, :meth:`~.APIRegistry.register` takes an additional ctype
+parameter, but the :class:`~.APIWrapper` takes care of that for us (in most
+cases, it is the name of the XEP plugin, such as ``'xep_0XXX'``).
+
+When you register a handler, you register it for an ``op``, for **operation**.
+For example, ``get_vcard``.
+
+``handler`` and ``op`` are the only two required parameters (and in many cases,
+all you will ever need). You can, however, go further and register handlers
+for specific values of the ``jid`` and ``node`` parameters of the calls.
+
+The priority of the execution of handlers is as follows:
+
+- Check if a handler for both values of ``node`` and ``jid`` has been defined
+- If not found, check if a handler for this value of ``jid`` has been defined
+- If not found, check if a handler for this value of ``node`` has been defined
+- If still not found, get the global handler (no parameter registered)
+
+
+Raw documentation
+~~~~~~~~~~~~~~~~~
+
+This documentation is provided for reference, but :meth:`~.APIRegistry.register`
+should be all you need.
+
+
+.. module:: slixmpp.api
+
+.. autoclass:: APIRegistry
+    :members:
+
+.. autoclass:: APIWrapper
+

docs/api/index.rst

@@ -14,3 +14,4 @@ API Reference
     xmlstream/matcher
     xmlstream/xmlstream
     xmlstream/tostring
+    api

docs/api/plugins/xep_0012.rst

@@ -9,6 +9,44 @@ XEP-0012: Last Activity
     :exclude-members: session_bind, plugin_init, plugin_end
 
 
+.. _api-0012:
+
+Internal API methods
+--------------------
+
+This plugin uses an in-memory storage by default to keep track of the
+received and sent last activities.
+
+.. glossary::
+
+    get_last_activity
+        - **jid**: :class:`~.JID` of whom to retrieve the last activity
+        - **node**: unused
+        - **ifrom**: who the request is from (None = local)
+        - **args**: ``None`` or an :class:`~.Iq` that is requesting the
+        - **returns**
+          information.
+
+        Get the last activity of a JID from the storage.
+
+    set_last_activity
+        - **jid**: :class:`~.JID` of whom to set the last activity
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: A dict containing ``'seconds'`` and ``'status'``
+          ``{'seconds': Optional[int], 'status': Optional[str]}``
+
+        Set the last activity of a JID in the storage.
+
+    del_last_activity
+        - **jid**: :class:`~.JID` to delete from the storage
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: unused
+
+        Remove the last activity of a JID from the storage.
+
+
 Stanza elements
 ---------------
 

docs/api/plugins/xep_0027.rst

@@ -9,6 +9,50 @@ XEP-0027: Current Jabber OpenPGP Usage
     :exclude-members: session_bind, plugin_init, plugin_end
 
 
+.. _api-0027:
+
+Internal API methods
+--------------------
+
+The default API here is managing a JID→Keyid dict in-memory.
+
+.. glossary::
+
+    get_keyid
+        - **jid**: :class:`~.JID` to get.
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: unused
+        - **returns**: ``Optional[str]``, the keyid or None
+
+        Get the KeyiD for a JID, None if it is not found.
+
+    set_keyid
+        - **jid**: :class:`~.JID` to set the id for.
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: ``str``, keyid to set
+
+        Set the KeyiD for a JID.
+
+    del_keyid
+        - **jid**: :class:`~.JID` to delete from the mapping.
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: unused
+
+        Delete the KeyiD for a JID.
+
+    get_keyids
+        - **jid**: unused
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: unused
+        - **returns**: ``Dict[JID, str]`` the full internal mapping
+
+        Get all currently stored KeyIDs.
+
+
 Stanza elements
 ---------------
 

docs/api/plugins/xep_0047.rst

@@ -8,11 +8,78 @@ XEP-0047: In-band Bytestreams
     :members:
     :exclude-members: session_bind, plugin_init, plugin_end
 
-.. module:: slixmpp.plugins.xep_0047
-
 .. autoclass:: IBBytestream
     :members:
 
+.. _api-0047:
+
+Internal API methods
+--------------------
+
+The API here is used to manage streams and authorize. The default handlers
+work with the config parameters.
+
+.. glossary::
+
+    authorized_sid (0047 version)
+        - **jid**: :class:`~.JID` receiving the stream initiation.
+        - **node**:  stream id
+        - **ifrom**: who the stream is from.
+        - **args**: :class:`~.Iq` of the stream request.
+        - **returns**: ``True`` if the stream should be accepted,
+          ``False`` otherwise.
+
+        Check if the stream should be accepted. Uses
+        the information setup by :term:`preauthorize_sid (0047 version)`
+        by default.
+
+    authorized (0047 version)
+        - **jid**: :class:`~.JID` receiving the stream initiation.
+        - **node**:  stream id
+        - **ifrom**: who the stream is from.
+        - **args**: :class:`~.Iq` of the stream request.
+        - **returns**: ``True`` if the stream should be accepted,
+          ``False`` otherwise.
+
+        A fallback handler (run after :term:`authorized_sid (0047 version)`)
+        to check if a stream should be accepted. Uses the ``auto_accept``
+        parameter by default.
+
+    preauthorize_sid (0047 version)
+        - **jid**: :class:`~.JID` receiving the stream initiation.
+        - **node**:  stream id
+        - **ifrom**: who the stream will be from.
+        - **args**: Unused.
+
+        Register a stream id to be accepted automatically (called from
+        other plugins such as XEP-0095).
+
+    get_stream
+        - **jid**: :class:`~.JID` of local receiver.
+        - **node**:  stream id
+        - **ifrom**: who the stream is from.
+        - **args**: unused
+        - **returns**: :class:`~.IBBytestream`
+
+        Return a currently opened stream between two JIDs.
+
+    set_stream
+        - **jid**: :class:`~.JID` of local receiver.
+        - **node**:  stream id
+        - **ifrom**: who the stream is from.
+        - **args**: unused
+
+        Register an opened stream between two JIDs.
+
+    del_stream
+        - **jid**: :class:`~.JID` of local receiver.
+        - **node**:  stream id
+        - **ifrom**: who the stream is from.
+        - **args**: unused
+
+        Delete a stream between two JIDs.
+
+
 Stanza elements
 ---------------
 

docs/api/plugins/xep_0054.rst

@@ -8,6 +8,40 @@ XEP-0054: vcard-temp
     :members:
     :exclude-members: session_bind, plugin_init, plugin_end
 
+.. _api-0054:
+
+Internal API methods
+--------------------
+
+This plugin maintains by default an in-memory cache of the received
+VCards.
+
+.. glossary::
+
+    set_vcard
+        - **jid**: :class:`~.JID` of whom to set the vcard
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: :class:`~.VCardTemp` object to store for this JID.
+
+        Set a VCard for a JID.
+
+    get_vcard
+        - **jid**: :class:`~.JID` of whom to set the vcard
+        - **node**: unused
+        - **ifrom**: :class:`~.JID` the request is coming from
+        - **args**: unused
+        - **returns**: :class:`~.VCardTemp` object for this JID or None.
+
+        Get a stored VCard for a JID.
+
+    del_vcard
+        - **jid**: :class:`~.JID` of whom to set the vcard
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: unused
+
+        Delete a stored VCard for a JID.
 
 Stanza elements
 ---------------

docs/api/plugins/xep_0065.rst

@@ -8,6 +8,48 @@ XEP-0065: SOCKS5 Bytestreams
     :members:
     :exclude-members: session_bind, plugin_init, plugin_end
 
+.. _api-0065:
+
+Internal API methods
+--------------------
+
+The internal API is used here to authorize or pre-authorize streams.
+
+.. glossary::
+
+    authorized_sid (0065 version)
+        - **jid**: :class:`~.JID` receiving the stream initiation.
+        - **node**:  stream id
+        - **ifrom**: who the stream is from.
+        - **args**: :class:`~.Iq` of the stream request.
+        - **returns**: ``True`` if the stream should be accepted,
+          ``False`` otherwise.
+
+        Check if the stream should be accepted. Uses
+        the information setup by :term:`preauthorize_sid (0065 version)`
+        by default.
+
+    authorized (0065 version)
+        - **jid**: :class:`~.JID` receiving the stream initiation.
+        - **node**:  stream id
+        - **ifrom**: who the stream is from.
+        - **args**: :class:`~.Iq` of the stream request.
+        - **returns**: ``True`` if the stream should be accepted,
+          ``False`` otherwise.
+
+        A fallback handler (run after :term:`authorized_sid (0065 version)`)
+        to check if a stream should be accepted. Uses the ``auto_accept``
+        parameter by default.
+
+    preauthorize_sid (0065 version)
+        - **jid**: :class:`~.JID` receiving the stream initiation.
+        - **node**:  stream id
+        - **ifrom**: who the stream will be from.
+        - **args**: Unused.
+
+        Register a stream id to be accepted automatically (called from
+        other plugins such as XEP-0095).
+
 
 Stanza elements
 ---------------

docs/api/plugins/xep_0077.rst

@@ -8,6 +8,53 @@ XEP-0077: In-Band Registration
     :members:
     :exclude-members: session_bind, plugin_init, plugin_end
 
+Internal APi methods
+--------------------
+
+The API here is made to allow components to manage registered users.
+The default handlers make use of the plugin options and store users
+in memory.
+
+.. glossary::
+
+    user_get
+        - **jid**: unused
+        - **node**: unused
+        - **ifrom**: who the request is coming from
+        - **args**: :class:`~.Iq` registration request.
+        - **returns**: ``dict`` containing user data or None.
+
+        Get user data for a user.
+
+    user_validate
+        - **jid**: unused
+        - **node**: unused
+        - **ifrom**: who the request is coming from
+        - **args**: :class:`~.Iq` registration request, 'register' payload.
+        - **raises**: ValueError if some fields are invalid
+
+        Validate form fields and save user data.
+
+    user_remove
+        - **jid**: unused
+        - **node**: unused
+        - **ifrom**: who the request is coming from
+        - **args**: :class:`~.Iq` registration removal request.
+        - **raises**: KeyError if the user is not found.
+
+        Remove a user from the store.
+
+    make_registration_form
+        - **jid**: unused
+        - **node**: unused
+        - **ifrom**: who the request is coming from
+        - **args**: :class:`~.Iq` registration request.
+        - **raises**: KeyError if the user is not found.
+
+        Return an :class:`~.Iq` reply for the request, with a form and
+        options set. By default, use ``form_fields`` and ``form_instructions``
+        plugin config options.
+
 
 Stanza elements
 ---------------

docs/api/plugins/xep_0115.rst

@@ -8,6 +8,54 @@ XEP-0115: Entity Capabilities
     :members:
     :exclude-members: session_bind, plugin_init, plugin_end
 
+.. _api-0115:
+
+Internal API methods
+--------------------
+
+This internal API extends the Disco internal API, and also manages an
+in-memory cache of verstring→disco info, and fulljid→verstring.
+
+.. glossary::
+
+    cache_caps
+        - **jid**: unused
+        - **node**:  unused
+        - **ifrom**: unused
+        - **args**: a ``dict`` containing the verstring and
+          :class:`~.DiscoInfo` payload (
+          ``{'verstring': Optional[str], 'info': Optional[DiscoInfo]}``)
+
+        Cache a verification string with its payload.
+
+    get_caps
+        - **jid**: JID to retrieve the verstring for (unused with the default
+          handler)
+        - **node**:  unused
+        - **ifrom**: unused
+        - **args**: a ``dict`` containing the verstring
+          ``{'verstring': str}``
+        - **returns**: The :class:`~.DiscoInfo` payload for that verstring.
+
+        Get a disco payload from a verstring.
+
+    assign_verstring
+        - **jid**: :class:`~.JID` (full) to assign the verstring to
+        - **node**:  unused
+        - **ifrom**: unused
+        - **args**: a ``dict`` containing the verstring
+          ``{'verstring': str}``
+
+        Cache JID→verstring information.
+
+    get_verstring
+        - **jid**: :class:`~.JID` to use for fetching the verstring
+        - **node**:  unused
+        - **ifrom**: unused
+        - **args**: unused
+        - **returns**: ``str``, the verstring
+
+        Retrieve a verstring for a JID.
 
 Stanza elements
 ---------------

docs/api/plugins/xep_0128.rst

@@ -7,3 +7,38 @@ XEP-0128: Service Discovery Extensions
 .. autoclass:: XEP_0128
     :members:
     :exclude-members: session_bind, plugin_init, plugin_end
+
+.. _api-0128:
+
+Internal API methods
+--------------------
+
+
+
+.. glossary::
+
+    add_extended_info
+        - **jid**: JID to set the extended info for
+        - **node**:  note to set the info at
+        - **ifrom**: unused
+        - **args**: A :class:`~.Form` or list of forms to add to the disco
+          extended info for this JID/node.
+
+        Add extended info for a JID/node.
+
+    set_extended_info
+        - **jid**: JID to set the extended info for
+        - **node**:  note to set the info at
+        - **ifrom**: unused
+        - **args**: A :class:`~.Form` or list of forms to set as the disco
+          extended info for this JID/node.
+
+        Set extended info for a JID/node.
+
+    del_extended_info
+        - **jid**: JID to delete the extended info from
+        - **node**:  note to delete the info from
+        - **ifrom**: unused
+        - **args**: unused
+
+        Delete extended info for a JID/node.

docs/api/plugins/xep_0153.rst

@@ -8,6 +8,43 @@ XEP-0153: vCard-Based Avatars
     :members:
     :exclude-members: session_bind, plugin_init, plugin_end
 
+.. _api-0153:
+
+Internal API methods
+--------------------
+
+The internal API is used here to maintain an in-memory JID→avatar hash
+cache.
+
+.. glossary::
+
+    set_hash
+        - **jid**: :class:`~.JID` of whom to retrieve the last activity
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: ``str``, avatar hash
+
+        Set the avatar hash for a JID.
+
+    reset_hash
+        - **jid**: :class:`~.JID` of whom to retrieve the last activity
+        - **node**: unused
+        - **ifrom**: :class:`~.JID` of the entity requesting the reset.
+        - **args**: unused
+        - **returns**
+          information.
+
+        Reset the avatar hash for a JID. This downloads the vcard and computes
+        the hash.
+
+    get_hash
+        - **jid**: :class:`~.JID` of whom to retrieve the last activity
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: unused
+        - **returns**: ``Optional[str]``, the avatar hash
+
+        Get the avatar hash for a JID.
 
 Stanza elements
 ---------------

docs/api/plugins/xep_0231.rst

@@ -8,6 +8,41 @@ XEP-0231: Bits of Binary
     :members:
     :exclude-members: session_bind, plugin_init, plugin_end
 
+.. _api-0231:
+
+Internal API methods
+--------------------
+
+The default API handlers for this plugin manage an in-memory cache of
+bits of binary by content-id.
+
+.. glossary::
+
+    set_bob
+        - **jid**: :class:`~.JID` sending the bob
+        - **node**: unused
+        - **ifrom**: :class:`~JID` receiving the bob
+        - **args**: :class:`~.BitsOfBinary` element.
+
+        Set a BoB in the cache.
+
+    get_bob
+        - **jid**: :class:`~.JID` receiving the bob
+        - **node**: unused
+        - **ifrom**: :class:`~JID` sending the bob
+        - **args**: ``str`` content-id of the bob
+        - **returns**: :class:`~.BitsOfBinary` element.
+
+        Get a BoB from the cache.
+
+    del_bob
+        - **jid**: unused
+        - **node**: unused
+        - **ifrom**: :class:`~JID` sending the bob
+        - **args**: ``str`` content-id of the bob
+
+        Delete a BoB from the cache.
+
 
 Stanza elements
 ---------------

docs/api/plugins/xep_0319.rst

@@ -9,6 +9,33 @@ XEP-0319: Last User Interaction in Presence
     :exclude-members: session_bind, plugin_init, plugin_end
 
 
+.. _api-0319:
+
+Internal API methods
+--------------------
+
+The default API manages an in-memory cache of idle periods.
+
+.. glossary::
+
+    set_idle
+        - **jid**: :class:`~.JID` who has been idling
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: :class:`datetime`, timestamp of the idle start
+
+        Set the idle start for a JID.
+
+    get_idle
+        - **jid**: :class:`~.JID` to get the idle time of
+        - **node**: unused
+        - **ifrom**: unused
+        - **args**: : unused
+        - **returns**: :class:`datetime`
+
+        Get the idle start timestamp for a JID.
+
+
 Stanza elements
 ---------------
 

docs/howto/index.rst

@@ -6,6 +6,7 @@ Tutorials, FAQs, and How To Guides
 
     stanzas
     create_plugin
+    internal_api
     features
     sasl
     handlersmatchers

docs/howto/internal_api.rst

@@ -0,0 +1,94 @@
+.. _api-simple-tuto:
+
+Flexible internal API usage
+===========================
+
+The :ref:`internal-api` in slixmpp is used to override behavior or simply
+to override the default, in-memory storage backend with something persistent.
+
+
+We will use the XEP-0231 (Bits of Binary) plugin as an example here to show
+very basic functionality. Its API reference is in the plugin documentation:
+:ref:`api-0231`.
+
+Let us assume we want to keep each bit of binary in a file named with its
+content-id, with all metadata.
+
+First, we have to load the plugin:
+
+.. code-block:: python
+
+    from slixmpp import ClientXMPP
+    xmpp = ClientXMPP(...)
+    xmpp.register_plugin('xep_0231')
+
+This enables the default, in-memory storage.
+
+We have 3 methods to override to provide similar functionality and keep things
+coherent.
+
+Here is a class implementing very basic file storage for BoB:
+
+.. code-block:: python
+
+    from slixmpp.plugins.xep_0231 import BitsOfBinary
+    from os import makedirs, remove
+    from os.path import join, exists
+    import base64
+    import json
+
+    class BobLoader:
+        def __init__(self, directory):
+            makedirs(directory, exist_ok=True)
+            self.dir = directory
+
+        def set_bob(self, jid=None, node=None, ifrom=None, args=None):
+            payload = {
+                'data': base64.b64encode(args['data']).decode(),
+                'type': args['type'],
+                'cid': args['cid'],
+                'max_age': args['max_age']
+            }
+            with open(join(self.dir, args['cid']), 'w') as fd:
+                fd.write(json.dumps(payload))
+
+        def get_bob(self, jid=None, node=None, ifrom=None, args=None):
+            with open(join(self.dir, args), 'r') as fd:
+                payload = json.loads(fd.read())
+            bob = BitsOfBinary()
+            bob['data'] = base64.b64decode(payload['data'])
+            bob['type'] = payload['type']
+            bob['max_age'] = payload['max_age']
+            bob['cid'] = payload['cid']
+            return bob
+
+        def del_bob(self, jid=None, node=None, ifrom=None, args=None):
+            path = join(self.dir, args)
+            if exists(path):
+                remove(path)
+
+Now we need to replace the default handler with ours:
+
+.. code-block:: python
+
+
+    bobhandler = BobLoader('/tmp/bobcache')
+    xmpp.plugin['xep_0231'].api.register(bobhandler.set_bob, 'set_bob')
+    xmpp.plugin['xep_0231'].api.register(bobhandler.get_bob, 'get_bob')
+    xmpp.plugin['xep_0231'].api.register(bobhandler.del_bob, 'del_bob')
+
+
+And that’s it, the BoB storage is now made of JSON files living in a
+directory (``/tmp/bobcache`` here).
+
+
+To check that everything works, you can do the following:
+
+.. code-block:: python
+
+    cid = await xmpp.plugin['xep_0231'].set_bob(b'coucou', 'text/plain')
+    # A new bob file should appear
+    content = await xmpp.plugin['xep_0231'].get_bob(cid=cid)
+    assert content['bob']['data'] == b'coucou'
+
+A file should have been created in that directory.

itests/test_bob.py

@@ -20,7 +20,7 @@ class TestBOB(SlixIntegration):
 
     async def test_bob(self):
         """Check we can send and receive a BOB."""
-        cid = self.clients[0]['xep_0231'].set_bob(
+        cid = await self.clients[0]['xep_0231'].set_bob(
             self.data,
             'image/jpeg',
         )

itests/test_last_activity.py

@@ -18,7 +18,7 @@ class TestLastActivity(SlixIntegration):
 
     async def test_activity(self):
         """Check we can set and get last activity"""
-        self.clients[0]['xep_0012'].set_last_activity(
+        await self.clients[0]['xep_0012'].set_last_activity(
             status='coucou',
             seconds=4242,
         )

slixmpp/api.py

@@ -1,7 +1,19 @@
+from typing import Any, Optional, Callable
+from asyncio import iscoroutinefunction, Future
 from slixmpp.xmlstream import JID
 
+APIHandler = Callable[
+    [Optional[JID], Optional[str], Optional[JID], Any],
+    Any
+]
 
 class APIWrapper(object):
+    """Slixmpp API wrapper.
+
+    This class provide a shortened binding to access ``self.api`` from
+    plugins without having to specify the plugin name or the global
+    :class:`~.APIRegistry`.
+    """
 
     def __init__(self, api, name):
         self.api = api
@@ -37,6 +49,11 @@ class APIWrapper(object):
 
 
 class APIRegistry(object):
+    """API Registry.
+
+    This class is the global Slixmpp API registry, on which any handler will
+    be registed.
+    """
 
     def __init__(self, xmpp):
         self._handlers = {}
@@ -44,11 +61,11 @@ class APIRegistry(object):
         self.xmpp = xmpp
         self.settings = {}
 
-    def _setup(self, ctype, op):
+    def _setup(self, ctype: str, op: str):
         """Initialize the API callback dictionaries.
 
-        :param string ctype: The name of the API to initialize.
+        :param ctype: The name of the API to initialize.
-        :param string op: The API operation to initialize.
+        :param op: The API operation to initialize.
         """
         if ctype not in self.settings:
             self.settings[ctype] = {}
@@ -61,27 +78,32 @@ class APIRegistry(object):
                                          'jid': {},
                                          'node': {}}
 
-    def wrap(self, ctype):
+    def wrap(self, ctype: str) -> APIWrapper:
         """Return a wrapper object that targets a specific API."""
         return APIWrapper(self, ctype)
 
-    def purge(self, ctype):
+    def purge(self, ctype: str):
         """Remove all information for a given API."""
         del self.settings[ctype]
         del self._handler_defaults[ctype]
         del self._handlers[ctype]
 
-    def run(self, ctype, op, jid=None, node=None, ifrom=None, args=None):
+    def run(self, ctype: str, op: str, jid: Optional[JID] = None,
+            node: Optional[str] = None, ifrom: Optional[JID] = None,
+            args: Any = None) -> Future:
         """Execute an API callback, based on specificity.
 
         The API callback that is executed is chosen based on the combination
         of the provided JID and node:
 
-        JID   | node  | Handler
+        ====== ======= ===================
-        ==============================
+        JID     node    Handler
-        Given | Given | Node handler
+        ====== ======= ===================
-        Given | None  | JID handler
+        Given   Given   Node + JID handler
-        None  | None  | Global handler
+        Given   None    JID handler
+        None    Given   Node handler
+        None    None    Global handler
+        ====== ======= ===================
 
         A node handler is responsible for servicing a single node at a single
         JID, while a JID handler may respond for any node at a given JID, and
@@ -90,12 +112,16 @@ class APIRegistry(object):
         Handlers should check that the JID ``ifrom`` is authorized to perform
         the desired action.
 
-        :param string ctype: The name of the API to use.
+        .. versionchanged:: 1.8.0
-        :param string op: The API operation to perform.
+            ``run()`` always returns a future, if the handler is a coroutine
-        :param JID jid: Optionally provide specific JID.
+            the future should be awaited on.
-        :param string node: Optionally provide specific node.
+
-        :param JID ifrom: Optionally provide the requesting JID.
+        :param ctype: The name of the API to use.
-        :param tuple args: Optional positional arguments to the handler.
+        :param op: The API operation to perform.
+        :param jid: Optionally provide specific JID.
+        :param node: Optionally provide specific node.
+        :param ifrom: Optionally provide the requesting JID.
+        :param args: Optional arguments to the handler.
         """
         self._setup(ctype, op)
 
@@ -130,24 +156,32 @@ class APIRegistry(object):
 
         if handler:
             try:
-                return handler(jid, node, ifrom, args)
+                if iscoroutinefunction(handler):
+                    return self.xmpp.wrap(handler(jid, node, ifrom, args))
+                else:
+                    future: Future = Future()
+                    result = handler(jid, node, ifrom, args)
+                    future.set_result(result)
+                    return future
             except TypeError:
                 # To preserve backward compatibility, drop the ifrom
                 # parameter for existing handlers that don't understand it.
                 return handler(jid, node, args)
 
-    def register(self, handler, ctype, op, jid=None, node=None, default=False):
+    def register(self, handler: APIHandler, ctype: str, op: str,
+                 jid: Optional[JID] = None, node: Optional[str] = None,
+                 default: bool = False):
         """Register an API callback, with JID+node specificity.
 
         The API callback can later be executed based on the
         specificity of the provided JID+node combination.
 
-        See :meth:`~ApiRegistry.run` for more details.
+        See :meth:`~.APIRegistry.run` for more details.
 
-        :param string ctype: The name of the API to use.
+        :param ctype: The name of the API to use.
-        :param string op: The API operation to perform.
+        :param op: The API operation to perform.
-        :param JID jid: Optionally provide specific JID.
+        :param jid: Optionally provide specific JID.
-        :param string node: Optionally provide specific node.
+        :param node: Optionally provide specific node.
         """
         self._setup(ctype, op)
         if jid is None and node is None:
@@ -162,17 +196,18 @@ class APIRegistry(object):
         if default:
             self.register_default(handler, ctype, op)
 
-    def register_default(self, handler, ctype, op):
+    def register_default(self, handler, ctype: str, op: str):
         """Register a default, global handler for an operation.
 
-        :param func handler: The default, global handler for the operation.
+        :param handler: The default, global handler for the operation.
-        :param string ctype: The name of the API to modify.
+        :param ctype: The name of the API to modify.
-        :param string op: The API operation to use.
+        :param op: The API operation to use.
         """
         self._setup(ctype, op)
         self._handler_defaults[ctype][op] = handler
 
-    def unregister(self, ctype, op, jid=None, node=None):
+    def unregister(self, ctype: str, op: str, jid: Optional[JID] = None,
+                   node: Optional[str] = None):
         """Remove an API callback.
 
         The API callback chosen for removal is based on the
@@ -180,21 +215,22 @@ class APIRegistry(object):
 
         See :meth:`~ApiRegistry.run` for more details.
 
-        :param string ctype: The name of the API to use.
+        :param ctype: The name of the API to use.
-        :param string op: The API operation to perform.
+        :param op: The API operation to perform.
-        :param JID jid: Optionally provide specific JID.
+        :param jid: Optionally provide specific JID.
-        :param string node: Optionally provide specific node.
+        :param node: Optionally provide specific node.
         """
         self._setup(ctype, op)
         self.register(None, ctype, op, jid, node)
 
-    def restore_default(self, ctype, op, jid=None, node=None):
+    def restore_default(self, ctype: str, op: str, jid: Optional[JID] = None,
+                        node: Optional[str] = None):
         """Reset an API callback to use a default handler.
 
-        :param string ctype: The name of the API to use.
+        :param ctype: The name of the API to use.
-        :param string op: The API operation to perform.
+        :param op: The API operation to perform.
-        :param JID jid: Optionally provide specific JID.
+        :param jid: Optionally provide specific JID.
-        :param string node: Optionally provide specific node.
+        :param node: Optionally provide specific node.
         """
         self.unregister(ctype, op, jid, node)
         self.register(self._handler_defaults[ctype][op], ctype, op, jid, node)

slixmpp/plugins/xep_0012/last_activity.py

@@ -16,7 +16,7 @@ from slixmpp import future_wrapper, JID
 from slixmpp.stanza import Iq
 from slixmpp.exceptions import XMPPError
 from slixmpp.xmlstream import JID, register_stanza_plugin
-from slixmpp.xmlstream.handler import Callback
+from slixmpp.xmlstream.handler import CoroutineCallback
 from slixmpp.xmlstream.matcher import StanzaPath
 from slixmpp.plugins.xep_0012 import stanza, LastActivity
 
@@ -41,7 +41,7 @@ class XEP_0012(BasePlugin):
         self._last_activities = {}
 
         self.xmpp.register_handler(
-            Callback('Last Activity',
+            CoroutineCallback('Last Activity',
                  StanzaPath('iq@type=get/last_activity'),
                  self._handle_get_last_activity))
 
@@ -62,28 +62,50 @@ class XEP_0012(BasePlugin):
     def session_bind(self, jid):
         self.xmpp['xep_0030'].add_feature('jabber:iq:last')
 
-    def begin_idle(self, jid: Optional[JID] = None, status: str = None):
+    def begin_idle(self, jid: Optional[JID] = None, status: Optional[str] = None) -> Future:
         """Reset the last activity for the given JID.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param status: Optional status.
         """
-        self.set_last_activity(jid, 0, status)
+        return self.set_last_activity(jid, 0, status)
 
-    def end_idle(self, jid=None):
+    def end_idle(self, jid: Optional[JID] = None) -> Future:
-        self.del_last_activity(jid)
+        """Remove the last activity of a JID.
 
-    def start_uptime(self, status=None):
+        .. versionchanged:: 1.8.0
-        self.set_last_activity(None, 0, status)
+            This function now returns a Future.
+        """
+        return self.del_last_activity(jid)
 
-    def set_last_activity(self, jid=None, seconds=None, status=None):
+    def start_uptime(self, status: Optional[str] = None) -> Future:
-        self.api['set_last_activity'](jid, args={
+        """
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
+        return self.set_last_activity(None, 0, status)
+
+    def set_last_activity(self, jid=None, seconds=None, status=None) -> Future:
+        """Set last activity for a JID.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
+        return self.api['set_last_activity'](jid, args={
             'seconds': seconds,
-            'status': status})
+            'status': status
+        })
 
-    def del_last_activity(self, jid):
+    def del_last_activity(self, jid: JID) -> Future:
-        self.api['del_last_activity'](jid)
+        """Remove the last activity of a JID.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
+        return self.api['del_last_activity'](jid)
 
-    @future_wrapper
     def get_last_activity(self, jid: JID, local: bool = False,
                           ifrom: Optional[JID] = None, **iqkwargs) -> Future:
         """Get last activity for a specific JID.
@@ -109,10 +131,10 @@ class XEP_0012(BasePlugin):
         iq.enable('last_activity')
         return iq.send(**iqkwargs)
 
-    def _handle_get_last_activity(self, iq: Iq):
+    async def _handle_get_last_activity(self, iq: Iq):
         log.debug("Received last activity query from " + \
                   "<%s> to <%s>.", iq['from'], iq['to'])
-        reply = self.api['get_last_activity'](iq['to'], None, iq['from'], iq)
+        reply = await self.api['get_last_activity'](iq['to'], None, iq['from'], iq)
         reply.send()
 
     # =================================================================

slixmpp/plugins/xep_0027/gpg.py

@@ -5,9 +5,11 @@
 # See the file LICENSE for copying permission.
 from slixmpp.thirdparty import GPG
 
+from asyncio import Future
+
 from slixmpp.stanza import Presence, Message
-from slixmpp.plugins.base import BasePlugin, register_plugin
+from slixmpp.plugins.base import BasePlugin
-from slixmpp.xmlstream import ElementBase, register_stanza_plugin
+from slixmpp.xmlstream import register_stanza_plugin
 from slixmpp.xmlstream.handler import Callback
 from slixmpp.xmlstream.matcher import StanzaPath
 from slixmpp.plugins.xep_0027 import stanza, Signed, Encrypted
@@ -32,6 +34,9 @@ def _extract_data(data, kind):
 
 
 class XEP_0027(BasePlugin):
+    """
+    XEP-0027: Current Jabber OpenPGP Usage
+    """
 
     name = 'xep_0027'
     description = 'XEP-0027: Current Jabber OpenPGP Usage'
@@ -122,16 +127,36 @@ class XEP_0027(BasePlugin):
         v = self.gpg.verify(template % (data, sig))
         return v
 
-    def set_keyid(self, jid=None, keyid=None):
+    def set_keyid(self, jid=None, keyid=None) -> Future:
-        self.api['set_keyid'](jid, args=keyid)
+        """Set a keyid for a specific JID.
 
-    def get_keyid(self, jid=None):
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
+        return self.api['set_keyid'](jid, args=keyid)
+
+    def get_keyid(self, jid=None) -> Future:
+        """Get a keyid for a jid.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
         return self.api['get_keyid'](jid)
 
-    def del_keyid(self, jid=None):
+    def del_keyid(self, jid=None) -> Future:
-        self.api['del_keyid'](jid)
+        """Delete a keyid.
 
-    def get_keyids(self):
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
+        return self.api['del_keyid'](jid)
+
+    def get_keyids(self) -> Future:
+        """Get stored keyids.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
         return self.api['get_keyids']()
 
     def _handle_signed_presence(self, pres):

slixmpp/plugins/xep_0030/disco.py

@@ -6,13 +6,13 @@
 import asyncio
 import logging
 
-
+from asyncio import Future
 from typing import Optional, Callable
 
 from slixmpp import Iq
 from slixmpp import future_wrapper
 from slixmpp.plugins import BasePlugin
-from slixmpp.xmlstream.handler import Callback
+from slixmpp.xmlstream.handler import Callback, CoroutineCallback
 from slixmpp.xmlstream.matcher import StanzaPath
 from slixmpp.xmlstream import register_stanza_plugin, JID
 from slixmpp.plugins.xep_0030 import stanza, DiscoInfo, DiscoItems
@@ -91,12 +91,12 @@ class XEP_0030(BasePlugin):
         Start the XEP-0030 plugin.
         """
         self.xmpp.register_handler(
-                Callback('Disco Info',
+                CoroutineCallback('Disco Info',
                          StanzaPath('iq/disco_info'),
                          self._handle_disco_info))
 
         self.xmpp.register_handler(
-                Callback('Disco Items',
+                CoroutineCallback('Disco Items',
                          StanzaPath('iq/disco_items'),
                          self._handle_disco_items))
 
@@ -228,10 +228,13 @@ class XEP_0030(BasePlugin):
             self.api.restore_default(op, jid, node)
 
     def supports(self, jid=None, node=None, feature=None, local=False,
-                       cached=True, ifrom=None):
+                 cached=True, ifrom=None) -> Future:
         """
         Check if a JID supports a given feature.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         Return values:
         :param True: The feature is supported
         :param False: The feature is not listed as supported
@@ -259,10 +262,13 @@ class XEP_0030(BasePlugin):
         return self.api['supports'](jid, node, ifrom, data)
 
     def has_identity(self, jid=None, node=None, category=None, itype=None,
-                     lang=None, local=False, cached=True, ifrom=None):
+                     lang=None, local=False, cached=True, ifrom=None) -> Future:
         """
         Check if a JID provides a given identity.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         Return values:
         :param True: The identity is provided
         :param False: The identity is not listed
@@ -324,8 +330,7 @@ class XEP_0030(BasePlugin):
             callback(results)
         return results
 
-    @future_wrapper
+    async def get_info(self, jid=None, node=None, local=None,
-    def get_info(self, jid=None, node=None, local=None,
                        cached=None, **kwargs):
         """
         Retrieve the disco#info results from a given JID/node combination.
@@ -338,6 +343,9 @@ class XEP_0030(BasePlugin):
         If requesting items from a local JID/node, then only a DiscoInfo
         stanza will be returned. Otherwise, an Iq stanza will be returned.
 
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+
         :param jid: Request info from this JID.
         :param node: The particular node to query.
         :param local: If true, then the query is for a JID/node
@@ -369,18 +377,21 @@ class XEP_0030(BasePlugin):
         if local:
             log.debug("Looking up local disco#info data " + \
                       "for %s, node %s.", jid, node)
-            info = self.api['get_info'](jid, node,
+            info = await self.api['get_info'](
-                    kwargs.get('ifrom', None),
+                jid, node, kwargs.get('ifrom', None),
-                    kwargs)
+                kwargs
+            )
             info = self._fix_default_info(info)
             return self._wrap(kwargs.get('ifrom', None), jid, info)
 
         if cached:
             log.debug("Looking up cached disco#info data " + \
                       "for %s, node %s.", jid, node)
-            info = self.api['get_cached_info'](jid, node,
+            info = await self.api['get_cached_info'](
-                    kwargs.get('ifrom', None),
+                jid, node,
-                    kwargs)
+                kwargs.get('ifrom', None),
+                kwargs
+            )
             if info is not None:
                 return self._wrap(kwargs.get('ifrom', None), jid, info)
 
@@ -390,21 +401,24 @@ class XEP_0030(BasePlugin):
         iq['to'] = jid
         iq['type'] = 'get'
         iq['disco_info']['node'] = node if node else ''
-        return iq.send(timeout=kwargs.get('timeout', None),
+        return await iq.send(timeout=kwargs.get('timeout', None),
                        callback=kwargs.get('callback', None),
                        timeout_callback=kwargs.get('timeout_callback', None))
 
-    def set_info(self, jid=None, node=None, info=None):
+    def set_info(self, jid=None, node=None, info=None) -> Future:
         """
         Set the disco#info data for a JID/node based on an existing
         disco#info stanza.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         """
         if isinstance(info, Iq):
             info = info['disco_info']
-        self.api['set_info'](jid, node, None, info)
+        return self.api['set_info'](jid, node, None, info)
 
-    @future_wrapper
+    async def get_items(self, jid=None, node=None, local=False, **kwargs):
-    def get_items(self, jid=None, node=None, local=False, **kwargs):
         """
         Retrieve the disco#items results from a given JID/node combination.
 
@@ -416,6 +430,9 @@ class XEP_0030(BasePlugin):
         If requesting items from a local JID/node, then only a DiscoItems
         stanza will be returned. Otherwise, an Iq stanza will be returned.
 
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+
         :param jid: Request info from this JID.
         :param node: The particular node to query.
         :param local: If true, then the query is for a JID/node
@@ -428,7 +445,7 @@ class XEP_0030(BasePlugin):
                          Otherwise the parameter is ignored.
         """
         if local or local is None and jid is None:
-            items = self.api['get_items'](jid, node,
+            items = await self.api['get_items'](jid, node,
                     kwargs.get('ifrom', None),
                     kwargs)
             return self._wrap(kwargs.get('ifrom', None), jid, items)
@@ -440,43 +457,52 @@ class XEP_0030(BasePlugin):
         iq['type'] = 'get'
         iq['disco_items']['node'] = node if node else ''
         if kwargs.get('iterator', False) and self.xmpp['xep_0059']:
-            raise NotImplementedError("XEP 0059 has not yet been fixed")
             return self.xmpp['xep_0059'].iterate(iq, 'disco_items')
         else:
-            return iq.send(timeout=kwargs.get('timeout', None),
+            return await iq.send(timeout=kwargs.get('timeout', None),
                            callback=kwargs.get('callback', None),
                            timeout_callback=kwargs.get('timeout_callback', None))
 
-    def set_items(self, jid=None, node=None, **kwargs):
+    def set_items(self, jid=None, node=None, **kwargs) -> Future:
         """
         Set or replace all items for the specified JID/node combination.
 
         The given items must be in a list or set where each item is a
         tuple of the form: (jid, node, name).
 
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: Optional node to modify.
         :param items: A series of items in tuple format.
         """
-        self.api['set_items'](jid, node, None, kwargs)
+        return self.api['set_items'](jid, node, None, kwargs)
 
-    def del_items(self, jid=None, node=None, **kwargs):
+    def del_items(self, jid=None, node=None, **kwargs) -> Future:
         """
         Remove all items from the given JID/node combination.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         Arguments:
         :param jid: The JID to modify.
         :param node: Optional node to modify.
         """
-        self.api['del_items'](jid, node, None, kwargs)
+        return self.api['del_items'](jid, node, None, kwargs)
 
-    def add_item(self, jid='', name='', node=None, subnode='', ijid=None):
+    def add_item(self, jid='', name='', node=None, subnode='', ijid=None) -> Future:
         """
         Add a new item element to the given JID/node combination.
 
         Each item is required to have a JID, but may also specify
         a node value to reference non-addressable entities.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID for the item.
         :param name: Optional name for the item.
         :param node: The node to modify.
@@ -488,9 +514,9 @@ class XEP_0030(BasePlugin):
         kwargs = {'ijid': jid,
                   'name': name,
                   'inode': subnode}
-        self.api['add_item'](ijid, node, None, kwargs)
+        return self.api['add_item'](ijid, node, None, kwargs)
 
-    def del_item(self, jid=None, node=None, **kwargs):
+    def del_item(self, jid=None, node=None, **kwargs) -> Future:
         """
         Remove a single item from the given JID/node combination.
 
@@ -499,10 +525,10 @@ class XEP_0030(BasePlugin):
         :param ijid: The item's JID.
         :param inode: The item's node.
         """
-        self.api['del_item'](jid, node, None, kwargs)
+        return self.api['del_item'](jid, node, None, kwargs)
 
     def add_identity(self, category='', itype='', name='',
-                     node=None, jid=None, lang=None):
+                     node=None, jid=None, lang=None) -> Future:
         """
         Add a new identity to the given JID/node combination.
 
@@ -514,6 +540,9 @@ class XEP_0030(BasePlugin):
         category/type/xml:lang pairs are allowed so long as the
         names are different. A category and type is always required.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param category: The identity's category.
         :param itype: The identity's type.
         :param name: Optional name for the identity.
@@ -525,24 +554,31 @@ class XEP_0030(BasePlugin):
                   'itype': itype,
                   'name': name,
                   'lang': lang}
-        self.api['add_identity'](jid, node, None, kwargs)
+        return self.api['add_identity'](jid, node, None, kwargs)
 
     def add_feature(self, feature: str, node: Optional[str] = None,
-                    jid: Optional[JID] = None):
+                    jid: Optional[JID] = None) -> Future:
         """
         Add a feature to a JID/node combination.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param feature: The namespace of the supported feature.
         :param node: The node to modify.
         :param jid: The JID to modify.
         """
         kwargs = {'feature': feature}
-        self.api['add_feature'](jid, node, None, kwargs)
+        return self.api['add_feature'](jid, node, None, kwargs)
 
-    def del_identity(self, jid: Optional[JID] = None, node: Optional[str] = None, **kwargs):
+    def del_identity(self, jid: Optional[JID] = None,
+                     node: Optional[str] = None, **kwargs) -> Future:
         """
         Remove an identity from the given JID/node combination.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         :param category: The identity's category.
@@ -550,67 +586,82 @@ class XEP_0030(BasePlugin):
         :param name: Optional, human readable name for the identity.
         :param lang: Optional, the identity's xml:lang value.
         """
-        self.api['del_identity'](jid, node, None, kwargs)
+        return self.api['del_identity'](jid, node, None, kwargs)
 
-    def del_feature(self, jid=None, node=None, **kwargs):
+    def del_feature(self, jid=None, node=None, **kwargs) -> Future:
         """
         Remove a feature from a given JID/node combination.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         :param feature: The feature's namespace.
         """
-        self.api['del_feature'](jid, node, None, kwargs)
+        return self.api['del_feature'](jid, node, None, kwargs)
 
-    def set_identities(self, jid=None, node=None, **kwargs):
+    def set_identities(self, jid=None, node=None, **kwargs) -> Future:
         """
         Add or replace all identities for the given JID/node combination.
 
         The identities must be in a set where each identity is a tuple
         of the form: (category, type, lang, name)
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         :param identities: A set of identities in tuple form.
         :param lang: Optional, xml:lang value.
         """
-        self.api['set_identities'](jid, node, None, kwargs)
+        return self.api['set_identities'](jid, node, None, kwargs)
 
-    def del_identities(self, jid=None, node=None, **kwargs):
+    def del_identities(self, jid=None, node=None, **kwargs) -> Future:
         """
         Remove all identities for a JID/node combination.
 
         If a language is specified, only identities using that
         language will be removed.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         :param lang: Optional. If given, only remove identities
                     using this xml:lang value.
         """
-        self.api['del_identities'](jid, node, None, kwargs)
+        return self.api['del_identities'](jid, node, None, kwargs)
 
-    def set_features(self, jid=None, node=None, **kwargs):
+    def set_features(self, jid=None, node=None, **kwargs) -> Future:
         """
         Add or replace the set of supported features
         for a JID/node combination.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         :param features: The new set of supported features.
         """
-        self.api['set_features'](jid, node, None, kwargs)
+        return self.api['set_features'](jid, node, None, kwargs)
 
-    def del_features(self, jid=None, node=None, **kwargs):
+    def del_features(self, jid=None, node=None, **kwargs) -> Future:
         """
         Remove all features from a JID/node combination.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         """
-        self.api['del_features'](jid, node, None, kwargs)
+        return self.api['del_features'](jid, node, None, kwargs)
 
-    def _run_node_handler(self, htype, jid, node=None, ifrom=None, data=None):
+    async def _run_node_handler(self, htype, jid, node=None, ifrom=None, data=None):
         """
         Execute the most specific node handler for the given
         JID/node combination.
@@ -623,9 +674,9 @@ class XEP_0030(BasePlugin):
         if not data:
             data = {}
 
-        return self.api[htype](jid, node, ifrom, data)
+        return await self.api[htype](jid, node, ifrom, data)
 
-    def _handle_disco_info(self, iq):
+    async def _handle_disco_info(self, iq):
         """
         Process an incoming disco#info stanza. If it is a get
         request, find and return the appropriate identities
@@ -637,10 +688,10 @@ class XEP_0030(BasePlugin):
         if iq['type'] == 'get':
             log.debug("Received disco info query from " + \
                       "<%s> to <%s>.", iq['from'], iq['to'])
-            info = self.api['get_info'](iq['to'],
+            info = await self.api['get_info'](iq['to'],
-                                        iq['disco_info']['node'],
+                                              iq['disco_info']['node'],
-                                        iq['from'],
+                                              iq['from'],
-                                        iq)
+                                              iq)
             if isinstance(info, Iq):
                 info['id'] = iq['id']
                 info.send()
@@ -662,13 +713,13 @@ class XEP_0030(BasePlugin):
                     ito = iq['to'].full
                 else:
                     ito = None
-                self.api['cache_info'](iq['from'],
+                await self.api['cache_info'](iq['from'],
-                                       iq['disco_info']['node'],
+                                             iq['disco_info']['node'],
-                                       ito,
+                                             ito,
-                                       iq)
+                                             iq)
             self.xmpp.event('disco_info', iq)
 
-    def _handle_disco_items(self, iq):
+    async def _handle_disco_items(self, iq):
         """
         Process an incoming disco#items stanza. If it is a get
         request, find and return the appropriate items. If it
@@ -679,10 +730,10 @@ class XEP_0030(BasePlugin):
         if iq['type'] == 'get':
             log.debug("Received disco items query from " + \
                       "<%s> to <%s>.", iq['from'], iq['to'])
-            items = self.api['get_items'](iq['to'],
+            items = await self.api['get_items'](iq['to'],
-                                          iq['disco_items']['node'],
+                                                iq['disco_items']['node'],
-                                          iq['from'],
+                                                iq['from'],
-                                          iq)
+                                                iq)
             if isinstance(items, Iq):
                 items.send()
             else:

slixmpp/plugins/xep_0030/static.py

@@ -109,7 +109,7 @@ class StaticDisco(object):
     # the requester's JID, except for cached results. To do that,
     # register a custom node handler.
 
-    def supports(self, jid, node, ifrom, data):
+    async def supports(self, jid, node, ifrom, data):
         """
         Check if a JID supports a given feature.
 
@@ -137,8 +137,8 @@ class StaticDisco(object):
             return False
 
         try:
-            info = self.disco.get_info(jid=jid, node=node,
+            info = await self.disco.get_info(jid=jid, node=node,
-                                       ifrom=ifrom, **data)
+                                             ifrom=ifrom, **data)
             info = self.disco._wrap(ifrom, jid, info, True)
             features = info['disco_info']['features']
             return feature in features
@@ -147,7 +147,7 @@ class StaticDisco(object):
         except IqTimeout:
             return None
 
-    def has_identity(self, jid, node, ifrom, data):
+    async def has_identity(self, jid, node, ifrom, data):
         """
         Check if a JID has a given identity.
 
@@ -176,8 +176,8 @@ class StaticDisco(object):
                 'cached': data.get('cached', True)}
 
         try:
-            info = self.disco.get_info(jid=jid, node=node,
+            info = await self.disco.get_info(jid=jid, node=node,
-                                       ifrom=ifrom, **data)
+                                             ifrom=ifrom, **data)
             info = self.disco._wrap(ifrom, jid, info, True)
             trunc = lambda i: (i[0], i[1], i[2])
             return identity in map(trunc, info['disco_info']['identities'])

slixmpp/plugins/xep_0047/ibb.py

@@ -1,7 +1,6 @@
 # Slixmpp: The Slick XMPP Library
 # This file is part of Slixmpp
 # See the file LICENSE for copying permission
-import asyncio
 import uuid
 import logging
 
@@ -13,7 +12,7 @@ from typing import (
 from slixmpp import JID
 from slixmpp.stanza import Message, Iq
 from slixmpp.exceptions import XMPPError
-from slixmpp.xmlstream.handler import Callback
+from slixmpp.xmlstream.handler import CoroutineCallback
 from slixmpp.xmlstream.matcher import StanzaPath
 from slixmpp.xmlstream import register_stanza_plugin
 from slixmpp.plugins import BasePlugin
@@ -41,6 +40,12 @@ class XEP_0047(BasePlugin):
     - ``auto_accept`` (default: ``False``): if incoming streams should be
         accepted automatically.
 
+    - :term:`authorized (0047 version)`
+    - :term:`authorized_sid (0047 version)`
+    - :term:`preauthorize_sid (0047 version)`
+    - :term:`get_stream`
+    - :term:`set_stream`
+    - :term:`del_stream`
     """
 
     name = 'xep_0047'
@@ -62,22 +67,22 @@ class XEP_0047(BasePlugin):
         register_stanza_plugin(Iq, Data)
         register_stanza_plugin(Message, Data)
 
-        self.xmpp.register_handler(Callback(
+        self.xmpp.register_handler(CoroutineCallback(
             'IBB Open',
             StanzaPath('iq@type=set/ibb_open'),
             self._handle_open_request))
 
-        self.xmpp.register_handler(Callback(
+        self.xmpp.register_handler(CoroutineCallback(
             'IBB Close',
             StanzaPath('iq@type=set/ibb_close'),
             self._handle_close))
 
-        self.xmpp.register_handler(Callback(
+        self.xmpp.register_handler(CoroutineCallback(
             'IBB Data',
             StanzaPath('iq@type=set/ibb_data'),
             self._handle_data))
 
-        self.xmpp.register_handler(Callback(
+        self.xmpp.register_handler(CoroutineCallback(
             'IBB Message Data',
             StanzaPath('message/ibb_data'),
             self._handle_data))
@@ -109,14 +114,14 @@ class XEP_0047(BasePlugin):
         if (jid, sid, peer_jid) in self._streams:
             del self._streams[(jid, sid, peer_jid)]
 
-    def _accept_stream(self, iq):
+    async def _accept_stream(self, iq):
         receiver = iq['to']
         sender = iq['from']
         sid = iq['ibb_open']['sid']
 
-        if self.api['authorized_sid'](receiver, sid, sender, iq):
+        if await self.api['authorized_sid'](receiver, sid, sender, iq):
             return True
-        return self.api['authorized'](receiver, sid, sender, iq)
+        return await self.api['authorized'](receiver, sid, sender, iq)
 
     def _authorized(self, jid, sid, ifrom, iq):
         if self.auto_accept:
@@ -169,14 +174,14 @@ class XEP_0047(BasePlugin):
         stream.self_jid = result['to']
         stream.peer_jid = result['from']
         stream.stream_started = True
-        self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream)
+        await self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream)
         if callback is not None:
             self.xmpp.add_event_handler('ibb_stream_start', callback, disposable=True)
         self.xmpp.event('ibb_stream_start', stream)
         self.xmpp.event('stream:%s:%s' % (stream.sid, stream.peer_jid), stream)
         return stream
 
-    def _handle_open_request(self, iq: Iq):
+    async def _handle_open_request(self, iq: Iq):
         sid = iq['ibb_open']['sid']
         size = iq['ibb_open']['block_size'] or self.block_size
 
@@ -185,7 +190,7 @@ class XEP_0047(BasePlugin):
         if not sid:
             raise XMPPError(etype='modify', condition='bad-request')
 
-        if not self._accept_stream(iq):
+        if not await self._accept_stream(iq):
             raise XMPPError(etype='cancel', condition='not-acceptable')
 
         if size > self.max_block_size:
@@ -194,25 +199,25 @@ class XEP_0047(BasePlugin):
         stream = IBBytestream(self.xmpp, sid, size,
                               iq['to'], iq['from'])
         stream.stream_started = True
-        self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream)
+        await self.api['set_stream'](stream.self_jid, stream.sid, stream.peer_jid, stream)
         iq.reply().send()
 
         self.xmpp.event('ibb_stream_start', stream)
         self.xmpp.event('stream:%s:%s' % (sid, stream.peer_jid), stream)
 
-    def _handle_data(self, stanza: Union[Iq, Message]):
+    async def _handle_data(self, stanza: Union[Iq, Message]):
         sid = stanza['ibb_data']['sid']
-        stream = self.api['get_stream'](stanza['to'], sid, stanza['from'])
+        stream = await self.api['get_stream'](stanza['to'], sid, stanza['from'])
         if stream is not None and stanza['from'] == stream.peer_jid:
             stream._recv_data(stanza)
         else:
             raise XMPPError('item-not-found')
 
-    def _handle_close(self, iq: Iq):
+    async def _handle_close(self, iq: Iq):
         sid = iq['ibb_close']['sid']
-        stream = self.api['get_stream'](iq['to'], sid, iq['from'])
+        stream = await self.api['get_stream'](iq['to'], sid, iq['from'])
         if stream is not None and iq['from'] == stream.peer_jid:
             stream._closed(iq)
-            self.api['del_stream'](stream.self_jid, stream.sid, stream.peer_jid)
+            await self.api['del_stream'](stream.self_jid, stream.sid, stream.peer_jid)
         else:
             raise XMPPError('item-not-found')

slixmpp/plugins/xep_0054/vcard_temp.py

@@ -11,7 +11,7 @@ from slixmpp import JID
 from slixmpp.stanza import Iq
 from slixmpp.exceptions import XMPPError
 from slixmpp.xmlstream import register_stanza_plugin
-from slixmpp.xmlstream.handler import Callback
+from slixmpp.xmlstream.handler import CoroutineCallback
 from slixmpp.xmlstream.matcher import StanzaPath
 from slixmpp.plugins import BasePlugin
 from slixmpp.plugins.xep_0054 import VCardTemp, stanza
@@ -46,7 +46,7 @@ class XEP_0054(BasePlugin):
         self._vcard_cache = {}
 
         self.xmpp.register_handler(
-                Callback('VCardTemp',
+                CoroutineCallback('VCardTemp',
                     StanzaPath('iq/vcard_temp'),
                     self._handle_get_vcard))
 
@@ -61,13 +61,15 @@ class XEP_0054(BasePlugin):
         """Return an empty vcard element."""
         return VCardTemp()
 
-    @future_wrapper
+    async def get_vcard(self, jid: Optional[JID] = None, *,
-    def get_vcard(self, jid: Optional[JID] = None, *,
+                        local: Optional[bool] = None, cached: bool = False,
-                  local: Optional[bool] = None, cached: bool = False,
+                        ifrom: Optional[JID] = None,
-                  ifrom: Optional[JID] = None,
+                        **iqkwargs) -> Iq:
-                  **iqkwargs) -> Future:
         """Retrieve a VCard.
 
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+
         :param jid: JID of the entity to fetch the VCard from.
         :param local: Only check internally for a vcard.
         :param cached: Whether to check in the local cache before
@@ -87,7 +89,7 @@ class XEP_0054(BasePlugin):
                 local = True
 
         if local:
-            vcard = self.api['get_vcard'](jid, None, ifrom)
+            vcard = await self.api['get_vcard'](jid, None, ifrom)
             if not isinstance(vcard, Iq):
                 iq = self.xmpp.Iq()
                 if vcard is None:
@@ -97,7 +99,7 @@ class XEP_0054(BasePlugin):
             return vcard
 
         if cached:
-            vcard = self.api['get_vcard'](jid, None, ifrom)
+            vcard = await self.api['get_vcard'](jid, None, ifrom)
             if vcard is not None:
                 if not isinstance(vcard, Iq):
                     iq = self.xmpp.Iq()
@@ -107,31 +109,33 @@ class XEP_0054(BasePlugin):
 
         iq = self.xmpp.make_iq_get(ito=jid, ifrom=ifrom)
         iq.enable('vcard_temp')
-        return iq.send(**iqkwargs)
+        return await iq.send(**iqkwargs)
 
-    @future_wrapper
+    async def publish_vcard(self, vcard: Optional[VCardTemp] = None,
-    def publish_vcard(self, vcard: Optional[VCardTemp] = None,
+                            jid: Optional[JID] = None,
-                      jid: Optional[JID] = None,
+                            ifrom: Optional[JID] = None, **iqkwargs):
-                      ifrom: Optional[JID] = None, **iqkwargs) -> Future:
         """Publish a vcard.
 
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+
         :param vcard: The VCard to publish.
         :param jid: The JID to publish the VCard to.
         """
-        self.api['set_vcard'](jid, None, ifrom, vcard)
+        await self.api['set_vcard'](jid, None, ifrom, vcard)
         if self.xmpp.is_component:
             return
 
         iq = self.xmpp.make_iq_set(ito=jid, ifrom=ifrom)
         iq.append(vcard)
-        return iq.send(**iqkwargs)
+        await iq.send(**iqkwargs)
 
-    def _handle_get_vcard(self, iq: Iq):
+    async def _handle_get_vcard(self, iq: Iq):
         if iq['type'] == 'result':
-            self.api['set_vcard'](jid=iq['from'], args=iq['vcard_temp'])
+            await self.api['set_vcard'](jid=iq['from'], args=iq['vcard_temp'])
             return
         elif iq['type'] == 'get' and self.xmpp.is_component:
-            vcard = self.api['get_vcard'](iq['to'].bare, ifrom=iq['from'])
+            vcard = await self.api['get_vcard'](iq['to'].bare, ifrom=iq['from'])
             if isinstance(vcard, Iq):
                 vcard.send()
             else:

slixmpp/plugins/xep_0065/proxy.py

@@ -8,7 +8,7 @@ from uuid import uuid4
 from slixmpp.stanza import Iq
 from slixmpp.exceptions import XMPPError
 from slixmpp.xmlstream import register_stanza_plugin
-from slixmpp.xmlstream.handler import Callback
+from slixmpp.xmlstream.handler import CoroutineCallback
 from slixmpp.xmlstream.matcher import StanzaPath
 from slixmpp.plugins.base import BasePlugin
 
@@ -34,10 +34,11 @@ class XEP_0065(BasePlugin):
         self._sessions = {}
         self._preauthed_sids = {}
 
-        self.xmpp.register_handler(
+        self.xmpp.register_handler(CoroutineCallback(
-            Callback('Socks5 Bytestreams',
+            'Socks5 Bytestreams',
-                     StanzaPath('iq@type=set/socks/streamhost'),
+             StanzaPath('iq@type=set/socks/streamhost'),
-                     self._handle_streamhost))
+             self._handle_streamhost
+        ))
 
         self.api.register(self._authorized, 'authorized', default=True)
         self.api.register(self._authorized_sid, 'authorized_sid', default=True)
@@ -158,13 +159,13 @@ class XEP_0065(BasePlugin):
         digest.update(str(target).encode('utf8'))
         return digest.hexdigest()
 
-    def _handle_streamhost(self, iq):
+    async def _handle_streamhost(self, iq):
         """Handle incoming SOCKS5 session request."""
         sid = iq['socks']['sid']
         if not sid:
             raise XMPPError(etype='modify', condition='bad-request')
 
-        if not self._accept_stream(iq):
+        if not await self._accept_stream(iq):
             raise XMPPError(etype='modify', condition='not-acceptable')
 
         streamhosts = iq['socks']['streamhosts']
@@ -180,39 +181,37 @@ class XEP_0065(BasePlugin):
                     streamhost['host'],
                     streamhost['port']))
 
-        async def gather(futures, iq, streamhosts):
+        proxies = await asyncio.gather(*proxy_futures, return_exceptions=True)
-            proxies = await asyncio.gather(*futures, return_exceptions=True)
+        for streamhost, proxy in zip(streamhosts, proxies):
-            for streamhost, proxy in zip(streamhosts, proxies):
+            if isinstance(proxy, ValueError):
-                if isinstance(proxy, ValueError):
+                continue
-                    continue
+            elif isinstance(proxy, socket.error):
-                elif isinstance(proxy, socket.error):
+                log.error('Socket error while connecting to the proxy.')
-                    log.error('Socket error while connecting to the proxy.')
+                continue
-                    continue
+            proxy = proxy[1]
-                proxy = proxy[1]
+            # TODO: what if the future never happens?
-                # TODO: what if the future never happens?
+            try:
-                try:
+                addr, port = await proxy.connected
-                    addr, port = await proxy.connected
+            except socket.error:
-                except socket.error:
+                log.exception('Socket error while connecting to the proxy.')
-                    log.exception('Socket error while connecting to the proxy.')
+                continue
-                    continue
+            # TODO: make a better choice than just the first working one.
-                # TODO: make a better choice than just the first working one.
+            used_streamhost = streamhost['jid']
-                used_streamhost = streamhost['jid']
+            conn = proxy
-                conn = proxy
+            break
-                break
+        else:
-            else:
+            raise XMPPError(etype='cancel', condition='item-not-found')
-                raise XMPPError(etype='cancel', condition='item-not-found')
 
-            # TODO: close properly the connection to the other proxies.
+        # TODO: close properly the connection to the other proxies.
 
-            iq = iq.reply()
+        iq = iq.reply()
-            self._sessions[sid] = conn
+        self._sessions[sid] = conn
-            iq['socks']['sid'] = sid
+        iq['socks']['sid'] = sid
-            iq['socks']['streamhost_used']['jid'] = used_streamhost
+        iq['socks']['streamhost_used']['jid'] = used_streamhost
-            iq.send()
+        iq.send()
-            self.xmpp.event('socks5_stream', conn)
+        self.xmpp.event('socks5_stream', conn)
-            self.xmpp.event('stream:%s:%s' % (sid, requester), conn)
+        self.xmpp.event('stream:%s:%s' % (sid, requester), conn)
 
-        asyncio.ensure_future(gather(proxy_futures, iq, streamhosts))
 
     def activate(self, proxy, sid, target, ifrom=None, timeout=None, callback=None):
         """Activate the socks5 session that has been negotiated."""
@@ -253,14 +252,14 @@ class XEP_0065(BasePlugin):
         factory = lambda: Socks5Protocol(dest, 0, self.xmpp.event)
         return self.xmpp.loop.create_connection(factory, proxy, proxy_port)
 
-    def _accept_stream(self, iq):
+    async def _accept_stream(self, iq):
         receiver = iq['to']
         sender = iq['from']
         sid = iq['socks']['sid']
 
-        if self.api['authorized_sid'](receiver, sid, sender, iq):
+        if await self.api['authorized_sid'](receiver, sid, sender, iq):
             return True
-        return self.api['authorized'](receiver, sid, sender, iq)
+        return await self.api['authorized'](receiver, sid, sender, iq)
 
     def _authorized(self, jid, sid, ifrom, iq):
         return self.auto_accept

slixmpp/plugins/xep_0077/register.py

@@ -106,10 +106,9 @@ class XEP_0077(BasePlugin):
     def _user_remove(self, jid, node, ifrom, iq):
         return self._user_store.pop(iq["from"].bare)
 
-    def _make_registration_form(self, jid, node, ifrom, iq: Iq):
+    async def _make_registration_form(self, jid, node, ifrom, iq: Iq):
         reg = iq["register"]
-        user = self.api["user_get"](None, None, None, iq)
+        user = await self.api["user_get"](None, None, iq['from'], iq)
-
 
         if user is None:
             user = {}
@@ -135,11 +134,11 @@ class XEP_0077(BasePlugin):
 
     async def _handle_registration(self, iq: Iq):
         if iq["type"] == "get":
-            self._send_form(iq)
+            await self._send_form(iq)
         elif iq["type"] == "set":
             if iq["register"]["remove"]:
                 try:
-                    self.api["user_remove"](None, None, iq["from"], iq)
+                    await self.api["user_remove"](None, None, iq["from"], iq)
                 except KeyError:
                     _send_error(
                         iq,
@@ -168,7 +167,7 @@ class XEP_0077(BasePlugin):
                     return
 
             try:
-                self.api["user_validate"](None, None, iq["from"], iq["register"])
+                await self.api["user_validate"](None, None, iq["from"], iq["register"])
             except ValueError as e:
                 _send_error(
                     iq,
@@ -182,8 +181,8 @@ class XEP_0077(BasePlugin):
                 reply.send()
                 self.xmpp.event("user_register", iq)
 
-    def _send_form(self, iq):
+    async def _send_form(self, iq):
-        reply = self.api["make_registration_form"](None, None, iq["from"], iq)
+        reply = await self.api["make_registration_form"](None, None, iq["from"], iq)
         reply.send()
 
     def _force_registration(self, event):

slixmpp/plugins/xep_0095/stream_initiation.py

@@ -81,7 +81,7 @@ class XEP_0095(BasePlugin):
             self._methods_order.remove((order, method, plugin_name))
             self._methods_order.sort()
 
-    def _handle_request(self, iq):
+    async def _handle_request(self, iq):
         profile = iq['si']['profile']
         sid = iq['si']['id']
 
@@ -119,7 +119,7 @@ class XEP_0095(BasePlugin):
         receiver = iq['to']
         sender = iq['from']
 
-        self.api['add_pending'](receiver, sid, sender, {
+        await self.api['add_pending'](receiver, sid, sender, {
             'response_id': iq['id'],
             'method': selected_method,
             'profile': profile
@@ -153,8 +153,13 @@ class XEP_0095(BasePlugin):
                 options=methods)
         return si.send(**iqargs)
 
-    def accept(self, jid, sid, payload=None, ifrom=None, stream_handler=None):
+    async def accept(self, jid, sid, payload=None, ifrom=None, stream_handler=None):
-        stream = self.api['get_pending'](ifrom, sid, jid)
+        """Accept a stream initiation.
+
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+        """
+        stream = await self.api['get_pending'](ifrom, sid, jid)
         iq = self.xmpp.Iq()
         iq['id'] = stream['response_id']
         iq['to'] = jid
@@ -174,16 +179,21 @@ class XEP_0095(BasePlugin):
         method_plugin = self._methods[stream['method']][0]
         self.xmpp[method_plugin].api['preauthorize_sid'](ifrom, sid, jid)
 
-        self.api['del_pending'](ifrom, sid, jid)
+        await self.api['del_pending'](ifrom, sid, jid)
 
         if stream_handler:
             self.xmpp.add_event_handler('stream:%s:%s' % (sid, jid),
                     stream_handler,
                     disposable=True)
-        return iq.send()
+        return await iq.send()
 
-    def decline(self, jid, sid, ifrom=None):
+    async def decline(self, jid, sid, ifrom=None):
-        stream = self.api['get_pending'](ifrom, sid, jid)
+        """Decline a stream initiation.
+
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+        """
+        stream = await self.api['get_pending'](ifrom, sid, jid)
         if not stream:
             return
         iq = self.xmpp.Iq()
@@ -193,8 +203,8 @@ class XEP_0095(BasePlugin):
         iq['type'] = 'error'
         iq['error']['condition'] = 'forbidden'
         iq['error']['text'] = 'Offer declined'
-        self.api['del_pending'](ifrom, sid, jid)
+        await self.api['del_pending'](ifrom, sid, jid)
-        return iq.send()
+        return await iq.send()
 
     def _add_pending(self, jid, node, ifrom, data):
         with self._pending_lock:

slixmpp/plugins/xep_0115/caps.py

@@ -7,6 +7,8 @@ import logging
 import hashlib
 import base64
 
+from asyncio import Future
+
 from slixmpp import __version__
 from slixmpp.stanza import StreamFeatures, Presence, Iq
 from slixmpp.xmlstream import register_stanza_plugin, JID
@@ -104,14 +106,14 @@ class XEP_0115(BasePlugin):
     def session_bind(self, jid):
         self.xmpp['xep_0030'].add_feature(stanza.Capabilities.namespace)
 
-    def _filter_add_caps(self, stanza):
+    async def _filter_add_caps(self, stanza):
         if not isinstance(stanza, Presence) or not self.broadcast:
             return stanza
 
         if stanza['type'] not in ('available', 'chat', 'away', 'dnd', 'xa'):
             return stanza
 
-        ver = self.get_verstring(stanza['from'])
+        ver = await self.get_verstring(stanza['from'])
         if ver:
             stanza['caps']['node'] = self.caps_node
             stanza['caps']['hash'] = self.hash
@@ -145,13 +147,13 @@ class XEP_0115(BasePlugin):
 
         ver = pres['caps']['ver']
 
-        existing_verstring = self.get_verstring(pres['from'].full)
+        existing_verstring = await self.get_verstring(pres['from'].full)
         if str(existing_verstring) == str(ver):
             return
 
-        existing_caps = self.get_caps(verstring=ver)
+        existing_caps = await self.get_caps(verstring=ver)
         if existing_caps is not None:
-            self.assign_verstring(pres['from'], ver)
+            await self.assign_verstring(pres['from'], ver)
             return
 
         ifrom = pres['to'] if self.xmpp.is_component else None
@@ -174,13 +176,13 @@ class XEP_0115(BasePlugin):
             if isinstance(caps, Iq):
                 caps = caps['disco_info']
 
-            if self._validate_caps(caps, pres['caps']['hash'],
+            if await self._validate_caps(caps, pres['caps']['hash'],
-                                         pres['caps']['ver']):
+                                               pres['caps']['ver']):
-                self.assign_verstring(pres['from'], pres['caps']['ver'])
+                await self.assign_verstring(pres['from'], pres['caps']['ver'])
         except XMPPError:
             log.debug("Could not retrieve disco#info results for caps for %s", node)
 
-    def _validate_caps(self, caps, hash, check_verstring):
+    async def _validate_caps(self, caps, hash, check_verstring):
         # Check Identities
         full_ids = caps.get_identities(dedupe=False)
         deduped_ids = caps.get_identities()
@@ -232,7 +234,7 @@ class XEP_0115(BasePlugin):
                 verstring, check_verstring))
             return False
 
-        self.cache_caps(verstring, caps)
+        await self.cache_caps(verstring, caps)
         return True
 
     def generate_verstring(self, info, hash):
@@ -290,12 +292,13 @@ class XEP_0115(BasePlugin):
             if isinstance(info, Iq):
                 info = info['disco_info']
             ver = self.generate_verstring(info, self.hash)
-            self.xmpp['xep_0030'].set_info(
+            await self.xmpp['xep_0030'].set_info(
-                    jid=jid,
+                jid=jid,
-                    node='%s#%s' % (self.caps_node, ver),
+                node='%s#%s' % (self.caps_node, ver),
-                    info=info)
+                info=info
-            self.cache_caps(ver, info)
+            )
-            self.assign_verstring(jid, ver)
+            await self.cache_caps(ver, info)
+            await self.assign_verstring(jid, ver)
 
             if self.xmpp.sessionstarted and self.broadcast:
                 if self.xmpp.is_component or preserve:
@@ -306,32 +309,53 @@ class XEP_0115(BasePlugin):
         except XMPPError:
             return
 
-    def get_verstring(self, jid=None):
+    def get_verstring(self, jid=None) -> Future:
+        """Get the stored verstring for a JID.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
         if jid in ('', None):
             jid = self.xmpp.boundjid.full
         if isinstance(jid, JID):
             jid = jid.full
         return self.api['get_verstring'](jid)
 
-    def assign_verstring(self, jid=None, verstring=None):
+    def assign_verstring(self, jid=None, verstring=None) -> Future:
+        """Assign a vertification string to a jid.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
         if jid in (None, ''):
             jid = self.xmpp.boundjid.full
         if isinstance(jid, JID):
             jid = jid.full
         return self.api['assign_verstring'](jid, args={
-            'verstring': verstring})
+            'verstring': verstring
+        })
 
-    def cache_caps(self, verstring=None, info=None):
+    def cache_caps(self, verstring=None, info=None) -> Future:
+        """Add caps to the cache.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
         data = {'verstring': verstring, 'info': info}
         return self.api['cache_caps'](args=data)
 
-    def get_caps(self, jid=None, verstring=None):
+    async def get_caps(self, jid=None, verstring=None):
+        """Get caps for a JID.
+
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+        """
         if verstring is None:
             if jid is not None:
-                verstring = self.get_verstring(jid)
+                verstring = await self.get_verstring(jid)
             else:
                 return None
         if isinstance(jid, JID):
             jid = jid.full
         data = {'verstring': verstring}
-        return self.api['get_caps'](jid, args=data)
+        return await self.api['get_caps'](jid, args=data)

slixmpp/plugins/xep_0115/static.py

@@ -32,7 +32,7 @@ class StaticCaps(object):
         self.static = static
         self.jid_vers = {}
 
-    def supports(self, jid, node, ifrom, data):
+    async def supports(self, jid, node, ifrom, data):
         """
         Check if a JID supports a given feature.
 
@@ -65,8 +65,8 @@ class StaticCaps(object):
                 return True
 
         try:
-            info = self.disco.get_info(jid=jid, node=node,
+            info = await self.disco.get_info(jid=jid, node=node,
-                                       ifrom=ifrom, **data)
+                                             ifrom=ifrom, **data)
             info = self.disco._wrap(ifrom, jid, info, True)
             return feature in info['disco_info']['features']
         except IqError:
@@ -74,7 +74,7 @@ class StaticCaps(object):
         except IqTimeout:
             return None
 
-    def has_identity(self, jid, node, ifrom, data):
+    async def has_identity(self, jid, node, ifrom, data):
         """
         Check if a JID has a given identity.
 
@@ -110,8 +110,8 @@ class StaticCaps(object):
                 return True
 
         try:
-            info = self.disco.get_info(jid=jid, node=node,
+            info = await self.disco.get_info(jid=jid, node=node,
-                                       ifrom=ifrom, **data)
+                                             ifrom=ifrom, **data)
             info = self.disco._wrap(ifrom, jid, info, True)
             return identity in map(trunc, info['disco_info']['identities'])
         except IqError:

slixmpp/plugins/xep_0128/extended_disco.py

@@ -5,6 +5,7 @@
 # See the file LICENSE for copying permission.
 import logging
 
+from asyncio import Future
 from typing import Optional
 
 import slixmpp
@@ -53,37 +54,46 @@ class XEP_0128(BasePlugin):
         for op in self._disco_ops:
             self.api.register(getattr(self.static, op), op, default=True)
 
-    def set_extended_info(self, jid=None, node=None, **kwargs):
+    def set_extended_info(self, jid=None, node=None, **kwargs) -> Future:
         """
         Set additional, extended identity information to a node.
 
         Replaces any existing extended information.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         :param data: Either a form, or a list of forms to use
                      as extended information, replacing any
                      existing extensions.
         """
-        self.api['set_extended_info'](jid, node, None, kwargs)
+        return self.api['set_extended_info'](jid, node, None, kwargs)
 
-    def add_extended_info(self, jid=None, node=None, **kwargs):
+    def add_extended_info(self, jid=None, node=None, **kwargs) -> Future:
         """
         Add additional, extended identity information to a node.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         :param data: Either a form, or a list of forms to add
                      as extended information.
         """
-        self.api['add_extended_info'](jid, node, None, kwargs)
+        return self.api['add_extended_info'](jid, node, None, kwargs)
 
     def del_extended_info(self, jid: Optional[JID] = None,
-                          node: Optional[str] = None, **kwargs):
+                          node: Optional[str] = None, **kwargs) -> Future:
         """
         Remove all extended identity information to a node.
 
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+
         :param jid: The JID to modify.
         :param node: The node to modify.
         """
-        self.api['del_extended_info'](jid, node, None, kwargs)
+        return self.api['del_extended_info'](jid, node, None, kwargs)

slixmpp/plugins/xep_0153/vcard_avatar.py

@@ -5,7 +5,7 @@
 # See the file LICENSE for copying permission.
 import hashlib
 import logging
-from asyncio import Future, ensure_future
+from asyncio import Future
 from typing import (
     Dict,
     Optional,
@@ -13,7 +13,7 @@ from typing import (
 
 from slixmpp import JID
 from slixmpp.stanza import Presence
-from slixmpp.exceptions import XMPPError, IqTimeout
+from slixmpp.exceptions import XMPPError, IqTimeout, IqError
 from slixmpp.xmlstream import register_stanza_plugin, ElementBase
 from slixmpp.plugins.base import BasePlugin
 from slixmpp.plugins.xep_0153 import stanza, VCardTempUpdate
@@ -59,7 +59,6 @@ class XEP_0153(BasePlugin):
         self.xmpp.del_event_handler('presence_chat', self._recv_presence)
         self.xmpp.del_event_handler('presence_away', self._recv_presence)
 
-    @future_wrapper
     def set_avatar(self, jid: Optional[JID] = None,
                    avatar: Optional[bytes] = None,
                    mtype: Optional[str] = None, **iqkwargs) -> Future:
@@ -97,10 +96,10 @@ class XEP_0153(BasePlugin):
             except IqTimeout as exc:
                 timeout_cb(exc)
                 raise
-            self.api['reset_hash'](jid)
+            await self.api['reset_hash'](jid)
             self.xmpp.roster[jid].send_last_presence()
 
-        return ensure_future(get_and_set_avatar(), loop=self.xmpp.loop)
+        return self.xmpp.wrap(get_and_set_avatar())
 
     async def _start(self, event):
         try:
@@ -110,22 +109,22 @@ class XEP_0153(BasePlugin):
                 new_hash = ''
             else:
                 new_hash = hashlib.sha1(data).hexdigest()
-            self.api['set_hash'](self.xmpp.boundjid, args=new_hash)
+            await self.api['set_hash'](self.xmpp.boundjid, args=new_hash)
         except XMPPError:
             log.debug('Could not retrieve vCard for %s', self.xmpp.boundjid.bare)
 
-    def _update_presence(self, stanza: ElementBase) -> ElementBase:
+    async def _update_presence(self, stanza: ElementBase) -> ElementBase:
         if not isinstance(stanza, Presence):
             return stanza
 
         if stanza['type'] not in ('available', 'dnd', 'chat', 'away', 'xa'):
             return stanza
 
-        current_hash = self.api['get_hash'](stanza['from'])
+        current_hash = await self.api['get_hash'](stanza['from'])
         stanza['vcard_temp_update']['photo'] = current_hash
         return stanza
 
-    def _recv_presence(self, pres: Presence):
+    async def _recv_presence(self, pres: Presence):
         try:
             if pres.get_plugin('muc', check=True):
                 # Don't process vCard avatars for MUC occupants
@@ -135,7 +134,7 @@ class XEP_0153(BasePlugin):
             pass
 
         if not pres.match('presence/vcard_temp_update'):
-            self.api['set_hash'](pres['from'], args=None)
+            await self.api['set_hash'](pres['from'], args=None)
             return
 
         data = pres['vcard_temp_update']['photo']
@@ -145,33 +144,31 @@ class XEP_0153(BasePlugin):
 
     # =================================================================
 
-    def _reset_hash(self, jid: JID, node: str, ifrom: JID, args: Dict):
+    async def _reset_hash(self, jid: JID, node: str, ifrom: JID, args: Dict):
         own_jid = (jid.bare == self.xmpp.boundjid.bare)
         if self.xmpp.is_component:
             own_jid = (jid.domain == self.xmpp.boundjid.domain)
 
-        self.api['set_hash'](jid, args=None)
+        await self.api['set_hash'](jid, args=None)
         if own_jid:
             self.xmpp.roster[jid].send_last_presence()
 
-        def callback(iq):
+        try:
-            if iq['type'] == 'error':
+            iq = await self.xmpp['xep_0054'].get_vcard(jid=jid.bare, ifrom=ifrom)
-                log.debug('Could not retrieve vCard for %s', jid)
+        except (IqError, IqTimeout):
-                return
+            log.debug('Could not retrieve vCard for %s', jid)
-            try:
+            return
-                data = iq['vcard_temp']['PHOTO']['BINVAL']
+        try:
-            except ValueError:
+            data = iq['vcard_temp']['PHOTO']['BINVAL']
-                log.debug('Invalid BINVAL in vCard’s PHOTO for %s:', jid, exc_info=True)
+        except ValueError:
-                data = None
+            log.debug('Invalid BINVAL in vCard’s PHOTO for %s:', jid, exc_info=True)
-            if not data:
+            data = None
-                new_hash = ''
+        if not data:
-            else:
+            new_hash = ''
-                new_hash = hashlib.sha1(data).hexdigest()
+        else:
+            new_hash = hashlib.sha1(data).hexdigest()
 
-            self.api['set_hash'](jid, args=new_hash)
+        await self.api['set_hash'](jid, args=new_hash)
-
-        self.xmpp['xep_0054'].get_vcard(jid=jid.bare, ifrom=ifrom,
-                                        callback=callback)
 
     def _get_hash(self, jid: JID, node: str, ifrom: JID, args: Dict):
         return self._hashes.get(jid.bare, None)

slixmpp/plugins/xep_0231/bob.py

@@ -12,7 +12,7 @@ from typing import Optional
 from slixmpp import future_wrapper, JID
 from slixmpp.stanza import Iq, Message, Presence
 from slixmpp.exceptions import XMPPError
-from slixmpp.xmlstream.handler import Callback
+from slixmpp.xmlstream.handler import CoroutineCallback
 from slixmpp.xmlstream.matcher import StanzaPath
 from slixmpp.xmlstream import register_stanza_plugin
 from slixmpp.plugins.base import BasePlugin
@@ -40,17 +40,17 @@ class XEP_0231(BasePlugin):
         register_stanza_plugin(Presence, BitsOfBinary)
 
         self.xmpp.register_handler(
-            Callback('Bits of Binary - Iq',
+            CoroutineCallback('Bits of Binary - Iq',
                 StanzaPath('iq/bob'),
                 self._handle_bob_iq))
 
         self.xmpp.register_handler(
-            Callback('Bits of Binary - Message',
+            CoroutineCallback('Bits of Binary - Message',
                 StanzaPath('message/bob'),
                 self._handle_bob))
 
         self.xmpp.register_handler(
-            Callback('Bits of Binary - Presence',
+            CoroutineCallback('Bits of Binary - Presence',
                 StanzaPath('presence/bob'),
                 self._handle_bob))
 
@@ -67,13 +67,14 @@ class XEP_0231(BasePlugin):
     def session_bind(self, jid):
         self.xmpp['xep_0030'].add_feature('urn:xmpp:bob')
 
-    def set_bob(self, data: bytes, mtype: str, cid: Optional[str] = None,
+    async def set_bob(self, data: bytes, mtype: str, cid: Optional[str] = None,
-                max_age: Optional[int] = None) -> str:
+                      max_age: Optional[int] = None) -> str:
         """Register a blob of binary data as a BOB.
 
         .. versionchanged:: 1.8.0
             If ``max_age`` is specified, the registered data will be destroyed
             after that time.
+            This function is now a coroutine.
 
         :param data: Data to register.
         :param mtype: Mime Type of the data (e.g. ``image/jpeg``).
@@ -88,29 +89,30 @@ class XEP_0231(BasePlugin):
         bob['data'] = data
         bob['type'] = mtype
         bob['cid'] = cid
-        bob['max_age'] = max_age
+        if max_age is not None:
+            bob['max_age'] = max_age
 
-        self.api['set_bob'](args=bob)
+        await self.api['set_bob'](args=bob)
         # Schedule destruction of the data
         if max_age is not None and max_age > 0:
             self.xmpp.loop.call_later(max_age, self.del_bob,  cid)
         return cid
 
-    @future_wrapper
+    async def get_bob(self, jid: Optional[JID] = None, cid: Optional[str] = None,
-    def get_bob(self, jid: Optional[JID] = None, cid: Optional[str] = None,
+                      cached: bool = True, ifrom: Optional[JID] = None,
-                cached: bool = True, ifrom: Optional[JID] = None,
+                      **iqkwargs) -> Iq:
-                **iqkwargs) -> Future:
         """Get a BOB.
 
         .. versionchanged:: 1.8.0
             Results not in cache do not raise an error when ``cached`` is True.
+            This function is now a coroutine.
 
         :param jid: JID to fetch the BOB from.
         :param cid: Content ID (actually required).
         :param cached: To fetch the BOB from the local cache first (from CID only)
         """
         if cached:
-            data = self.api['get_bob'](None, None, ifrom, args=cid)
+            data = await self.api['get_bob'](None, None, ifrom, args=cid)
             if data is not None:
                 if not isinstance(data, Iq):
                     iq = self.xmpp.Iq()
@@ -120,19 +122,24 @@ class XEP_0231(BasePlugin):
 
         iq = self.xmpp.make_iq_get(ito=jid, ifrom=ifrom)
         iq['bob']['cid'] = cid
-        return iq.send(**iqkwargs)
+        return await iq.send(**iqkwargs)
 
-    def del_bob(self, cid: str):
+    def del_bob(self, cid: str) -> Future:
-        self.api['del_bob'](args=cid)
+        """Delete a stored BoB.
 
-    def _handle_bob_iq(self, iq: Iq):
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
+        return self.api['del_bob'](args=cid)
+
+    async def _handle_bob_iq(self, iq: Iq):
         cid = iq['bob']['cid']
 
         if iq['type'] == 'result':
-            self.api['set_bob'](iq['from'], None, iq['to'], args=iq['bob'])
+            await self.api['set_bob'](iq['from'], None, iq['to'], args=iq['bob'])
             self.xmpp.event('bob', iq)
         elif iq['type'] == 'get':
-            data = self.api['get_bob'](iq['to'], None, iq['from'], args=cid)
+            data = await self.api['get_bob'](iq['to'], None, iq['from'], args=cid)
             if isinstance(data, Iq):
                 data['id'] = iq['id']
                 data.send()
@@ -142,9 +149,11 @@ class XEP_0231(BasePlugin):
             iq.append(data)
             iq.send()
 
-    def _handle_bob(self, stanza):
+    async def _handle_bob(self, stanza):
-        self.api['set_bob'](stanza['from'], None,
+        await self.api['set_bob'](
-                            stanza['to'], args=stanza['bob'])
+            stanza['from'], None,
+            stanza['to'], args=stanza['bob']
+        )
         self.xmpp.event('bob', stanza)
 
     # =================================================================

slixmpp/plugins/xep_0231/stanza.py

@@ -18,10 +18,14 @@ class BitsOfBinary(ElementBase):
     interfaces = {'cid', 'max_age', 'type', 'data'}
 
     def get_max_age(self):
-        return int(self._get_attr('max-age'))
+        try:
+            return int(self._get_attr('max-age'))
+        except ValueError:
+            return None
 
     def set_max_age(self, value):
-        self._set_attr('max-age', str(value))
+        if value is not None:
+            self._set_attr('max-age', str(value))
 
     def get_data(self):
         return base64.b64decode(bytes(self.xml.text))

slixmpp/plugins/xep_0319/idle.py

@@ -3,8 +3,11 @@
 # Copyright (C) 2013 Nathanael C. Fritz, Lance J.T. Stout
 # This file is part of Slixmpp.
 # See the file LICENSE for copying permission.
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timezone
 
+from typing import Optional
+
+from slixmpp import JID
 from slixmpp.stanza import Presence
 from slixmpp.plugins import BasePlugin
 from slixmpp.xmlstream import register_stanza_plugin
@@ -26,16 +29,13 @@ class XEP_0319(BasePlugin):
     def plugin_init(self):
         self._idle_stamps = {}
         register_stanza_plugin(Presence, stanza.Idle)
-        self.api.register(self._set_idle,
+        self.api.register(self._set_idle, 'set_idle', default=True)
-                'set_idle',
+        self.api.register(self._get_idle, 'get_idle', default=True)
-                default=True)
+        self.xmpp.register_handler(Callback(
-        self.api.register(self._get_idle,
+            'Idle Presence',
-                'get_idle',
+            StanzaPath('presence/idle'),
-                default=True)
+            self._idle_presence
-        self.xmpp.register_handler(
+        ))
-                Callback('Idle Presence',
-                    StanzaPath('presence/idle'),
-                    self._idle_presence))
         self.xmpp.add_filter('out', self._stamp_idle_presence)
 
     def session_bind(self, jid):
@@ -46,19 +46,30 @@ class XEP_0319(BasePlugin):
         self.xmpp.del_filter('out', self._stamp_idle_presence)
         self.xmpp.remove_handler('Idle Presence')
 
-    def idle(self, jid=None, since=None):
+    async def idle(self, jid: Optional[JID] = None,
+                   since: Optional[datetime] = None):
+        """Set an idle duration for a JID
+
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+        """
         seconds = None
         timezone = get_local_timezone()
         if since is None:
             since = datetime.now(timezone)
         else:
             seconds = datetime.now(timezone) - since
-        self.api['set_idle'](jid, None, None, since)
+        await self.api['set_idle'](jid, None, None, since)
-        self.xmpp['xep_0012'].set_last_activity(jid=jid, seconds=seconds)
+        await self.xmpp['xep_0012'].set_last_activity(jid=jid, seconds=seconds)
 
-    def active(self, jid=None):
+    async def active(self, jid: Optional[JID] = None):
-        self.api['set_idle'](jid, None, None, None)
+        """Reset the idle timer.
-        self.xmpp['xep_0012'].del_last_activity(jid)
+
+        .. versionchanged:: 1.8.0
+            This function is now a coroutine.
+        """
+        await self.api['set_idle'](jid, None, None, None)
+        await self.xmpp['xep_0012'].del_last_activity(jid)
 
     def _set_idle(self, jid, node, ifrom, data):
         self._idle_stamps[jid] = data
@@ -69,9 +80,9 @@ class XEP_0319(BasePlugin):
     def _idle_presence(self, pres):
         self.xmpp.event('presence_idle', pres)
 
-    def _stamp_idle_presence(self, stanza):
+    async def _stamp_idle_presence(self, stanza):
         if isinstance(stanza, Presence):
-            since = self.api['get_idle'](stanza['from'] or self.xmpp.boundjid)
+            since = await self.api['get_idle'](stanza['from'] or self.xmpp.boundjid)
             if since:
                 stanza['idle']['since'] = since
         return stanza

slixmpp/xmlstream/xmlstream.py

@@ -9,6 +9,7 @@
 # :license: MIT, see LICENSE for more details
 from typing import (
     Any,
+    Coroutine,
     Callable,
     Iterable,
     Iterator,
@@ -1251,3 +1252,13 @@ class XMLStream(asyncio.BaseProtocol):
             raise
         finally:
             self.del_event_handler(event, handler)
+
+    def wrap(self, coroutine: Coroutine[Any, Any, Any]) -> Future:
+        """Make a Future out of a coroutine with the current loop.
+
+        :param coroutine: The coroutine to wrap.
+        """
+        return asyncio.ensure_future(
+            coroutine,
+            loop=self.loop,
+        )

tests/test_stream_xep_0030.py

@@ -1,5 +1,5 @@
+import asyncio
 import time
-import threading
 
 import unittest
 from slixmpp.test import SlixTest
@@ -288,7 +288,9 @@ class TestStreamDisco(SlixTest):
 
         self.xmpp.add_event_handler('disco_info', handle_disco_info)
 
-        self.xmpp['xep_0030'].get_info('user@localhost', 'foo')
+
+        self.xmpp.wrap(self.xmpp['xep_0030'].get_info('user@localhost', 'foo'))
+        self.wait_()
 
         self.send("""
           <iq type="get" to="user@localhost" id="1">
@@ -483,7 +485,8 @@ class TestStreamDisco(SlixTest):
 
         self.xmpp.add_event_handler('disco_items', handle_disco_items)
 
-        self.xmpp['xep_0030'].get_items('user@localhost', 'foo')
+        self.xmpp.wrap(self.xmpp['xep_0030'].get_items('user@localhost', 'foo'))
+        self.wait_()
 
         self.send("""
           <iq type="get" to="user@localhost" id="1">

tests/test_stream_xep_0047.py

@@ -14,7 +14,7 @@ class TestInBandByteStreams(SlixTest):
     def tearDown(self):
         self.stream_close()
 
-    async def testOpenStream(self):
+    def testOpenStream(self):
         """Test requesting a stream, successfully"""
 
         events = []
@@ -25,8 +25,9 @@ class TestInBandByteStreams(SlixTest):
 
         self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
 
-        await self.xmpp['xep_0047'].open_stream('tester@localhost/receiver',
+        self.xmpp.wrap(self.xmpp['xep_0047'].open_stream('tester@localhost/receiver',
-                                                sid='testing')
+                                                         sid='testing'))
+        self.wait_()
 
         self.send("""
           <iq type="set" to="tester@localhost/receiver" id="1">
@@ -45,7 +46,7 @@ class TestInBandByteStreams(SlixTest):
 
         self.assertEqual(events, ['ibb_stream_start'])
 
-    async def testAysncOpenStream(self):
+    def testAysncOpenStream(self):
         """Test requesting a stream, aysnc"""
 
         events = set()
@@ -58,9 +59,10 @@ class TestInBandByteStreams(SlixTest):
 
         self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
 
-        await self.xmpp['xep_0047'].open_stream('tester@localhost/receiver',
+        self.xmpp.wrap(self.xmpp['xep_0047'].open_stream('tester@localhost/receiver',
-                                                sid='testing',
+                                                         sid='testing',
-                                                callback=stream_callback)
+                                                         callback=stream_callback))
+        self.wait_()
 
         self.send("""
           <iq type="set" to="tester@localhost/receiver" id="1">
@@ -79,7 +81,7 @@ class TestInBandByteStreams(SlixTest):
 
         self.assertEqual(events, {'ibb_stream_start', 'callback'})
 
-    async def testSendData(self):
+    def testSendData(self):
         """Test sending data over an in-band bytestream."""
 
         streams = []
@@ -89,13 +91,14 @@ class TestInBandByteStreams(SlixTest):
             streams.append(stream)
 
         def on_stream_data(d):
-            data.append(d['data'])
+            data.append(d.read())
 
         self.xmpp.add_event_handler('ibb_stream_start', on_stream_start)
         self.xmpp.add_event_handler('ibb_stream_data', on_stream_data)
 
-        self.xmpp['xep_0047'].open_stream('tester@localhost/receiver',
+        self.xmpp.wrap(self.xmpp['xep_0047'].open_stream('tester@localhost/receiver',
-                                          sid='testing')
+                                                         sid='testing'))
+        self.wait_()
 
         self.send("""
           <iq type="set" to="tester@localhost/receiver" id="1">
@@ -116,7 +119,8 @@ class TestInBandByteStreams(SlixTest):
 
 
         # Test sending data out
-        await stream.send("Testing")
+        self.xmpp.wrap(stream.send("Testing"))
+        self.wait_()
 
         self.send("""
           <iq type="set" id="2"

tests/test_stream_xep_0077.py

@@ -91,7 +91,9 @@ class TestRegistration(SlixTest):
         self.send("<iq type='result' id='reg2' from='shakespeare.lit' to='bill@shakespeare.lit/globe'/>")
         pseudo_iq = self.xmpp.Iq()
         pseudo_iq["from"] = "bill@shakespeare.lit/globe"
-        user = self.xmpp["xep_0077"].api["user_get"](None, None, None, pseudo_iq)
+        fut = self.xmpp.wrap(self.xmpp["xep_0077"].api["user_get"](None, None, None, pseudo_iq))
+        self.run_coro(fut)
+        user = fut.result()
         self.assertEqual(user["username"], "bill")
         self.assertEqual(user["password"], "Calliope")
         self.recv(