# Path: rx/disposable.py
# class CompositeDisposable(Cancelable):
#   """Represents a group of disposable resources that
#   are disposed together"""
# 
#   def __init__(self, *disposables):
#     super(CompositeDisposable, self).__init__()
# 
#     if len(disposables) == 0:
#       self.disposables = []
#     elif len(disposables) == 1 and not isinstance(disposables[0], Disposable):
#       self.disposables = [d for d in disposables[0] if d != None]
#     else:
#       self.disposables = [d for d in disposables if d != None]
# 
#     self.length = len(self.disposables)
# 
#   def add(self, disposable):
#     shouldDispose = False
# 
#     with self.lock:
#       shouldDispose = self.isDisposed
# 
#       if not shouldDispose:
#         self.disposables.append(disposable)
#         self.length += 1
# 
#     if shouldDispose:
#       disposable.dispose()
# 
#   def contains(self, disposable):
#     with self.lock:
#       return self.disposables.index(disposable) >= 0
# 
#   def remove(self, disposable):
#     shouldDispose = False
# 
#     with self.lock:
#       if self.isDisposed:
#         return False
# 
#       index = self.disposables.index(disposable)
# 
#       if index < 0:
#         return False
# 
#       shouldDispose = True
# 
#       self.disposables.remove(disposable)
#       self.length -= 1
# 
#     if shouldDispose:
#       disposable.dispose()
# 
#     return shouldDispose
# 
#   def clear(self):
#     disposables = []
# 
#     with self.lock:
#       disposables = self.disposables
#       self.disposables = []
#       self.length = 0
# 
#     for disposable in disposables:
#       disposable.dispose()
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       self.clear()
# 
# class SingleAssignmentDisposable(Cancelable):
#   """Represents a disposable resource which only allows
#   a single assignment of its underlying disposable resource.
#   If an underlying disposable resource has already been set,
#   future attempts to set the underlying disposable resource
#   will throw an Error."""
# 
#   def __init__(self):
#     super(SingleAssignmentDisposable, self).__init__()
#     self.current = None
# 
#   def disposable():
#     def fget(self):
#       with self.lock:
#         if self.isDisposed:
#           return Disposable.empty()
#         else:
#           return self.current
#     def fset(self, value):
#       old = self.current
#       self.current = value
# 
#       if old != None: raise Exception("Disposable has already been assigned")
# 
#       if self.isDisposed and value != None: value.dispose()
#     return locals()
# 
#   disposable = property(**disposable())
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       if self.current != None:
#         self.current.dispose()
# 
# Path: rx/internal.py
# 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__)
# 
# Path: rx/observable.py
# class Producer(Observable):
#   """Base class for implementation of query operators, providing
#   performance benefits over the use of Observable.Create"""
# 
#   def subscribeCore(self, observer):
#     return self.subscribeRaw(observer, True)
# 
#   def subscribeRaw(self, observer, enableSafequard):
#     sink = SingleAssignmentDisposable()
#     subscription = SingleAssignmentDisposable()
# 
#     d = CompositeDisposable(sink, subscription)
# 
#     if enableSafequard:
#       observer = AutoDetachObserver(observer, d)
# 
#     def assignSink(s):
#       sink.disposable = s
# 
#     def scheduled():
#       subscription.disposable = self.run(observer, subscription, assignSink)
#       return Disposable.empty()
# 
#     if Scheduler.currentThread.isScheduleRequired():
#       Scheduler.currentThread.schedule(scheduled)
#     else:
#       scheduled()
# 
#     return d
# 
#   def run(self, observer, cancel, setSink):
#     raise NotImplementedError()


from rx.disposable import CompositeDisposable, SingleAssignmentDisposable
from rx.internal import Struct
from rx.observable import Producer
from collections import deque
import rx.linq.sink
      res = list(self.queue)

      self.observer.onNext(res)
      self.observer.onCompleted()
      self.dispose()


class TakeLastBufferTime(Producer):
  def __init__(self, source, duration, scheduler):
    self.source = source
    self.duration = duration
    self.scheduler = scheduler

  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(TakeLastBufferTime.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.subscription = SingleAssignmentDisposable()
      self.loop = SingleAssignmentDisposable()

      self.startTime = self.parent.scheduler.now()
      self.subscription.disposable = self.parent.source.subscribeSafe(self)


# Path: rx/disposable.py
# class CompositeDisposable(Cancelable):
#   """Represents a group of disposable resources that
#   are disposed together"""
# 
#   def __init__(self, *disposables):
#     super(CompositeDisposable, self).__init__()
# 
#     if len(disposables) == 0:
#       self.disposables = []
#     elif len(disposables) == 1 and not isinstance(disposables[0], Disposable):
#       self.disposables = [d for d in disposables[0] if d != None]
#     else:
#       self.disposables = [d for d in disposables if d != None]
# 
#     self.length = len(self.disposables)
# 
#   def add(self, disposable):
#     shouldDispose = False
# 
#     with self.lock:
#       shouldDispose = self.isDisposed
# 
#       if not shouldDispose:
#         self.disposables.append(disposable)
#         self.length += 1
# 
#     if shouldDispose:
#       disposable.dispose()
# 
#   def contains(self, disposable):
#     with self.lock:
#       return self.disposables.index(disposable) >= 0
# 
#   def remove(self, disposable):
#     shouldDispose = False
# 
#     with self.lock:
#       if self.isDisposed:
#         return False
# 
#       index = self.disposables.index(disposable)
# 
#       if index < 0:
#         return False
# 
#       shouldDispose = True
# 
#       self.disposables.remove(disposable)
#       self.length -= 1
# 
#     if shouldDispose:
#       disposable.dispose()
# 
#     return shouldDispose
# 
#   def clear(self):
#     disposables = []
# 
#     with self.lock:
#       disposables = self.disposables
#       self.disposables = []
#       self.length = 0
# 
#     for disposable in disposables:
#       disposable.dispose()
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       self.clear()
# 
# class SingleAssignmentDisposable(Cancelable):
#   """Represents a disposable resource which only allows
#   a single assignment of its underlying disposable resource.
#   If an underlying disposable resource has already been set,
#   future attempts to set the underlying disposable resource
#   will throw an Error."""
# 
#   def __init__(self):
#     super(SingleAssignmentDisposable, self).__init__()
#     self.current = None
# 
#   def disposable():
#     def fget(self):
#       with self.lock:
#         if self.isDisposed:
#           return Disposable.empty()
#         else:
#           return self.current
#     def fset(self, value):
#       old = self.current
#       self.current = value
# 
#       if old != None: raise Exception("Disposable has already been assigned")
# 
#       if self.isDisposed and value != None: value.dispose()
#     return locals()
# 
#   disposable = property(**disposable())
# 
#   def dispose(self):
#     if not self._isDisposed.exchange(True):
#       if self.current != None:
#         self.current.dispose()
# 
# Path: rx/internal.py
# 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__)
# 
# Path: rx/observable.py
# class Producer(Observable):
#   """Base class for implementation of query operators, providing
#   performance benefits over the use of Observable.Create"""
# 
#   def subscribeCore(self, observer):
#     return self.subscribeRaw(observer, True)
# 
#   def subscribeRaw(self, observer, enableSafequard):
#     sink = SingleAssignmentDisposable()
#     subscription = SingleAssignmentDisposable()
# 
#     d = CompositeDisposable(sink, subscription)
# 
#     if enableSafequard:
#       observer = AutoDetachObserver(observer, d)
# 
#     def assignSink(s):
#       sink.disposable = s
# 
#     def scheduled():
#       subscription.disposable = self.run(observer, subscription, assignSink)
#       return Disposable.empty()
# 
#     if Scheduler.currentThread.isScheduleRequired():
#       Scheduler.currentThread.schedule(scheduled)
#     else:
#       scheduled()
# 
#     return d
# 
#   def run(self, observer, cancel, setSink):
#     raise NotImplementedError()


# Path: rx/linq/takeLastBuffer.py
from rx.disposable import CompositeDisposable, SingleAssignmentDisposable
from rx.internal import Struct
from rx.observable import Producer
from collections import deque
import rx.linq.sink

      res = list(self.queue)

      self.observer.onNext(res)
      self.observer.onCompleted()
      self.dispose()


class TakeLastBufferTime(Producer):
  def __init__(self, source, duration, scheduler):
    self.source = source
    self.duration = duration
    self.scheduler = scheduler

  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(TakeLastBufferTime.Sink, self).__init__(observer, cancel)
      self.parent = parent

    def run(self):
      self.subscription = SingleAssignmentDisposable()
      self.loop = SingleAssignmentDisposable()

      self.startTime = self.parent.scheduler.now()
      self.subscription.disposable = self.parent.source.subscribeSafe(self)


      return CompositeDisposable(self.subscription, self.loop)