class Disposable(object):
  """Represents a disposable object"""

  def dispose(self):
    pass

  def __enter__(self):
    return self

  def __exit__(self, exc_type, exc_value, traceback):
    self.dispose()

  @staticmethod
  def create(action):
    return AnonymouseDisposable(action)

  @staticmethod
  def empty():
    return disposableEmpty class Struct(object):
  def __init__(self, **entries):
    self.__dict__.update(entries)

  def __eq__(self, other):
    if not isinstance(other, Struct):
      return False

    return self.__dict__ == other.__dict__

  def __repr__(self):
    return repr(self.__dict__) class Observable(object):
  """Provides all extension methods to Observable"""

  def subscribe(self, observerOrOnNext=noop, onError=noop, onComplete=noop):
    observer = observerOrOnNext

    if observerOrOnNext == None or callable(observerOrOnNext):
      observer = Observer.create(observerOrOnNext, onError, onComplete)

    return self.subscribeCore(observer)

  def subscribeCore(self, observer):
    raise NotImplementedError()

  def subscribeSafe(self, observer):
    if isinstance(self, ObservableBase):
      return self.subscribeCore(observer)
    elif isinstance(self, Producer):
      return self.subscribeRaw(observer, False)

    d = Disposable.empty()

    try:
      d = self.subscribeCore(observer)
    except Exception as e:
      observer.onError(e)

    return d class Scheduler(object, metaclass=MetaScheduler):
  """Provides a set of static properties to access commonly
  used Schedulers and implements all the scheduling methods that
  then use the overrides of the implementation."""

  @staticmethod
  def invokeAction(scheduler, action):
    action()
    return Disposable.empty()

  @staticmethod
  def invokeRecImmediate(scheduler, pair):
    state = pair[0]
    action = pair[1]

    scheduled = RecursiveScheduledFunction(action, scheduler)
    scheduled.run(state)

    return scheduled.group

  @staticmethod
  def invokeRecDate(scheduler, pair, method):
    state = pair[0]
    action = pair[1]

    scheduled = RecursiveScheduledFunction(action, scheduler, method)
    scheduled.run(state)

    return scheduled.group

  @staticmethod
  def now():
    return defaultNow()

  @staticmethod
  def normalize(timeSpan):
    if timeSpan < 0:
      return 0
    else:
      return timeSpan

  def catchException(self, handler):
    return CatchScheduler(self, handler)

  # longrunning scheduling
  # action takes as parameter: state and cancel
  # and returns None
  def scheduleLongRunning(self, action):
    return self.scheduleLongRunningWithState(None, lambda s, cancel: action(cancel))

  ## commented out such that one can test if a scheduler can schedule long running
  # def scheduleLongRunningWithState(self, state, action):
  #   raise NotImplementedError()
  @property
  def isLongRunning(self):
    return hasattr(self, "scheduleLongRunningWithState")

  # periodic scheduling
  # action takes as parameter: state
  # and returns: state
  def schedulePeriodic(self, period, action):
    return self.schedulePeriodicWithState(None, period, lambda s: action())

  def schedulePeriodicWithState(self, state, period, action):
    def gated():
      state = action(state)

    timer = PeriodicTimer(period, gated)

    return timer.start()

  # once scheduling
  # action takes as parameter: scheduler, state
  # and returns: disposable to cancel
  def schedule(self, action):
    return self._scheduleCore(action, Scheduler.invokeAction)

  def scheduleWithState(self, state, action):
    return self._scheduleCore(state, action)

  def scheduleWithRelative(self, dueTime, action):
    return self._scheduleRelativeCore(action, dueTime, Scheduler.invokeAction)

  def scheduleWithRelativeAndState(self, state, dueTime, action):
    return self._scheduleRelativeCore(state, dueTime, action)

  def scheduleWithAbsolute(self, dueTime, action):
    return self._scheduleAbsoluteCore(action, dueTime, Scheduler.invokeAction)

  def scheduleWithAbsoluteAndState(self, state, dueTime, action):
    return self._scheduleAbsoluteCore(state, dueTime, action)

  # recursive scheduling
  # action takes as parameter: continuation([state], [period])
  # and returns: None
  def scheduleRecursive(self, action):
    return self.scheduleRecursiveWithState(
      None,
      lambda _, _continuation: action(lambda: _continuation(None))
    )

  def scheduleRecursiveWithState(self, state, action):
    return self.scheduleWithState((state, action), Scheduler.invokeRecImmediate)

  def scheduleRecursiveWithRelative(self, dueTime, action):
    return self.scheduleRecursiveWithRelativeAndState(
      None,
      dueTime,
      lambda _, _continuation: action(lambda dt: _continuation(None, dt))
    )

  def scheduleRecursiveWithRelativeAndState(self, state, dueTime, action):
    return self._scheduleRelativeCore(
      (state, action),
      dueTime,
      lambda s, p: Scheduler.invokeRecDate(s, p, 'scheduleWithRelativeAndState')
    )

  def scheduleRecursiveWithAbsolute(self, dueTime, action):
    return self.scheduleRecursiveWithAbsoluteAndState(
      action,
      lambda _action, _self: _action(lambda: _self(_action))
    )

  def scheduleRecursiveWithAbsoluteAndState(self, state, dueTime, action):
    return self._scheduleAbsoluteCore(
      (state, action),
      dueTime,
      lambda s, p: Scheduler.invokeRecDate(s, p, 'scheduleWithAbsoluteAndState')
    )

  def _scheduleCore(self, state, action):
    raise NotImplementedError()

  def _scheduleRelativeCore(self, state, dueTime, action):
    raise NotImplementedError()

  def _scheduleAbsoluteCore(self, state, dueTime, action):
    raise NotImplementedError() class Subject(Observable, Observer):
  def __init__(self):
    super(Subject, self).__init__()
    self.isDisposed = False
    self.isStopped = False
    self.exception = None
    self.observer = Atomic(NoopObserver.instance)

  def onCompleted(self):
    old = None
    new = DoneObserver.completed

    while True:
      old = self.observer.value

      if old is DoneObserver.completed or isinstance(old, DoneObserver):
        break

      current = self.observer.compareExchange(new, old)

      if old is current:
        break

    old.onCompleted()

  def onError(self, exception):
    old = None
    new = DoneObserver(exception)

    while True:
      old = self.observer.value

      if old is DoneObserver.completed or isinstance(old, DoneObserver):
        break

      current = self.observer.compareExchange(new, old)

      if old is current:
        break

    old.onError(exception)

  def onNext(self, value):
    self.observer.value.onNext(value)

  class Subscription(Disposable):
    def __init__(self, subject, observer):
      self.subject = subject
      self.observer = Atomic(observer)

    def dispose(self):
      old = self.observer.exchange(None)

      if old != None:
        self.subject.unsubscribe(old)
        self.subject = None

  def subscribeCore(self, observer):
    old = None
    new = None

    while True:
      old = self.observer.value

      if old is DisposedObserver.instance:
        raise Exception("Object has been disposed")

      if old is DoneObserver.completed:
        observer.onCompleted()
        return Disposable.empty()

      if isinstance(old, DoneObserver):
        observer.onError(old.exception)
        return Disposable.empty()

      if old is NoopObserver.instance:
        new = observer
      else:
        if isinstance(old, ListObserver):
          new = old.add(observer)
        else:
          new = ListObserver((old, observer))

      current = self.observer.compareExchange(new, old)

      if old is current:
        break

    return self.Subscription(self, observer)

  def unsubscribe(self, observer):
    old = None
    new = None

    while True:
      old = self.observer.value

      if old is DisposedObserver.instance or isinstance(old, DoneObserver):
        return

      if isinstance(old, ListObserver):
        new = old.remove(observer)
      elif observer is not old:
        return
      else:
        new = NoopObserver.instance

      current = self.observer.compareExchange(new, old)

      if old is current:
        return

  def dispose(self):
    self.observer.exchange(NoopObserver.instance)

  @staticmethod
  def create(observer, observable):
    return AnonymousSubject(observer, observable)

  @staticmethod
  def synchronize(subject, scheduler=None):
    if scheduler == None:
      return AnonymousSubject(Observer.synchronize(subject), subject)
    else:
      return AnonymousSubject(Observer.synchronize(subject), subject.observeOn(scheduler)) def OnNext(value):
  return Notification.createOnNext(value) def OnError(exception):
  return Notification.createOnError(exception) def OnCompleted():
  return Notification.createOnCompleted() class TestScheduler(HistoricalScheduler):
  TIME_CREATE = 100
  TIME_SUBSCRIBE = 200
  TIME_DISPOSE = 1000

  def __init__(self):
    super(TestScheduler, self).__init__()

  def createObserver(self):
    return self.MockObserver(self)

  def createHotObservable(self, *messages):
    return HotObservable(self, list(messages))

  def createColdObservable(self, *messages):
    return ColdObservable(self, list(messages))

  def start(self, factory, created=TIME_CREATE, subscribed=TIME_SUBSCRIBE, disposed=TIME_DISPOSE):
    state = Struct(
      source=None,
      subscription=None,
      observer=self.createObserver()
    )

    def scheduledCreate():
      state.source = factory()
      return Disposable.empty()

    def scheduledSubscribe():
      state.subscription = state.source.subscribe(state.observer)
      return Disposable.empty()

    def scheduledDispose():
      state.subscription.dispose()
      return Disposable.empty()

    self.scheduleWithAbsolute(created, scheduledCreate)
    self.scheduleWithAbsolute(subscribed, scheduledSubscribe)
    self.scheduleWithAbsolute(disposed, scheduledDispose)

    super(TestScheduler, self).start()

    return state.observer

  class MockObserver(Observer):
    def __init__(self, scheduler):
      super(TestScheduler.MockObserver, self).__init__()
      self.scheduler = scheduler
      self.messages = []

    def onNext(self, value):
      self.messages.append((
        self.scheduler.now(),
        OnNext(value)
      ))

    def onError(self, exception):
      self.messages.append((
        self.scheduler.now(),
        OnError(exception)
      ))

    def onCompleted(self):
      self.messages.append((
        self.scheduler.now(),
        OnCompleted()
      )) class ReactiveTest(unittest.TestCase):
  def simpleHot(self, *values):
    scheduler = TestScheduler()

    def gen(start):
      while True:
        yield start
        start += 10

    messages = list(
      zip(
        gen(TestScheduler.TIME_SUBSCRIBE + 10),
        [OnNext(x) for x in values] + [OnCompleted()]
      )
    )
    observable = scheduler.createHotObservable(*messages)

    return (
      scheduler,
      observable,
      messages
    )

  def simpleCold(self, *values):
    scheduler = TestScheduler()

    def gen(start):
      while True:
        yield start
        start += 10

    messages = list(
      zip(
        gen(10),
        [OnNext(x) for x in values] + [OnCompleted()]
      )
    )
    observable = scheduler.createColdObservable(*messages)

    return (
      scheduler,
      observable,
      messages
    )

  def assertHasValues(self, observer, vals, endAt, text):
    errors = [x for x in observer.messages if x[1].kind == Notification.KIND_ERROR]

    if len(errors) > 0:
      raise errors[0][1].exception

    messages = [
      (x[0], OnNext(x[1])) for x in vals
    ]

    if endAt != None:
      messages += [
        (endAt, OnCompleted())
      ]

    self.assertSequenceEqual(messages, observer.messages, text)

  def assertHasSingleValue(self, observer, value, at, text):
    errors = [x for x in observer.messages if x[1].kind == Notification.KIND_ERROR]
    values = [x for x in observer.messages if x[1].kind == Notification.KIND_NEXT]

    if len(errors) > 0:
      raise errors[0][1].exception

    self.assertEqual(1, len(values), text)
    self.assertEqual(at, values[0][0], text)
    self.assertEqual(value, values[0][1].value, text)

  def assertHasError(self, observer, exceptionText, at, text):
    errors = [x for x in observer.messages if x[1].kind == Notification.KIND_ERROR]

    self.assertEqual(exceptionText, str(errors[0][1].exception), text)
    self.assertEqual(at, errors[0][0], text)
    self.assertEqual(1, len(errors), text)

  def assertHasRecorded(self, observer, messages, text):
    errors = [x for x in observer.messages if x[1].kind == Notification.KIND_ERROR]
    errorsExpected = [x for x in messages if x[1].kind == Notification.KIND_ERROR]

    if len(errors) > 0:
      if len(errorsExpected) == 0:
        raise errors[0][1].exception
      else:
        self.assertEqual(errorsExpected[0][0], errors[0][0], text)
        self.assertEqual(str(errorsExpected[0][1].exception), str(errors[0][1].exception), text)

    self.assertSequenceEqual(messages, observer.messages, text)
import unittest
import concurrent.futures
from rx.disposable import Disposable
from rx.internal import Struct
from rx.observable import Observable
from rx.scheduler import Scheduler
from rx.subject import Subject
from test.reactive import OnNext, OnError, OnCompleted, TestScheduler, ReactiveTest

    o = sched.start(
      lambda: o1.groupJoin(
        o2,
        lambda x: Observable.never(),
        lambda x: Observable.never(),
        lambda leftValue, leftWindow: leftWindow.select(lambda x: (leftValue, x))
      ).merge()
    )

    self.assertHasValues(o, [
        (220, (1, 6)),
        (230, (1, 7)),
        (240, (2, 6)),
        (240, (2, 7)),
      ],
      None,
      "groupJoin should yield tuples according to selector function"
    )

  def test_join(self):
    sched = TestScheduler()
    o1 = sched.createHotObservable(
      (210, OnNext(1)),
      (240, OnNext(2)),
      (250, OnCompleted())
    )
    o2 = sched.createHotObservable(
      (220, OnNext(6)),
      (230, OnNext(7)),
      (240, OnCompleted())
    )

    o = sched.start(
      lambda: o1.join(
        o2,
        lambda x: Observable.never(),
        lambda x: Observable.never(),
        lambda left, right: (left, right)
      )
    )

    self.assertHasValues(o, [
        (220, (1, 6)),
        (230, (1, 7)),
        (240, (2, 6)),
        (240, (2, 7)),
      ],
      250,
      "groupJoin should yield tuples according to selector function"
    )

  def test_of_type(self):
    sched = TestScheduler()
    xs = sched.createHotObservable(
      (210, OnNext(1)),
      (240, OnNext("Hello")),
      (250, OnCompleted())
    )

    o = sched.start(
      lambda: xs.ofType(int)
    )

    self.assertHasValues(o, [
        (210, 1),
      ],
      250,
      "ofType should ignore values that are of wrong type"
    )

  def test_select(self):
    sched, xs, messages = self.simpleHot(1, 2, 3)

    o = sched.start(
      lambda: xs.select(lambda x: x * x)
    )

    self.assertHasValues(o, [
        (210, 1),
        (220, 4),
        (230, 9),
      ],
      240,
      "select should yield selector applied to values"
    )

  def test_select_enumerate(self):
    sched, xs, messages = self.simpleHot(1, 2, 3)

    o = sched.start(
      lambda: xs.selectEnumerate(lambda x, i: x * i)
    )

    self.assertHasValues(o, [
        (210, 0),
        (220, 2),
        (230, 6),
      ],
      240,
      "selectEnumerate should yield selector applied to values"
    )

  def test_select_many(self):
    ex = Exception("Test Exception")
    sched = TestScheduler()
    xs = sched.createHotObservable(
      (210, OnNext(2)),
      (250, OnError(ex))
    )

    o = sched.start(
      lambda: xs.selectMany(
        lambda x: Observable.returnValue(x*x),
        lambda e: Observable.throw(e)
      )
    )

    self.assertHasRecorded(o, [
        (210, OnNext(4)),

        (250, OnError(ex)),