from rx.disposable import CompositeDisposable, RefCountDisposable, SingleAssignmentDisposable
from rx.observable import Producer
from rx.observer import Observer
import rx.linq.sink
from threading import RLock


class Join(Producer):
  def __init__(self, left, right, leftDurationSelector, rightDurationSelector, resultSelector):
    super(Join, self).__init__()
    self.left = left
    self.right = right
    self.leftDurationSelector = leftDurationSelector
    self.rightDurationSelector = rightDurationSelector
    self.resultSelector = resultSelector


  def run(self, observer, cancel, setSink):
    sink = self.Sink(self, observer, cancel)
    setSink(sink)
    return sink.run()

  class Sink(rx.linq.sink.Sink):
    def __init__(self, parent, observer, cancel):
      super(Join.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.gate = RLock()
      self.group = CompositeDisposable()
      self.refCount = RefCountDisposable(self.group)

      leftSubscription = SingleAssignmentDisposable()
      self.group.add(leftSubscription)