Feature/subscribe scheduler (ReactiveX#222)

Pass in scheduler using subscribe

Author: dbrattli

.gitignore

@@ -9,6 +9,7 @@
 dist
 build
 eggs
+.eggs
 parts
 bin
 var

.travis.yml

@@ -1,14 +1,15 @@
 language: python
 python:
-  - "3.5"
+  - "3.6"
 # command to install dependencies, e.g. pip install -r requirements.txt --use-mirrors
 install:
   - python setup.py install
   - pip install coveralls
   - pip install coverage
-  - pip install nose
+  - 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
+script:
+  - coverage run --source=rx setup.py test
 after_success:
-  coveralls
+  - coveralls

examples/ironpython/meetdotnet.py

@@ -0,0 +1,2 @@
+[pytest]
+testpaths = tests

pytest.ini

@@ -1,4 +1,4 @@
-from rx.core import Observable, Observable
+from rx.core import Observable
 from rx.internal import extensionmethod
 
 from .controlledsubject import ControlledSubject
@@ -12,8 +12,8 @@ def __init__(self, source, enable_queue, scheduler=None):
         self.subject = ControlledSubject(enable_queue, scheduler)
         self.source = source.multicast(self.subject).ref_count()
 
-    def _subscribe_core(self, observer):
-        return self.source.subscribe(observer)
+    def _subscribe_core(self, observer, scheduler=None):
+        return self.source.subscribe(observer, scheduler)
 
     def request(self, number_of_items):
         if number_of_items is None:

rx/backpressure/controlledobservable.py

@@ -19,8 +19,8 @@ def __init__(self, enable_queue=True, scheduler=None):
         self.has_completed = False
         self.scheduler = scheduler or current_thread_scheduler
 
-    def _subscribe_core(self, observer):
-        return self.subject.subscribe(observer)
+    def _subscribe_core(self, observer, scheduler=None):
+        return self.subject.subscribe(observer, scheduler)
 
     def close(self):
         self.has_completed = True

rx/backpressure/controlledsubject.py

@@ -17,7 +17,7 @@ def __init__(self, source, pauser=None):
 
         super(PausableObservable, self).__init__()
 
-    def _subscribe_core(self, observer):
+    def _subscribe_core(self, observer, scheduler=None):
         conn = self.source.publish()
         subscription = conn.subscribe(observer)
         connection = [Disposable.empty()]
@@ -29,7 +29,7 @@ def send(b):
                 connection[0].dispose()
                 connection[0] = Disposable.empty()
 
-        pausable = self.pauser.distinct_until_changed().subscribe_callbacks(send)
+        pausable = self.pauser.distinct_until_changed().subscribe_callbacks(send, scheduler=scheduler)
         return CompositeDisposable(subscription, connection[0], pausable)
 
     def pause(self):

rx/backpressure/pausable.py

@@ -5,7 +5,7 @@
 
 
 def combine_latest_source(source, subject, result_selector):
-    def subscribe(observer):
+    def subscribe(observer, scheduler=None):
         has_value = [False, False]
         has_value_all = [False]
         values = [None, None]
@@ -47,8 +47,8 @@ def close_subject():
             next(True, 1)
 
         return CompositeDisposable(
-            source.subscribe_callbacks(lambda x: next(x, 0), throw_source, close_source),
-            subject.subscribe_callbacks(lambda x: next(x, 1), observer.throw, close_subject)
+            source.subscribe_callbacks(lambda x: next(x, 0), throw_source, close_source, scheduler),
+            subject.subscribe_callbacks(lambda x: next(x, 1), observer.throw, close_subject, scheduler)
         )
     return AnonymousObservable(subscribe)
 
@@ -66,7 +66,7 @@ def __init__(self, source, pauser=None):
 
         super(PausableBufferedObservable, self).__init__()
 
-    def _subscribe_core(self, observer):
+    def _subscribe_core(self, observer, scheduler=None):
         previous_should_fire = [None]
         queue = []
 
@@ -106,7 +106,7 @@ def close():
             self.source,
             self.pauser.distinct_until_changed().start_with(False),
             result_selector
-        ).subscribe_callbacks(send, throw, close)
+        ).subscribe_callbacks(send, throw, close, scheduler)
 
         return subscription
 

rx/backpressure/pausablebuffered.py

@@ -53,9 +53,9 @@ def __init__(self, source, scheduler=None):
         self.source = source
         self.subscription = None
 
-    def _subscribe_core(self, observer):
+    def _subscribe_core(self, observer, scheduler=None):
         observer = StopAndWaitObserver(observer, self, self.subscription, self.scheduler)
-        self.subscription = self.source.subscribe(observer)
+        self.subscription = self.source.subscribe(observer, scheduler)
 
         def action(scheduler, state=None):
             self.source.request(1)

rx/backpressure/stopandwaitobservable.py

@@ -53,9 +53,9 @@ def __init__(self, source, window_size, scheduler=None):
         self.scheduler = scheduler or current_thread_scheduler
         self.subscription = None
 
-    def _subscribe_core(self, observer):
+    def _subscribe_core(self, observer, scheduler=None):
         observer = WindowedObserver(observer, self, self.subscription, self.scheduler)
-        self.subscription = self.source.subscribe(observer)
+        self.subscription = self.source.subscribe(observer, scheduler)
 
         def action(scheduler, state):
             self.source.request(self.window_size)

rx/backpressure/windowedobservable.py

@@ -46,7 +46,7 @@ def __init__(self):
     def schedule(self, action, state=None):
         """Schedules an action to be executed."""
 
-        log.debug("CurrentThreadScheduler.schedule(state=%s)", state)
+        #log.debug("CurrentThreadScheduler.schedule(state=%s)", state)
         return self.schedule_relative(timedelta(0), action, state)
 
     def schedule_relative(self, duetime, action, state=None):

rx/concurrency/currentthreadscheduler.py

@@ -8,6 +8,7 @@
 
 log = logging.getLogger("Rx")
 
+MAX_SPINNING = 100
 
 class VirtualTimeScheduler(SchedulerBase):
     """Virtual Scheduler. This scheduler should work with either
@@ -70,13 +71,23 @@ def start(self):
             return
 
         self.is_enabled = True
+
+        spinning = 0
         while self.is_enabled:
-            next = self.get_next()
-            if not next:
+            item = self.get_next()
+            if not item:
                 break
-            if next.duetime > self.clock:
-                self.clock = next.duetime
-            next.invoke()
+
+            if item.duetime > self.clock:
+                self.clock = item.duetime
+                spinning = 0
+
+            if spinning > MAX_SPINNING:
+                self.clock += 1
+                spinning = 0
+
+            item.invoke()
+            spinning += 1
 
         self.is_enabled = False
 
@@ -104,18 +115,18 @@ def advance_to(self, time):
         self.is_enabled = True
 
         while self.is_enabled:
-            next = self.get_next()
-            if not next:
+            item = self.get_next()
+            if not item:
                 break
 
-            if next.duetime > time:
-                self.queue.enqueue(next)
+            if item.duetime > time:
+                self.queue.enqueue(item)
                 break
 
-            if next.duetime > self.clock:
-                self.clock = next.duetime
+            if item.duetime > self.clock:
+                self.clock = item.duetime
 
-            next.invoke()
+            item.invoke()
 
         self.is_enabled = False
         self.clock = time
@@ -150,10 +161,10 @@ def sleep(self, time):
     def get_next(self):
         """Returns the next scheduled item to be executed."""
 
-        while len(self.queue):
-            next = self.queue.dequeue()
-            if not next.is_cancelled():
-                return next
+        while self.queue:
+            item = self.queue.dequeue()
+            if not item.is_cancelled():
+                return item
 
         return None
 

rx/concurrency/virtualtimescheduler.py

@@ -1,11 +1,12 @@
+from typing import Callable
 from .observable import Observable
 
 
 class AnonymousObservable(Observable):
     """Class to create an Observable instance from a delegate-based
     implementation of the Subscribe method."""
 
-    def __init__(self, subscribe):
+    def __init__(self, subscribe: Callable) -> None:
         """Creates an observable sequence object from the specified
         subscription function.
 
@@ -16,5 +17,5 @@ def __init__(self, subscribe):
         self._subscribe = subscribe
         super(AnonymousObservable, self).__init__()
 
-    def _subscribe_core(self, observer):
-        return self._subscribe(observer)
+    def _subscribe_core(self, observer, scheduler=None):
+        return self._subscribe(observer, scheduler)

rx/core/anonymousobservable.py

@@ -8,7 +8,7 @@ def __init__(self, send=None, throw=None, close=None):
 
         self._next = send or noop
         self._error = throw or default_error
-        self._completed = close or noop
+        self._close = close or noop
 
     def _send_core(self, value):
         self._next(value)
@@ -17,4 +17,4 @@ def _throw_core(self, error):
         self._error(error)
 
     def _close_core(self):
-        self._completed()
+        self._close()

rx/core/anonymousobserver.py

@@ -3,5 +3,5 @@
 
 class Observable(metaclass=ABCMeta):
     @abstractmethod
-    def subscribe(self, observer):
+    def subscribe(self, observer=None, scheduler=None):
         raise NotImplementedError

rx/core/bases/observable.py

@@ -15,5 +15,5 @@ def __init__(self, observable=None):
         self.observable = observable
         super(BlockingObservable, self).__init__()
 
-    def _subscribe_core(self, observer):
-        return self.observable.subscribe(observer)
+    def _subscribe_core(self, observer, scheduler=None):
+        return self.observable.subscribe(observer, scheduler)

rx/core/blockingobservable.py

@@ -45,7 +45,7 @@ def to_observable(self, scheduler=None):
         notification = self
         scheduler = scheduler or immediate_scheduler
 
-        def subscribe(observer):
+        def subscribe(observer, scheduler=None):
             def action(scheduler, state):
                 notification._accept_observable(observer)
                 if notification.kind == 'N':