class 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();

}