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

Make the internal "api" async

See merge request poezio/slixmpp!128

18 files changed, 553 insertions, 336 deletions

diff --git a/slixmpp/api.py b/slixmpp/api.py
index f09e0365..39fed490 100644
--- a/slixmpp/api.py
+++ b/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 string op: The API operation to initialize.
+        :param ctype: The name of the API 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
-        ==============================
-        Given | Given | Node handler
-        Given | None  | JID handler
-        None  | None  | Global handler
+        ====== ======= ===================
+        JID     node    Handler
+        ====== ======= ===================
+        Given   Given   Node + JID 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.
-        :param string op: The API operation to perform.
-        :param JID jid: Optionally provide specific JID.
-        :param string node: Optionally provide specific node.
-        :param JID ifrom: Optionally provide the requesting JID.
-        :param tuple args: Optional positional arguments to the handler.
+        .. versionchanged:: 1.8.0
+            ``run()`` always returns a future, if the handler is a coroutine
+            the future should be awaited on.
+
+        :param ctype: The name of the API to use.
+        :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 string op: The API operation to perform.
-        :param JID jid: Optionally provide specific JID.
-        :param string node: Optionally provide specific node.
+        :param ctype: The name of the API to use.
+        :param op: The API operation to perform.
+        :param jid: Optionally provide specific JID.
+        :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 string ctype: The name of the API to modify.
-        :param string op: The API operation to use.
+        :param handler: The default, global handler for the operation.
+        :param ctype: The name of the API to modify.
+        :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 string op: The API operation to perform.
-        :param JID jid: Optionally provide specific JID.
-        :param string node: Optionally provide specific node.
+        :param ctype: The name of the API to use.
+        :param op: The API operation to perform.
+        :param jid: Optionally provide specific JID.
+        :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 string op: The API operation to perform.
-        :param JID jid: Optionally provide specific JID.
-        :param string node: Optionally provide specific node.
+        :param ctype: The name of the API to use.
+        :param op: The API operation to perform.
+        :param jid: Optionally provide specific JID.
+        :param node: Optionally provide specific node.
         """
         self.unregister(ctype, op, jid, node)
         self.register(self._handler_defaults[ctype][op], ctype, op, jid, node)
diff --git a/slixmpp/plugins/xep_0012/last_activity.py b/slixmpp/plugins/xep_0012/last_activity.py
index 27e16e21..61531431 100644
--- a/slixmpp/plugins/xep_0012/last_activity.py
+++ b/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: Optional[JID] = None) -> Future:
+        """Remove the last activity of a JID.
 
-    def end_idle(self, jid=None):
-        self.del_last_activity(jid)
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
+        return self.del_last_activity(jid)
 
-    def start_uptime(self, status=None):
-        self.set_last_activity(None, 0, status)
+    def start_uptime(self, status: Optional[str] = None) -> Future:
+        """
+        .. 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):
-        self.api['set_last_activity'](jid, args={
+    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):
-        self.api['del_last_activity'](jid)
+    def del_last_activity(self, jid: JID) -> Future:
+        """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()
 
     # =================================================================
diff --git a/slixmpp/plugins/xep_0027/gpg.py b/slixmpp/plugins/xep_0027/gpg.py
index af5df044..61da7ff0 100644
--- a/slixmpp/plugins/xep_0027/gpg.py
+++ b/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.xmlstream import ElementBase, register_stanza_plugin
+from slixmpp.plugins.base import BasePlugin
+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):
-        self.api['set_keyid'](jid, args=keyid)
+    def set_keyid(self, jid=None, keyid=None) -> Future:
+        """Set a keyid for a specific JID.
+
+        .. 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.
 
-    def get_keyid(self, jid=None):
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
         return self.api['get_keyid'](jid)
 
-    def del_keyid(self, jid=None):
-        self.api['del_keyid'](jid)
+    def del_keyid(self, jid=None) -> Future:
+        """Delete a keyid.
+
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
+        return self.api['del_keyid'](jid)
+
+    def get_keyids(self) -> Future:
+        """Get stored keyids.
 
-    def get_keyids(self):
+        .. versionchanged:: 1.8.0
+            This function now returns a Future.
+        """
         return self.api['get_keyids']()
 
     def _handle_signed_presence(self, pres):
diff --git a/slixmpp/plugins/xep_0030/disco.py b/slixmpp/plugins/xep_0030/disco.py
index 9c4c5269..6aeadc84 100644
--- a/slixmpp/plugins/xep_0030/disco.py
+++ b/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
-    def get_info(self, jid=None, node=None, local=None,
+    async 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,
-                    kwargs.get('ifrom', None),
-                    kwargs)
+            info = await self.api['get_info'](
+                jid, node, kwargs.get('ifrom', None),
+                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,
-                    kwargs.get('ifrom', None),
-                    kwargs)
+            info = await self.api['get_cached_info'](
+                jid, node,
+                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
-    def get_items(self, jid=None, node=None, local=False, **kwargs):
+    async 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'],
-                                        iq['disco_info']['node'],
-                                        iq['from'],
-                                        iq)
+            info = await self.api['get_info'](iq['to'],
+                                              iq['disco_info']['node'],
+                                              iq['from'],
+                                              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'],
-                                       iq['disco_info']['node'],
-                                       ito,
-                                       iq)
+                await self.api['cache_info'](iq['from'],
+                                             iq['disco_info']['node'],
+                                             ito,
+                                             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'],
-                                          iq['disco_items']['node'],
-                                          iq['from'],
-                                          iq)
+            items = await self.api['get_items'](iq['to'],
+                                                iq['disco_items']['node'],
+                                                iq['from'],
+                                                iq)
             if isinstance(items, Iq):
                 items.send()
             else:
diff --git a/slixmpp/plugins/xep_0030/static.py b/slixmpp/plugins/xep_0030/static.py
index 1b5ff2d8..1ae34148 100644
--- a/slixmpp/plugins/xep_0030/static.py
+++ b/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,
-                                       ifrom=ifrom, **data)
+            info = await self.disco.get_info(jid=jid, node=node,
+                                             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,
-                                       ifrom=ifrom, **data)
+            info = await self.disco.get_info(jid=jid, node=node,
+                                             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'])
diff --git a/slixmpp/plugins/xep_0047/ibb.py b/slixmpp/plugins/xep_0047/ibb.py
index ec08a8b3..6d4ab71f 100644
--- a/slixmpp/plugins/xep_0047/ibb.py
+++ b/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')
diff --git a/slixmpp/plugins/xep_0054/vcard_temp.py b/slixmpp/plugins/xep_0054/vcard_temp.py
index bee20ce0..460013b8 100644
--- a/slixmpp/plugins/xep_0054/vcard_temp.py
+++ b/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
-    def get_vcard(self, jid: Optional[JID] = None, *,
-                  local: Optional[bool] = None, cached: bool = False,
-                  ifrom: Optional[JID] = None,
-                  **iqkwargs) -> Future:
+    async def get_vcard(self, jid: Optional[JID] = None, *,
+                        local: Optional[bool] = None, cached: bool = False,
+                        ifrom: Optional[JID] = None,
+                        **iqkwargs) -> Iq:
         """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
-    def publish_vcard(self, vcard: Optional[VCardTemp] = None,
-                      jid: Optional[JID] = None,
-                      ifrom: Optional[JID] = None, **iqkwargs) -> Future:
+    async def publish_vcard(self, vcard: Optional[VCardTemp] = None,
+                            jid: Optional[JID] = None,
+                            ifrom: Optional[JID] = None, **iqkwargs):
         """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:
diff --git a/slixmpp/plugins/xep_0065/proxy.py b/slixmpp/plugins/xep_0065/proxy.py
index e1ca4096..cf4edfe1 100644
--- a/slixmpp/plugins/xep_0065/proxy.py
+++ b/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(
-            Callback('Socks5 Bytestreams',
-                     StanzaPath('iq@type=set/socks/streamhost'),
-                     self._handle_streamhost))
+        self.xmpp.register_handler(CoroutineCallback(
+            'Socks5 Bytestreams',
+             StanzaPath('iq@type=set/socks/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(*futures, return_exceptions=True)
-            for streamhost, proxy in zip(streamhosts, proxies):
-                if isinstance(proxy, ValueError):
-                    continue
-                elif isinstance(proxy, socket.error):
-                    log.error('Socket error while connecting to the proxy.')
-                    continue
-                proxy = proxy[1]
-                # TODO: what if the future never happens?
-                try:
-                    addr, port = await proxy.connected
-                except socket.error:
-                    log.exception('Socket error while connecting to the proxy.')
-                    continue
-                # TODO: make a better choice than just the first working one.
-                used_streamhost = streamhost['jid']
-                conn = proxy
-                break
-            else:
-                raise XMPPError(etype='cancel', condition='item-not-found')
+        proxies = await asyncio.gather(*proxy_futures, return_exceptions=True)
+        for streamhost, proxy in zip(streamhosts, proxies):
+            if isinstance(proxy, ValueError):
+                continue
+            elif isinstance(proxy, socket.error):
+                log.error('Socket error while connecting to the proxy.')
+                continue
+            proxy = proxy[1]
+            # TODO: what if the future never happens?
+            try:
+                addr, port = await proxy.connected
+            except socket.error:
+                log.exception('Socket error while connecting to the proxy.')
+                continue
+            # TODO: make a better choice than just the first working one.
+            used_streamhost = streamhost['jid']
+            conn = proxy
+            break
+        else:
+            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()
-            self._sessions[sid] = conn
-            iq['socks']['sid'] = sid
-            iq['socks']['streamhost_used']['jid'] = used_streamhost
-            iq.send()
-            self.xmpp.event('socks5_stream', conn)
-            self.xmpp.event('stream:%s:%s' % (sid, requester), conn)
+        iq = iq.reply()
+        self._sessions[sid] = conn
+        iq['socks']['sid'] = sid
+        iq['socks']['streamhost_used']['jid'] = used_streamhost
+        iq.send()
+        self.xmpp.event('socks5_stream', 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
diff --git a/slixmpp/plugins/xep_0077/register.py b/slixmpp/plugins/xep_0077/register.py
index 1850b2c9..c5d1fa27 100644
--- a/slixmpp/plugins/xep_0077/register.py
+++ b/slixmpp/plugins/xep_0077/register.py
@@ -29,7 +29,7 @@ class XEP_0077(BasePlugin):
         user_register           -- After succesful validation and add to the user store
                                    in api["user_validate"]
         user_unregister         -- After succesful user removal in api["user_remove"]
-    
+
     Config:
 
     ::
@@ -38,7 +38,7 @@ class XEP_0077(BasePlugin):
         in case api["make_registration_form"] is not overriden.
 
     API:
-    
+
     ::
 
         user_get(jid, node, ifrom, iq)
@@ -102,14 +102,13 @@ class XEP_0077(BasePlugin):
 
     def _user_get(self, jid, node, ifrom, iq):
         return self._user_store.get(iq["from"].bare)
-    
+
     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):
-        reply = self.api["make_registration_form"](None, None, iq["from"], iq)
+    async def _send_form(self, iq):
+        reply = await self.api["make_registration_form"](None, None, iq["from"], iq)
         reply.send()
 
     def _force_registration(self, event):
diff --git a/slixmpp/plugins/xep_0095/stream_initiation.py b/slixmpp/plugins/xep_0095/stream_initiation.py
index c7a86780..fd82711a 100644
--- a/slixmpp/plugins/xep_0095/stream_initiation.py
+++ b/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):
-        stream = self.api['get_pending'](ifrom, sid, jid)
+    async def accept(self, jid, sid, payload=None, ifrom=None, stream_handler=None):
+        """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()
+
+    async def decline(self, jid, sid, ifrom=None):
+        """Decline a stream initiation.
 
-    def decline(self, jid, sid, ifrom=None):
-        stream = self.api['get_pending'](ifrom, sid, jid)
+        .. 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)
-        return iq.send()
+        await self.api['del_pending'](ifrom, sid, jid)
+        return await iq.send()
 
     def _add_pending(self, jid, node, ifrom, data):
         with self._pending_lock:
diff --git a/slixmpp/plugins/xep_0115/caps.py b/slixmpp/plugins/xep_0115/caps.py
index 5f71165c..75c96410 100644
--- a/slixmpp/plugins/xep_0115/caps.py
+++ b/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'],
-                                         pres['caps']['ver']):
-                self.assign_verstring(pres['from'], pres['caps']['ver'])
+            if await self._validate_caps(caps, pres['caps']['hash'],
+                                               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(
-                    jid=jid,
-                    node='%s#%s' % (self.caps_node, ver),
-                    info=info)
-            self.cache_caps(ver, info)
-            self.assign_verstring(jid, ver)
+            await self.xmpp['xep_0030'].set_info(
+                jid=jid,
+                node='%s#%s' % (self.caps_node, ver),
+                info=info
+            )
+            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)
diff --git a/slixmpp/plugins/xep_0115/static.py b/slixmpp/plugins/xep_0115/static.py
index 2461d2e3..74f2beb8 100644
--- a/slixmpp/plugins/xep_0115/static.py
+++ b/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,
-                                       ifrom=ifrom, **data)
+            info = await self.disco.get_info(jid=jid, node=node,
+                                             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,
-                                       ifrom=ifrom, **data)
+            info = await self.disco.get_info(jid=jid, node=node,
+                                             ifrom=ifrom, **data)
             info = self.disco._wrap(ifrom, jid, info, True)
             return identity in map(trunc, info['disco_info']['identities'])
         except IqError:
diff --git a/slixmpp/plugins/xep_0128/extended_disco.py b/slixmpp/plugins/xep_0128/extended_disco.py
index d0264caf..759e5efe 100644
--- a/slixmpp/plugins/xep_0128/extended_disco.py
+++ b/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)
diff --git a/slixmpp/plugins/xep_0153/vcard_avatar.py b/slixmpp/plugins/xep_0153/vcard_avatar.py
index 56bf899a..e2d98b0a 100644
--- a/slixmpp/plugins/xep_0153/vcard_avatar.py
+++ b/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):
-            if iq['type'] == 'error':
-                log.debug('Could not retrieve vCard for %s', jid)
-                return
-            try:
-                data = iq['vcard_temp']['PHOTO']['BINVAL']
-            except ValueError:
-                log.debug('Invalid BINVAL in vCard’s PHOTO for %s:', jid, exc_info=True)
-                data = None
-            if not data:
-                new_hash = ''
-            else:
-                new_hash = hashlib.sha1(data).hexdigest()
-
-            self.api['set_hash'](jid, args=new_hash)
-
-        self.xmpp['xep_0054'].get_vcard(jid=jid.bare, ifrom=ifrom,
-                                        callback=callback)
+        try:
+            iq = await self.xmpp['xep_0054'].get_vcard(jid=jid.bare, ifrom=ifrom)
+        except (IqError, IqTimeout):
+            log.debug('Could not retrieve vCard for %s', jid)
+            return
+        try:
+            data = iq['vcard_temp']['PHOTO']['BINVAL']
+        except ValueError:
+            log.debug('Invalid BINVAL in vCard’s PHOTO for %s:', jid, exc_info=True)
+            data = None
+        if not data:
+            new_hash = ''
+        else:
+            new_hash = hashlib.sha1(data).hexdigest()
+
+        await self.api['set_hash'](jid, args=new_hash)
 
     def _get_hash(self, jid: JID, node: str, ifrom: JID, args: Dict):
         return self._hashes.get(jid.bare, None)
diff --git a/slixmpp/plugins/xep_0231/bob.py b/slixmpp/plugins/xep_0231/bob.py
index e554c38c..30722208 100644
--- a/slixmpp/plugins/xep_0231/bob.py
+++ b/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,
-                max_age: Optional[int] = None) -> str:
+    async def set_bob(self, data: bytes, mtype: str, cid: Optional[str] = None,
+                      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
-    def get_bob(self, jid: Optional[JID] = None, cid: Optional[str] = None,
-                cached: bool = True, ifrom: Optional[JID] = None,
-                **iqkwargs) -> Future:
+    async def get_bob(self, jid: Optional[JID] = None, cid: Optional[str] = None,
+                      cached: bool = True, ifrom: Optional[JID] = None,
+                      **iqkwargs) -> Iq:
         """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):
-        self.api['del_bob'](args=cid)
+    def del_bob(self, cid: str) -> Future:
+        """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):
-        self.api['set_bob'](stanza['from'], None,
-                            stanza['to'], args=stanza['bob'])
+    async def _handle_bob(self, stanza):
+        await self.api['set_bob'](
+            stanza['from'], None,
+            stanza['to'], args=stanza['bob']
+        )
         self.xmpp.event('bob', stanza)
 
     # =================================================================
diff --git a/slixmpp/plugins/xep_0231/stanza.py b/slixmpp/plugins/xep_0231/stanza.py
index 809253d4..6bde671b 100644
--- a/slixmpp/plugins/xep_0231/stanza.py
+++ b/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))
diff --git a/slixmpp/plugins/xep_0319/idle.py b/slixmpp/plugins/xep_0319/idle.py
index 14dd7f4c..3b712967 100644
--- a/slixmpp/plugins/xep_0319/idle.py
+++ b/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,
-                'set_idle',
-                default=True)
-        self.api.register(self._get_idle,
-                'get_idle',
-                default=True)
-        self.xmpp.register_handler(
-                Callback('Idle Presence',
-                    StanzaPath('presence/idle'),
-                    self._idle_presence))
+        self.api.register(self._set_idle, 'set_idle', default=True)
+        self.api.register(self._get_idle, 'get_idle', default=True)
+        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)
-        self.xmpp['xep_0012'].set_last_activity(jid=jid, seconds=seconds)
+        await self.api['set_idle'](jid, None, None, since)
+        await self.xmpp['xep_0012'].set_last_activity(jid=jid, seconds=seconds)
+
+    async def active(self, jid: Optional[JID] = None):
+        """Reset the idle timer.
 
-    def active(self, jid=None):
-        self.api['set_idle'](jid, None, None, None)
-        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
diff --git a/slixmpp/xmlstream/xmlstream.py b/slixmpp/xmlstream/xmlstream.py
index 2f506018..02f4598c 100644
--- a/slixmpp/xmlstream/xmlstream.py
+++ b/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,
+        )