RxConnectables { public static <I, O> ObservableTransformer<I, O> toTransformer( final Connectable<I, O> connectable) { return new ObservableTransformer<I, O>() { @Override public ObservableSource<O> apply(final Observable<I> upstream) { return Observable.create( new ObservableOnSubscribe<O>() { @Override public void subscribe(final ObservableEmitter<O> emitter) throws Exception { Consumer<O> output = new Consumer<O>() { @Override public void accept(O value) { emitter.onNext(value); } }; final Connection<I> input = connectable.connect(output); final Disposable disposable = upstream.subscribe( new io.reactivex.functions.Consumer<I>() { @Override public void accept(I f) { input.accept(f); } }, new io.reactivex.functions.Consumer<Throwable>() { @Override public void accept(Throwable throwable) throws Exception { emitter.onError(throwable); } }, new Action() { @Override public void run() { emitter.onComplete(); } }); emitter.setCancellable( new Cancellable() { @Override public void cancel() throws Exception { disposable.dispose(); input.dispose(); } }); } }); } }; } private  RxConnectables(); static Connectable<I, O> fromTransformer(
      final ObservableTransformer<I, O> transformer); 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.awaitTerminalEvent(1, TimeUnit.SECONDS); observer.assertComplete(); }
@Test public void shouldPropagateErrorsFromConnectable() throws Exception { TestObserver<Integer> observer = input.compose(RxConnectables.toTransformer(connectable)).test(); input.onNext("crash"); observer.awaitTerminalEvent(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.awaitTerminalEvent(1, TimeUnit.SECONDS); observer.assertError(expected); }