RxConnectables { public static <I, O> Observable.Transformer<I, O> toTransformer( final Connectable<I, O> connectable, final BackpressureMode backpressureMode) { return new Observable.Transformer<I, O>() { @Override public Observable<O> call(final Observable<I> upstream) { return Observable.create( new Action1<Emitter<O>>() { @Override public void call(final Emitter<O> emitter) { Consumer<O> output = new Consumer<O>() { @Override public void accept(O value) { emitter.onNext(value); } }; final Connection<I> input = connectable.connect(output); final Subscription subscription = upstream.subscribe( new Action1<I>() { @Override public void call(I f) { input.accept(f); } }, new Action1<Throwable>() { @Override public void call(Throwable throwable) { emitter.onError(throwable); } }, new Action0() { @Override public void call() { emitter.onCompleted(); } }); emitter.setCancellation( new Cancellable() { @Override public void cancel() throws Exception { subscription.unsubscribe(); input.dispose(); } }); } }, backpressureMode); } }; } private  RxConnectables(); static Connectable<I, O> fromTransformer(
      final Observable.Transformer<I, O> transformer); static Observable.Transformer<I, O> toTransformer(
      final Connectable<I, O> connectable, final BackpressureMode backpressureMode); static Observable.Transformer<I, O> toTransformer(
      final Connectable<I, O> connectable);  }
@Test public void shouldPropagateCompletion() throws Exception { AssertableSubscriber<Integer> subscriber = input.compose(RxConnectables.toTransformer(connectable)).test(); input.onNext("hi"); input.onCompleted(); subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); subscriber.assertCompleted(); }
@Test public void shouldPropagateErrorsFromConnectable() throws Exception { AssertableSubscriber<Integer> subscriber = input.compose(RxConnectables.toTransformer(connectable)).test(); input.onNext("crash"); subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); subscriber.assertFailureAndMessage(RuntimeException.class, "crashing!"); }
@Test public void shouldPropagateErrorsFromUpstream() throws Exception { final Throwable expected = new RuntimeException("expected"); AssertableSubscriber<Integer> subscriber = input.compose(RxConnectables.toTransformer(connectable)).test(); input.onError(expected); subscriber.awaitTerminalEvent(1, TimeUnit.SECONDS); subscriber.assertError(expected); }