@extensionclassmethod(Observable)
def generate(cls, initial_state, condition, iterate, result_selector,
             scheduler=None):

    scheduler = scheduler or current_thread_scheduler

    def subscribe(observer):
        first = [True]
        state = [initial_state]

        def action (action1, state1=None):
            has_result = False
            result = None

            try:
                if first[0]:
                    first[0] = False
                else:
                    state[0] = iterate(state[0])

                has_result = condition(state[0])
                if has_result:
                    result = result_selector(state[0])

            except Exception as exception:
                observer.on_error(exception)
                return

            if has_result:
                observer.on_next(result)
                action1()
            else:
                observer.on_completed()

        return scheduler.schedule_recursive(action)
    return AnonymousObservable(subscribe)
Watching a forum and creating a new thread as myself should not
dataset/ETHPy150Open ReactiveX/RxPY/rx/linq/observable/generate.py/generate--crossover with--dataset/ETHPy150Open mozilla/kitsune/kitsune/kbforums/tests/test_notifications.py/NotificationsTests.test_watch_forum_then_new_thread_as_self