Schedulers, CI (#306)

* Fix state propagation in CurrentThreadScheduler and NewThreadScheduler

* Fix TwistedScheduler import in tests

* Test QtScheduler with pyside2 (v1 unsupported as of Python 3.5)

* Add missing import in GtkScheduler

* Fix EventletScheduler, and its tests

* Fix GEventScheduler, and add its tests to CI

* Fix intermittent timing failures

* Use Tcl instead of Tk so we can test headless

* Add Windows and MacOS to Travis CI config

* Add optional dependencies to CI to increase coverage

* Remove spurious import in tkinter test

* Add PyYaml for coveralls reporting, remove redundant apt-get installs

.travis.yml

@@ -1,15 +1,48 @@
-language: python
-python:
-  - "3.6"
-# command to install dependencies, e.g. pip install -r requirements.txt --use-mirrors
+matrix:
+  include:
+    - os: linux
+      dist: xenial
+      language: python
+      python: 3.6
+      # Coverage for the baseline 3.6 build: include some optional packages
+      before_script:
+        - pip install eventlet gevent pyyaml tornado twisted
+        # pycairo / pygobject need native libraries
+        - sudo apt-get install -y libgirepository1.0-dev gir1.2-gtk-3.0
+        - pip install pycairo pygobject
+
+    - os: linux
+      dist: xenial
+      language: python
+      python: 3.7
+
+    - os: linux
+      dist: xenial
+      language: python
+      python: 3.8-dev
+
+    - os: osx
+      osx_image: xcode8.3
+      language: python
+      python: 3.7-dev
+
+    - os: windows
+      python: 3.7-dev
+      language: sh
+      before_install:
+        - choco install python3
+        - export PATH="/c/Python37:/c/Python37/Scripts:$PATH"
+
 install:
   - python setup.py install
-  - pip install coveralls
-  - pip install coverage
+  - pip install coveralls coverage
   - pip install pytest>=3.0.2 pytest-asyncio pytest-cov --upgrade
-  - pip install tornado
-# command to run tests, e.g. python setup.py test
+
 script:
   - coverage run --source=rx setup.py test
+
 after_success:
-  - coveralls
+  - if [[ "$TRAVIS_OS_NAME" == "linux" && "$TRAVIS_PYTHON_VERSION" == 3.6* ]];
+    then
+      coveralls;
+    fi

rx/concurrency/currentthreadscheduler.py

@@ -80,7 +80,7 @@ def schedule_absolute(self, duetime: typing.AbsoluteTime, action: typing.Schedul
         """Schedules an action to be executed at duetime."""
 
         duetime = self.to_datetime(duetime)
-        return self.schedule_relative(duetime - self.now, action, state=None)
+        return self.schedule_relative(duetime - self.now, action, state=state)
 
     def _get_queue(self) -> PriorityQueue:
         ident = threading.current_thread().ident

rx/concurrency/mainloopscheduler/eventletscheduler.py

@@ -24,6 +24,7 @@ def __init__(self) -> None:
         import eventlet
         import eventlet.hubs
 
+
     def schedule(self, action: typing.ScheduledAction, state: Any = None) -> typing.Disposable:
         """Schedules an action to be executed."""
 
@@ -52,17 +53,15 @@ def schedule_relative(self, duetime: typing.RelativeTime, action: typing.Schedul
             (best effort).
         """
 
-        scheduler = self
         seconds = self.to_seconds(duetime)
         if not seconds:
-            return scheduler.schedule(action, state)
+            return self.schedule(action, state)
 
         sad = SingleAssignmentDisposable()
 
         def interval():
             sad.disposable = self.invoke_action(action, state)
 
-        log.debug("timeout: %s", seconds)
         timer = [eventlet.spawn_after(seconds, interval)]
 
         def dispose():
@@ -92,4 +91,4 @@ def now(self) -> datetime:
         scheduled on a scheduler will adhere to the time denoted by
         this property."""
 
-        return self.to_datetime(eventlet.hubs.hub.time.time())
+        return self.to_datetime(eventlet.hubs.get_hub().clock())

rx/concurrency/mainloopscheduler/geventscheduler.py

@@ -4,9 +4,10 @@
 from rx.disposable import SingleAssignmentDisposable, CompositeDisposable
 from rx.concurrency.schedulerbase import SchedulerBase
 
-gevent = None
 log = logging.getLogger("Rx")
 
+gevent = None
+
 
 class GEventScheduler(SchedulerBase):
     """A scheduler that schedules work via the GEvent event loop.
@@ -18,7 +19,6 @@ def __init__(self):
         # Lazy import gevent
         global gevent
         import gevent
-        import gevent.core
 
     def schedule(self, action, state=None):
         """Schedules an action to be executed."""
@@ -82,4 +82,4 @@ def now(self):
         """Represents a notion of time for this scheduler. Tasks being scheduled
         on a scheduler will adhere to the time denoted by this property."""
 
-        return self.to_datetime(gevent.core.time())
+        return self.to_datetime(gevent.get_hub().loop.now())

rx/concurrency/mainloopscheduler/gtkscheduler.py

@@ -1,5 +1,5 @@
 from rx.core import typing
-from rx.disposable import SingleAssignmentDisposable, CompositeDisposable
+from rx.disposable import SingleAssignmentDisposable, CompositeDisposable, Disposable
 from rx.concurrency.schedulerbase import SchedulerBase
 
 

rx/concurrency/newthreadscheduler.py

@@ -43,7 +43,7 @@ def schedule_absolute(self, duetime: typing.AbsoluteTime, action: typing.Schedul
         """Schedules an action to be executed at duetime."""
 
         dt = SchedulerBase.to_datetime(duetime)
-        return self.schedule_relative(dt - self.now, action, state=None)
+        return self.schedule_relative(dt - self.now, action, state=state)
 
     def schedule_periodic(self, period: typing.RelativeTime, action: typing.ScheduledPeriodicAction,
                           state: typing.TState = None) -> typing.Disposable:

rx/concurrency/schedulerbase.py

@@ -1,10 +1,9 @@
 from datetime import datetime, timedelta
 from typing import Optional
 
-from rx.disposable import Disposable
 from rx.core import typing
 from rx.core.typing import ScheduledAction, ScheduledPeriodicAction, TState
-from rx.disposable import SerialDisposable
+from rx.disposable import Disposable, MultipleAssignmentDisposable
 from rx.internal.basic import default_now
 
 
@@ -35,15 +34,16 @@ def schedule_periodic(self, period: typing.RelativeTime, action: ScheduledPeriod
             The disposable object used to cancel the scheduled
             recurring action (best effort)."""
 
-        disp = SerialDisposable()
+        disp = MultipleAssignmentDisposable()
 
-        def invoke_action(scheduler: typing.Scheduler, _: TState) -> Optional[Disposable]:
+        def invoke_periodic(scheduler: typing.Scheduler, _: TState) -> Optional[Disposable]:
             nonlocal state
 
             if disp.is_disposed:
                 return None
 
-            disp.disposable = self.schedule_relative(period, invoke_action, None)
+            if period:
+                disp.disposable = self.schedule_relative(period, invoke_periodic, None)
 
             try:
                 new_state = action(state)
@@ -56,7 +56,7 @@ def invoke_action(scheduler: typing.Scheduler, _: TState) -> Optional[Disposable
 
             return None
 
-        disp.disposable = self.schedule_relative(period, invoke_action, state)
+        disp.disposable = self.schedule_relative(period, invoke_periodic, None)
         return disp
 
     @property

tests/test_concurrency/test_mainloopscheduler/test_asyncioscheduler.py

@@ -71,7 +71,7 @@ def action(scheduler, state):
             scheduler.schedule_relative(0.2, action)
 
             yield from asyncio.sleep(0.3, loop=loop)
-            diff = endtime-starttime
+            diff = endtime - starttime
             assert diff > 0.18
 
         loop.run_until_complete(go())
@@ -95,7 +95,7 @@ def schedule():
             threading.Thread(target=schedule).start()
 
             yield from asyncio.sleep(0.3, loop=loop)
-            diff = endtime-starttime
+            diff = endtime - starttime
             assert diff > 0.18
 
         loop.run_until_complete(go())
@@ -111,11 +111,11 @@ def go():
             def action(scheduler, state):
                 nonlocal ran
                 ran = True
-            d = scheduler.schedule_relative(0.010, action)
+            d = scheduler.schedule_relative(0.05, action)
             d.dispose()
 
-            yield from asyncio.sleep(0.1, loop=loop)
-            assert(not ran)
+            yield from asyncio.sleep(0.3, loop=loop)
+            assert not ran
 
         loop.run_until_complete(go())
 
@@ -132,12 +132,12 @@ def action(scheduler, state):
                 ran = True
 
             def schedule():
-                d = scheduler.schedule_relative(0.010, action)
+                d = scheduler.schedule_relative(0.05, action)
                 d.dispose()
 
             threading.Thread(target=schedule).start()
 
-            yield from asyncio.sleep(0.1, loop=loop)
-            assert(not ran)
+            yield from asyncio.sleep(0.3, loop=loop)
+            assert not ran
 
         loop.run_until_complete(go())

tests/test_concurrency/test_mainloopscheduler/test_eventletscheduler.py

@@ -12,7 +12,7 @@ class TestEventLetEventScheduler(unittest.TestCase):
     def test_eventlet_schedule_now(self):
         scheduler = EventLetEventScheduler()
         res = scheduler.now - datetime.now()
-        assert(res < timedelta(seconds=1))
+        assert res < timedelta(seconds=1)
 
     def test_eventlet_schedule_action(self):
         scheduler = EventLetEventScheduler()
@@ -23,7 +23,7 @@ def action(scheduler, state):
         scheduler.schedule(action)
 
         eventlet.sleep(0.1)
-        assert(ran[0] is True)
+        assert ran[0] is True
 
     def test_eventlet_schedule_action_due(self):
         scheduler = EventLetEventScheduler()
@@ -33,11 +33,11 @@ def test_eventlet_schedule_action_due(self):
         def action(scheduler, state):
             endtime[0] = datetime.now()
 
-        scheduler.schedule_relative(2.0, action)
+        scheduler.schedule_relative(0.2, action)
 
         eventlet.sleep(0.3)
-        diff = endtime[0]-starttime
-        assert(diff > timedelta(seconds=0.18))
+        diff = endtime[0] - starttime
+        assert diff > timedelta(seconds=0.18)
 
     def test_eventlet_schedule_action_cancel(self):
         scheduler = EventLetEventScheduler()
@@ -48,31 +48,30 @@ def action(scheduler, state):
         d = scheduler.schedule_relative(1.0, action)
         d.dispose()
 
-        eventlet.sleep(0.1)
-        assert(not ran[0])
+        eventlet.sleep(0.01)
+        assert not ran[0]
 
     def test_eventlet_schedule_action_periodic(self):
         scheduler = EventLetEventScheduler()
-        period = .050
+        period = 0.05
         counter = [3]
 
-
-        def action(scheduler, state):
+        def action(state):
             if counter[0]:
                 counter[0] -= 1
 
         scheduler.schedule_periodic(period, action)
-        eventlet.sleep(.3)
-        assert (counter[0] == 0)
+        eventlet.sleep(0.3)
+        assert counter[0] == 0
 
     def test_eventlet_schedule_action_periodic_now(self):
         scheduler = EventLetEventScheduler()
         period = 0
         num_times = [3]
 
-        def action(scheduler, state):
+        def action(state):
             num_times[0] -= 1
 
         scheduler.schedule_periodic(period, action)
-        eventlet.sleep(.3)
-        assert (num_times[0] == 2)
+        eventlet.sleep(0.3)
+        assert num_times[0] == 2

tests/test_concurrency/test_mainloopscheduler/test_qtscheduler.py

@@ -2,17 +2,19 @@
 from datetime import datetime, timedelta
 import pytest
 
+
+skip = False
 try:
-    from PyQt4 import QtCore
-    from PyQt4.QtGui import QApplication
+    from PyQt5 import QtCore
+    from PyQt5.QtWidgets import QApplication
 except ImportError:
     try:
-        from PyQt5 import QtCore
-        from PyQt5.QtWidgets import QApplication
+        from PyQt4 import QtCore
+        from PyQt4.QtGui import QApplication
     except ImportError:
         try:
-            from PySide import QtCore
-            from PySide.QtGui import QApplication
+            from PySide2 import QtCore
+            from PySide2.QtGui import QGuiApplication as QApplication
         except ImportError:
             skip = True
 

tests/test_concurrency/test_mainloopscheduler/test_tkinterscheduler.py

@@ -7,7 +7,7 @@
 tkinter = pytest.importorskip("tkinter")
 
 try:
-    root = tkinter.Tk()
+    root = tkinter.Tcl()
     display = True
 except Exception:
     display = False

tests/test_concurrency/test_mainloopscheduler/test_twistedscheduler.py

@@ -6,7 +6,7 @@
 from twisted.internet import reactor, defer
 from twisted.trial import unittest
 
-from rx.concurrency import TwistedScheduler
+from rx.concurrency.mainloopscheduler import TwistedScheduler
 
 
 class TestTwistedScheduler(unittest.TestCase):