RxConnectables { @NonNull public static <I, O> ObservableTransformer<I, O> toTransformer( final Connectable<I, O> connectable) { return new ObservableTransformer<I, O>() { @Override public @NonNull ObservableSource<O> apply(@NonNull Observable<I> upstream) { return Observable.create( new ObservableOnSubscribe<O>() { @Override public void subscribe(@NonNull ObservableEmitter<O> emitter) throws Throwable { com.spotify.mobius.functions.Consumer<O> output = emitter::onNext; final Connection<I> input = connectable.connect(output); final Disposable disposable = upstream.subscribe( new Consumer<I>() { @Override public void accept(I value) throws Throwable { input.accept(value); } }, new Consumer<Throwable>() { @Override public void accept(Throwable error) throws Throwable { emitter.onError(error); } }, new Action() { @Override public void run() throws Throwable { emitter.onComplete(); } }); emitter.setCancellable( new Cancellable() { @Override public void cancel() throws Throwable { disposable.dispose(); input.dispose(); } }); } }); } }; } private  RxConnectables(); static Connectable<I, O> fromTransformer(
      @NonNull final ObservableTransformer<I, O> transformer); @NonNull static ObservableTransformer<I, O> toTransformer(
      final Connectable<I, O> connectable);  }
@Test public void shouldPropagateCompletion() throws Exception { TestObserver<Integer> observer = input.compose(RxConnectables.toTransformer(connectable)).test(); input.onNext("hi"); input.onComplete(); observer.awaitDone(1, TimeUnit.SECONDS); observer.assertComplete(); }
@Test public void shouldPropagateErrorsFromConnectable() throws Exception { TestObserver<Integer> observer = input.compose(RxConnectables.toTransformer(connectable)).test(); input.onNext("crash"); observer.awaitDone(1, TimeUnit.SECONDS); observer.assertError(throwable -> throwable.getMessage().equals("crashing!")); }
@Test public void shouldPropagateErrorsFromUpstream() throws Exception { final Throwable expected = new RuntimeException("expected"); TestObserver<Integer> observer = input.compose(RxConnectables.toTransformer(connectable)).test(); input.onError(expected); observer.awaitDone(1, TimeUnit.SECONDS); observer.assertError(expected); }