class RxEventSources {

  @SafeVarargs
  public static <E> EventSource<E> fromObservables(Observable<E>... observables) {
    final Observable<E> eventSource = Observable.merge(observables);

    return new EventSource<E>() {
      @Nonnull
      @Override
      public Disposable subscribe(final Consumer<E> eventConsumer) {
        final Subscription subscription =
            eventSource.subscribe(
                new Observer<E>() {
                  @Override
                  public void onCompleted() {
                    // TODO: complain loudly! shouldn't ever complete
                  }

                  @Override
                  public void onError(Throwable e) {
                    RxJavaHooks.onError(e);
                  }

                  @Override
                  public void onNext(E e) {
                    eventConsumer.accept(e);
                  }
                });

        return new Disposable() {
          @Override
          public void dispose() {
            subscription.unsubscribe();
          }
        };
      }
    };
  }

  private  RxEventSources();

}