Repository: ReactiveX/RxPY

Issue:
AsyncIOScheduler for Python 2.7 via Trollius?
It seems possible to make the `AsyncIOScheduler` available for Python 2.7 through the use of [trollius](http://trollius.readthedocs.org/). Perhaps it can be configured through `rx.config`, like the class for `Future` can be?

Discussion:

Patch:
diff --git a/examples/asyncio/toasyncgenerator.py b/examples/asyncio/toasyncgenerator.py
index 076b82c4..9b3c2319 100644
--- a/examples/asyncio/toasyncgenerator.py
+++ b/examples/asyncio/toasyncgenerator.py
@@ -1,6 +1,6 @@
-import asyncio
-
 import rx
+asyncio = rx.config['asyncio']
+
 from rx.concurrency import AsyncIOScheduler
 from rx.observable import Observable
 from rx.internal import extensionmethod
diff --git a/examples/autocomplete/autocomplete_asyncio.py b/examples/autocomplete/autocomplete_asyncio.py
index 6ea4bd91..88cbd0cf 100644
--- a/examples/autocomplete/autocomplete_asyncio.py
+++ b/examples/autocomplete/autocomplete_asyncio.py
@@ -7,7 +7,8 @@ Uses the RxPY AsyncIOScheduler (Python 3.4 is required)
 """
 
 import os
-import asyncio
+import rx
+asyncio = rx.config['asyncio']
 
 from tornado.websocket import WebSocketHandler
 from tornado.web import RequestHandler, StaticFileHandler, Application, url
diff --git a/rx/__init__.py b/rx/__init__.py
index f227089e..84d1f56e 100644
--- a/rx/__init__.py
+++ b/rx/__init__.py
@@ -1,3 +1,11 @@
+try:
+    import asyncio
+except ImportError:
+    try:
+        import trollius as asyncio
+    except ImportError:
+        asyncio = None
+
 try:
     from threading import Lock
 except ImportError:
@@ -6,12 +14,17 @@ except ImportError:
 try:
     from asyncio import Future
 except ImportError:
-    Future = None
+    try:
+        from trollius import Future
+    except ImportError:
+        Future = None
+
 
 # Rx configuration dictionary
 config = {
     "Future": Future,
-    "Lock": Lock
+    "Lock": Lock,
+    "asyncio": asyncio
 }
 
 from .observable import Observable
diff --git a/rx/concurrency/mainloopscheduler/asyncioscheduler.py b/rx/concurrency/mainloopscheduler/asyncioscheduler.py
index 0dd0a2de..6e1ee5d8 100644
--- a/rx/concurrency/mainloopscheduler/asyncioscheduler.py
+++ b/rx/concurrency/mainloopscheduler/asyncioscheduler.py
@@ -16,7 +16,9 @@ class AsyncIOScheduler(Scheduler):
 
     def __init__(self, loop=None):
         global asyncio
-        import asyncio
+        import rx
+        asyncio = rx.config['asyncio']
+
         self.loop = loop or asyncio.get_event_loop()
 
     def schedule(self, action, state=None):


Test patch:
diff --git a/tests/test_concurrency/test_mainloopscheduler/py2_asyncioscheduler.py b/tests/test_concurrency/test_mainloopscheduler/py2_asyncioscheduler.py
new file mode 100644
index 00000000..60d32c45
--- /dev/null
+++ b/tests/test_concurrency/test_mainloopscheduler/py2_asyncioscheduler.py
@@ -0,0 +1,86 @@
+from nose import SkipTest
+
+import rx
+asyncio = rx.config['asyncio']
+if asyncio is None:
+    raise SkipTest("asyncio not available")
+
+try:
+    from trollius import From
+except ImportError:
+    raise SkipTest("trollius.From not available")
+
+import unittest
+
+from datetime import datetime, timedelta
+from time import sleep
+from rx.concurrency import AsyncIOScheduler
+
+class TestAsyncIOScheduler(unittest.TestCase):
+
+    def test_asyncio_schedule_now(self):
+        loop = asyncio.get_event_loop()
+        scheduler = AsyncIOScheduler(loop)
+        res = scheduler.now() - datetime.now()
+        assert(res < timedelta(seconds=1))
+
+    def test_asyncio_schedule_action(self):
+        loop = asyncio.get_event_loop()
+
+        @asyncio.coroutine
+        def go():
+            scheduler = AsyncIOScheduler(loop)
+
+            class Nonlocal:
+                ran = False
+
+            def action(scheduler, state):
+                Nonlocal.ran = True
+
+            scheduler.schedule(action)
+
+            yield From(asyncio.sleep(0.1, loop=loop))
+            assert(Nonlocal.ran == True)
+
+        loop.run_until_complete(go())
+
+    def test_asyncio_schedule_action_due(self):
+        loop = asyncio.get_event_loop()
+
+        @asyncio.coroutine
+        def go():
+            scheduler = AsyncIOScheduler(loop)
+            starttime = loop.time()
+
+            class Nonlocal:
+               endtime = None
+
+            def action(scheduler, state):
+                Nonlocal.endtime = loop.time()
+
+            scheduler.schedule_relative(0.2, action)
+
+            yield From(asyncio.sleep(0.3, loop=loop))
+            diff = Nonlocal.endtime-starttime
+            assert(diff > 0.18)
+
+        loop.run_until_complete(go())
+
+    def test_asyncio_schedule_action_cancel(self):
+        loop = asyncio.get_event_loop()
+
+        @asyncio.coroutine
+        def go():
+            class Nonlocal:
+                ran = False
+            scheduler = AsyncIOScheduler(loop)
+
+            def action(scheduler, state):
+                Nonlocal.ran = True
+            d = scheduler.schedule_relative(0.01, action)
+            d.dispose()
+
+            yield From(asyncio.sleep(0.1, loop=loop))
+            assert(not Nonlocal.ran)
+
+        loop.run_until_complete(go())
diff --git a/tests/test_concurrency/test_mainloopscheduler/py3_asyncioscheduler.py b/tests/test_concurrency/test_mainloopscheduler/py3_asyncioscheduler.py
index 096256bf..b0ac51e8 100644
--- a/tests/test_concurrency/test_mainloopscheduler/py3_asyncioscheduler.py
+++ b/tests/test_concurrency/test_mainloopscheduler/py3_asyncioscheduler.py
@@ -1,6 +1,8 @@
-try:
-    import asyncio
-except ImportError:
+from nose import SkipTest
+
+import rx
+asyncio = rx.config['asyncio']
+if asyncio is None:
     raise SkipTest("asyncio not available")
 
 import unittest
diff --git a/tests/test_concurrency/test_mainloopscheduler/test_asyncioscheduler.py b/tests/test_concurrency/test_mainloopscheduler/test_asyncioscheduler.py
index b7e4d0f5..ea47efe9 100644
--- a/tests/test_concurrency/test_mainloopscheduler/test_asyncioscheduler.py
+++ b/tests/test_concurrency/test_mainloopscheduler/test_asyncioscheduler.py
@@ -1,7 +1,8 @@
 from nose import SkipTest
-try:
-    import asyncio
-except ImportError:
-    raise SkipTest("asyncio not available")
 
-from .py3_asyncioscheduler import *
\ No newline at end of file
+import sys
+
+if sys.version_info.major < 3:
+    from .py2_asyncioscheduler import *
+else:
+    from .py3_asyncioscheduler import *
\ No newline at end of file
diff --git a/tests/test_observable/py3_fromfuture.py b/tests/test_observable/py3_fromfuture.py
index 122f544e..dbdf72a0 100644
--- a/tests/test_observable/py3_fromfuture.py
+++ b/tests/test_observable/py3_fromfuture.py
@@ -1,6 +1,11 @@
 import unittest
-import asyncio
-from asyncio import Future
+
+from nose import SkipTest
+import rx
+asyncio = rx.config['asyncio']
+if asyncio is None:
+    raise SkipTest("asyncio not available")
+Future = rx.config['Future']
 
 from rx import Observable
 
diff --git a/tests/test_observable/py3_start.py b/tests/test_observable/py3_start.py
index 7900d76b..2f884355 100644
--- a/tests/test_observable/py3_start.py
+++ b/tests/test_observable/py3_start.py
@@ -1,6 +1,7 @@
 import unittest
-import asyncio
-from asyncio import Future
+import rx
+asyncio = rx.config['asyncio']
+Future = rx.config['Future']
 
 from rx import Observable
 from rx.testing import TestScheduler, ReactiveTest, is_prime, MockDisposable
diff --git a/tests/test_observable/py3_tofuture.py b/tests/test_observable/py3_tofuture.py
index 468f58fa..9795c6c9 100644
--- a/tests/test_observable/py3_tofuture.py
+++ b/tests/test_observable/py3_tofuture.py
@@ -1,5 +1,10 @@
 import unittest
-import asyncio
+
+from nose import SkipTest
+import rx
+asyncio = rx.config['asyncio']
+if asyncio is None:
+    raise SkipTest("asyncio not available")
 
 from rx.observable import Observable
 from rx.testing import TestScheduler, ReactiveTest
diff --git a/tests/test_observable/test_fromfuture.py b/tests/test_observable/test_fromfuture.py
index 609987e3..c2731de4 100644
--- a/tests/test_observable/test_fromfuture.py
+++ b/tests/test_observable/test_fromfuture.py
@@ -1,7 +1,7 @@
 from nose import SkipTest
-try:
-    import asyncio
-except ImportError:
-    raise SkipTest("asyncio not available")
+
+import rx
+asyncio = rx.config['asyncio']
+Future = rx.config['Future']
 
 from .py3_fromfuture import *
\ No newline at end of file
diff --git a/tests/test_observable/test_start.py b/tests/test_observable/test_start.py
index ea6a0136..2bca7027 100644
--- a/tests/test_observable/test_start.py
+++ b/tests/test_observable/test_start.py
@@ -1,7 +1,8 @@
 from nose import SkipTest
-try:
-    import asyncio
-except ImportError:
+
+import rx
+asyncio = rx.config['asyncio']
+if asyncio is None:
     raise SkipTest("asyncio not available")
 
 from .py3_start import *
\ No newline at end of file
diff --git a/tests/test_observable/test_tofuture.py b/tests/test_observable/test_tofuture.py
index 2b790299..ed076ef8 100644
--- a/tests/test_observable/test_tofuture.py
+++ b/tests/test_observable/test_tofuture.py
@@ -1,7 +1,8 @@
 from nose import SkipTest
-try:
-    import asyncio
-except ImportError:
+
+import rx
+asyncio = rx.config['asyncio']
+if asyncio is None:
     raise SkipTest("asyncio not available")
 
 from .py3_tofuture import *
\ No newline at end of file