RxPY - How to use stop_and_wait?

Question

I tried to translate the following JavaScript code example in Python:

import Rx from "rx"

let source = Rx.Observable.interval(1000)
  .timestamp()
  .controlled();

source.stopAndWait().subscribe(
  (result) => console.log("onNext: ", result),
  (error) => console.log("onError: ", error),
  () => console.log("Done!")
);

The snippet was taken from the RxJS Release Notes. My Python interpretation looks like the following:

from __future__ import print_function
from rx import Observable


source = Observable.interval(1000).timestamp().controlled()
source.stop_and_wait().subscribe(
    on_next=lambda x: print("on_next %s" % x),
    on_error=lambda e: print("on_error %s" % e)
)

Unfortunately, while JavaScript version works just fine the Python version fails due to the fact that 'StopAndWaitObservable' object has no attribute 'subscription'.

Answer 1

A fix for the issue was merged into the rx library development line. With the fix Python version works just like JavaScript one.