Merge pull request CloudBotIRC#124 from linuxdaemon/gonzobot+asyncio-threadsafe-wrap

Fix various version compatibility wrappers

Author: edwardslabs

cloudbot/__main__.py

@@ -3,8 +3,8 @@
 import signal
 import sys
 import time
+
 # store the original working directory, for use when restarting
-from functools import partial
 
 original_wd = os.path.realpath(".")
 
@@ -47,9 +47,7 @@ def exit_gracefully(signum, frame):
             # we are currently in the process of restarting
             stopped_while_restarting = True
         else:
-            _bot.loop.call_soon_threadsafe(
-                partial(async_util.wrap_future, _bot.stop("Killed (Received SIGINT {})".format(signum)), loop=_bot.loop)
-            )
+            async_util.run_coroutine_threadsafe(_bot.stop("Killed (Received SIGINT {})".format(signum)), _bot.loop)
 
         logger.warning("Bot received Signal Interrupt ({})".format(signum))
 

cloudbot/bot.py

@@ -20,7 +20,7 @@
 from cloudbot.hook import Action
 from cloudbot.plugin import PluginManager
 from cloudbot.reloader import PluginReloader, ConfigReloader
-from cloudbot.util import database, formatting
+from cloudbot.util import database, formatting, async_util
 
 try:
     from cloudbot.web.main import WebInterface
@@ -65,7 +65,7 @@ def __init__(self, loop=asyncio.get_event_loop()):
         self.start_time = time.time()
         self.running = True
         # future which will be called when the bot stopsIf you
-        self.stopped_future = asyncio.Future(loop=self.loop)
+        self.stopped_future = async_util.create_future(self.loop)
 
         # stores each bot server connection
         self.connections = {}

cloudbot/clients/irc.py

@@ -4,6 +4,7 @@
 import re
 import ssl
 from _ssl import PROTOCOL_SSLv23
+from functools import partial
 from ssl import SSLContext
 
 from cloudbot.client import Client
@@ -141,7 +142,7 @@ def connect(self, timeout=None):
             optional_params["local_addr"] = self.local_bind
 
         coro = self.loop.create_connection(
-            lambda: _IrcProtocol(self), host=self.server, port=self.port, ssl=self.ssl_context, **optional_params
+            partial(_IrcProtocol, self), host=self.server, port=self.port, ssl=self.ssl_context, **optional_params
         )
 
         if timeout is not None:
@@ -292,7 +293,7 @@ def __init__(self, conn):
         self._transport = None
 
         # Future that waits until we are connected
-        self._connected_future = asyncio.Future(loop=self.loop)
+        self._connected_future = async_util.create_future(self.loop)
 
     def connection_made(self, transport):
         self._transport = transport
@@ -304,7 +305,7 @@ def connection_made(self, transport):
     def connection_lost(self, exc):
         self._connected = False
         # create a new connected_future for when we are connected.
-        self._connected_future = asyncio.Future(loop=self.loop)
+        self._connected_future = async_util.create_future(self.loop)
         if exc is None:
             # we've been closed intentionally, so don't reconnect
             return
@@ -314,7 +315,7 @@ def connection_lost(self, exc):
     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)
+        self._connected_future = async_util.create_future(self.loop)
         logger.info("[{}] EOF received.".format(self.conn.name))
         async_util.wrap_future(self.conn.connect(), loop=self.loop)
         return True

cloudbot/plugin.py

@@ -597,7 +597,7 @@ def launch(self, hook, event):
                     self._hook_waiting_queues[key] = queue
                 assert isinstance(queue, asyncio.Queue)
                 # create a future to represent this task
-                future = asyncio.Future()
+                future = async_util.create_future(self.bot.loop)
                 queue.put_nowait(future)
                 # wait until the last task is completed
                 yield from future

cloudbot/reloader.py

@@ -1,6 +1,5 @@
 import asyncio
 from abc import ABC
-from functools import partial
 from pathlib import Path
 
 from watchdog.events import PatternMatchingEventHandler
@@ -47,14 +46,14 @@ def reload(self, path):
         """
         path = Path(path).resolve()
         if path.exists():
-            self.bot.loop.call_soon_threadsafe(partial(async_util.wrap_future, self._reload(path), loop=self.bot.loop))
+            async_util.run_coroutine_threadsafe(self._reload(path), self.bot.loop)
 
     def unload(self, path):
         """
         Unloads a module, given its file path. Thread safe.
         """
         path = Path(path).resolve()
-        self.bot.loop.call_soon_threadsafe(partial(async_util.wrap_future, self._unload(path), loop=self.bot.loop))
+        async_util.run_coroutine_threadsafe(self._unload(path), self.bot.loop)
 
     @asyncio.coroutine
     def _reload(self, path):
@@ -112,5 +111,5 @@ def on_moved(self, event):
 
 
 class ConfigEventHandler(ReloadHandler):
-    def on_any_event(self, event) -> None:
+    def on_any_event(self, event):
         self.loader.reload(getattr(event, "dest_path", event.src_path))

cloudbot/util/async_util.py

@@ -31,3 +31,28 @@ def run_func(loop, func, *args, **kwargs):
         return (yield from part())
     else:
         return (yield from loop.run_in_executor(None, part))
+
+
+def run_coroutine_threadsafe(coro, loop):
+    """
+    Runs a coroutine in a threadsafe manner
+    :type coro: coroutine
+    :type loop: asyncio.AbstractEventLoop
+    """
+    if not asyncio.iscoroutine(coro):
+        raise TypeError('A coroutine object is required')
+
+    if sys.version_info < (3, 5, 1):
+        loop.call_soon_threadsafe(partial(wrap_future, coro, loop=loop))
+    else:
+        asyncio.run_coroutine_threadsafe(coro, loop)
+
+
+def create_future(loop=None):
+    if loop is None:
+        loop = asyncio.get_event_loop()
+
+    if sys.version_info < (3, 5, 2):
+        return asyncio.Future(loop=loop)
+
+    return loop.create_future()

plugins/core/cap.py

@@ -30,7 +30,7 @@ def handle_available_caps(conn, caplist, event, irc_paramlist, bot):
         ]
         results = yield from asyncio.gather(*tasks)
         if any(ok and (res or res is None) for ok, res in results):
-            cap_queue[name_cf] = conn.loop.create_future()
+            cap_queue[name_cf] = async_util.create_future(conn.loop)
             conn.cmd("CAP", "REQ", cap)
 
     if irc_paramlist[2] != '+':

plugins/sasl.py

@@ -2,6 +2,7 @@
 import base64
 
 from cloudbot import hook
+from cloudbot.util import async_util
 
 
 @hook.on_cap_available("sasl")
@@ -16,14 +17,14 @@ def sasl_ack(conn, logger):
     sasl_auth = conn.config.get('sasl')
     if sasl_auth and sasl_auth.get('enabled', True):
         sasl_mech = sasl_auth.get("mechanism", "PLAIN").upper()
-        auth_fut = conn.loop.create_future()
+        auth_fut = async_util.create_future(conn.loop)
         conn.memory["sasl_auth_future"] = auth_fut
         conn.cmd("AUTHENTICATE", sasl_mech)
         cmd, arg = yield from auth_fut
         if cmd == "908":
             logger.warning("[%s|sasl] SASL mechanism not supported", conn.name)
         elif cmd == "AUTHENTICATE" and arg[0] == '+':
-            num_fut = conn.loop.create_future()
+            num_fut = async_util.create_future(conn.loop)
             conn.memory["sasl_numeric_future"] = num_fut
             if sasl_mech == "PLAIN":
                 auth_str = "{nick}\0{user}\0{passwd}".format(