Add wrapper function for asyncio.async()/asyncio.ensure_future()

Author: linuxdaemon

cloudbot/__main__.py

@@ -6,6 +6,10 @@
 import signal
 
 # store the original working directory, for use when restarting
+from functools import partial
+
+from cloudbot.util import async_util
+
 original_wd = os.path.realpath(".")
 
 # set up environment - we need to make sure we are in the install directory
@@ -47,7 +51,8 @@ def exit_gracefully(signum, frame):
             stopped_while_restarting = True
         else:
             _bot.loop.call_soon_threadsafe(
-                lambda: asyncio.async(_bot.stop("Killed (Received SIGINT {})".format(signum)), loop=_bot.loop))
+                partial(async_util.wrap_future, _bot.stop("Killed (Received SIGINT {})".format(signum)), loop=_bot.loop)
+            )
 
         logger.warning("Bot received Signal Interrupt ({})".format(signum))
 

cloudbot/clients/irc.py

@@ -7,6 +7,7 @@
 
 from cloudbot.client import Client
 from cloudbot.event import Event, EventType
+from cloudbot.util import async_util
 
 logger = logging.getLogger("cloudbot")
 
@@ -218,8 +219,7 @@ def _send(self, line):
         :type line: str
         """
         logger.info("[{}] >> {}".format(self.name, line))
-        asyncio.async(self._protocol.send(line), loop=self.loop)
-
+        async_util.wrap_future(self._protocol.send(line), loop=self.loop)
 
     @property
     def connected(self):
@@ -272,14 +272,14 @@ def connection_lost(self, exc):
             # we've been closed intentionally, so don't reconnect
             return
         logger.error("[{}] Connection lost: {}".format(self.conn.name, exc))
-        asyncio.async(self.conn.connect(), loop=self.loop)
+        async_util.wrap_future(self.conn.connect(), loop=self.loop)
 
     def eof_received(self):
         self._connected = False
         # create a new connected_future for when we are connected.
         self._connected_future = asyncio.Future(loop=self.loop)
         logger.info("[{}] EOF received.".format(self.conn.name))
-        asyncio.async(self.conn.connect(), loop=self.loop)
+        async_util.wrap_future(self.conn.connect(), loop=self.loop)
         return True
 
     @asyncio.coroutine
@@ -339,7 +339,7 @@ def data_received(self, data):
             # Reply to pings immediately
 
             if command == "PING":
-                asyncio.async(self.send("PONG " + command_params[-1]), loop=self.loop)
+                async_util.wrap_future(self.send("PONG " + command_params[-1]), loop=self.loop)
 
             # Parse the command and params
 
@@ -406,7 +406,7 @@ def data_received(self, data):
                           irc_prefix=prefix, irc_command=command, irc_paramlist=command_params, irc_ctcp_text=ctcp_text)
 
             # handle the message, async
-            asyncio.async(self.bot.process(event), loop=self.loop)
+            async_util.wrap_future(self.bot.process(event), loop=self.loop)
 
 # Channel Commands
 # NOTICE #chan :Text

cloudbot/plugin.py

@@ -17,7 +17,7 @@
 
 from cloudbot.event import Event
 from cloudbot.hook import Priority, Action
-from cloudbot.util import database
+from cloudbot.util import database, async_util
 
 logger = logging.getLogger("cloudbot")
 
@@ -204,7 +204,7 @@ def load_plugin(self, path):
             self._log_hook(on_cap_ack_hook)
 
         for periodic_hook in plugin.periodic:
-            task = asyncio.async(self._start_periodic(periodic_hook))
+            task = async_util.wrap_future(self._start_periodic(periodic_hook))
             plugin.tasks.append(task)
             self._log_hook(periodic_hook)
 

cloudbot/reloader.py

@@ -1,9 +1,12 @@
 import asyncio
 import os.path
+from functools import partial
 
 from watchdog.observers import Observer
 from watchdog.events import PatternMatchingEventHandler
 
+from cloudbot.util import async_util
+
 
 class PluginReloader(object):
     def __init__(self, bot):
@@ -38,7 +41,8 @@ def reload(self, path):
 
         if isinstance(path, bytes):
             path = path.decode()
-        self.bot.loop.call_soon_threadsafe(lambda: asyncio.async(self._reload(path), loop=self.bot.loop))
+
+        self.bot.loop.call_soon_threadsafe(partial(async_util.wrap_future, self._reload(path), loop=self.bot.loop))
 
     def unload(self, path):
         """
@@ -48,7 +52,8 @@ def unload(self, path):
         """
         if isinstance(path, bytes):
             path = path.decode()
-        self.bot.loop.call_soon_threadsafe(lambda: asyncio.async(self._unload(path), loop=self.bot.loop))
+
+        self.bot.loop.call_soon_threadsafe(partial(async_util.wrap_future, self._unload(path), loop=self.bot.loop))
 
     @asyncio.coroutine
     def _reload(self, path):

cloudbot/util/async_util.py

@@ -0,0 +1,20 @@
+"""
+Wraps various asyncio functions
+"""
+
+import asyncio
+
+import sys
+
+
+def wrap_future(fut, *, loop=None):
+    """
+    Wraps asyncio.async()/asyncio.ensure_future() depending on the python version
+    :param fut: The awaitable, future, or coroutine to wrap
+    :param loop: The loop to run in
+    :return: The wrapped future
+    """
+    if sys.version_info < (3, 4, 4):
+        return asyncio.async(fut, loop=loop)
+
+    return asyncio.ensure_future(fut, loop=loop)

plugins/geoip.py

@@ -10,6 +10,7 @@
 import geoip2.errors
 
 from cloudbot import hook
+from cloudbot.util import async_util
 
 logger = logging.getLogger("cloudbot")
 
@@ -52,6 +53,7 @@ def update_db():
         return geoip2.database.Reader(PATH)
 
 
+@asyncio.coroutine
 def check_db(loop):
     """
     runs update_db in an executor thread and sets geoip_reader to the result
@@ -68,7 +70,7 @@ def check_db(loop):
 @asyncio.coroutine
 @hook.on_start
 def load_geoip(loop):
-    asyncio.async(check_db(loop), loop=loop)
+    async_util.wrap_future(check_db(loop), loop=loop)
 
 
 @asyncio.coroutine