from rx.observable import Observable
from rx.anonymousobservable import AnonymousObservable
from rx.concurrency import current_thread_scheduler
from rx.internal import extensionclassmethod


@extensionclassmethod(Observable, alias=["from_", "from_list"])
def from_iterable(cls, iterable, scheduler=None):
    """Converts an array to an observable sequence, using an optional
    scheduler to enumerate the array.

    1 - res = rx.Observable.from_iterable([1,2,3])
    2 - res = rx.Observable.from_iterable([1,2,3], rx.Scheduler.timeout)

    Keyword arguments:
    :param Observable cls: Observable class
    :param Scheduler scheduler: [Optional] Scheduler to run the
        enumeration of the input sequence on.

    :returns: The observable sequence whose elements are pulled from the
        given enumerable sequence.
    :rtype: Observable
    """

    scheduler = scheduler or current_thread_scheduler

    def subscribe(observer):
        iterator = iter(iterable)

        def action(action1, state=None):
            try:
                item = next(iterator)
            except StopIteration:
                observer.on_completed()
            else:
                observer.on_next(item)
                action1(action)

        return scheduler.schedule_recursive(action)
    return AnonymousObservable(subscribe)

@@ -137,16 +137,75 @@
 heduler%0A
+from rx.disposables import Disposable, CompositeDisposable%0A
 from rx.
@@ -1026,16 +1026,46 @@
 terable)
+%0A        is_disposed = %5BFalse%5D
 %0A%0A      
@@ -1095,24 +1095,78 @@
 tate=None):%0A
+            if is_disposed%5B0%5D:%0A                return%0A
             
@@ -1380,22 +1380,28 @@
         
-return
+cancelable =
  schedul
@@ -1430,16 +1430,144 @@
 action)%0A
+%0A        def dispose():%0A            is_disposed%5B0%5D = True%0A%0A        return CompositeDisposable(cancelable, Disposable(dispose))%0A%0A
     retu

Terminate Observable.from_iterable on dispose