code: public static <T, U> StoppableObservable<U> runAsync(Scheduler scheduler,
              final Subject<T, U> subject, 
              final Action2<? super Observer<? super T>, ? super Subscription> action) {
          final SerialSubscription csub = new SerialSubscription();
          
          StoppableObservable<U> co = new StoppableObservable<U>(new Observable.OnSubscribe<U>() {
              @Override
              public void call(Subscriber<? super U> t1) {
                  subject.subscribe(t1);
              }
          }, csub);
          
          final Worker inner = scheduler.createWorker();
          csub.set(inner);
          
          inner.schedule(new Action0() {
              @Override
              public void call() {
                  if (!csub.isUnsubscribed()) {
                      action.call(subject, csub);
                  }
              }
          });
          
          return co;
      }
docs: Runs the provided action on the given scheduler and allows propagation of multiple events to the
  observers of the returned StoppableObservable. The action is immediately executed and unobserved values
  might be lost, depending on the Subject type used.
  
  @param <T> the output value of the action
  @param <U> the output type of the observable sequence
  @param scheduler the Scheduler where the action is executed
  @param subject the subject to use to distribute values emitted by the action
  @param action the action to execute, receives an Observer where the events can be pumped and a
  Subscription which lets it check for cancellation condition
  @return an Observable that provides a Subscription interface to cancel the action
  @see <a href="https://github.com/ReactiveX/RxJava/wiki/Async-Operators#wiki-runasync">RxJava Wiki: runAsync()</a>