HystrixObservableCommandFacade { public static <T> Observable<T> observe(final Supplier<Observable<T>> observableFactory, Setter settings) { Observable<Result<T>> faultToleranceProtectedObservable = new HystrixObservableCommand<Result<T>>(settings) { @Override protected Observable<Result<T>> construct() { return observableFactory.get().map(new Func1<T, Result<T>>() { @Override public Result<T> call(T t1) { return Result.success(t1); } }).onErrorResumeNext(new Func1<Throwable, Observable<? extends Result<T>>>() { @Override public Observable<? extends Result<T>> call(Throwable t1) { if (t1 instanceof ServiceUnavailableException) { return Observable.error(t1); } return Observable.just(Result.<T>exception(t1)); } }); } @Override protected Observable<Result<T>> resumeWithFallback() { return Observable.just(Result.<T>exception(createServiceUnavailableException())); } private ServiceUnavailableException createServiceUnavailableException() { if (isResponseRejected()) { return new ServiceUnavailableException(String.format("cause=%s service=%s", "REJECTED_EXECUTION", getCommandKey().name())); } if (isResponseTimedOut()) { return new ServiceUnavailableException(String.format("cause=%s service=%s executionTime=%s", "TIMEOUT", getCommandKey().name(), getExecutionTimeInMilliseconds())); } if (isResponseShortCircuited()) { return new ServiceUnavailableException(String.format("cause=%s service=%s", "SHORT_CIRCUITED", getCommandKey().name())); } if (isFailedExecution() && (getFailedExecutionException() instanceof ServiceUnavailableException)) { ServiceUnavailableException result = (ServiceUnavailableException) getFailedExecutionException(); appendStackTrace(result, new ServiceUnavailableException(String.format("service=%s", getCommandKey().name()))); return result; } return new ServiceUnavailableException(String.format("cause=%s service=%s", "UNKNOWN", getCommandKey().name())); } }.observe(); return faultToleranceProtectedObservable .flatMap(resultWrapper -> resultWrapper.toObservable()) .onErrorResumeNext(error -> { if (error instanceof HystrixRuntimeException) { HystrixRuntimeException e = (HystrixRuntimeException) error; if (e.getCause() != null) { return Observable.error(e.getCause()); } return Observable.error(new ServiceUnavailableException(e.getFailureType().toString())); } return Observable.error(error); }); }  static Observable<T> observe(final Supplier<Observable<T>> observableFactory, Setter settings);  }
@Test public void subscribesEagerlyToCreatedObserver() throws Exception { AtomicBoolean subscribed = new AtomicBoolean(false); Supplier<Observable<String>> timeoutCommandSupplier = new Supplier<Observable<String>>() { @Override public Observable<String> get() { return Observable.create(t1 -> { subscribed.set(true); }); } }; HystrixObservableCommandFacade.observe(timeoutCommandSupplier, commandSettings); assertTrue(subscribed.get()); }
@Test public void doesNotInvokeSupplierWhenBulkHeadIsFull() throws Exception { final AtomicInteger supplierInvocationCount = new AtomicInteger(); Supplier<Observable<String>> timeoutCommandSupplier = new Supplier<Observable<String>>() { @Override public Observable<String> get() { supplierInvocationCount.incrementAndGet(); return Observable.create(new OnSubscribe<String>() { @Override public void call(Subscriber<? super String> t1) { } }); } }; Observable<String> ftObservable1 = HystrixObservableCommandFacade.observe(timeoutCommandSupplier, commandSettings); final Observable<String> ftObservable2 = HystrixObservableCommandFacade.observe(timeoutCommandSupplier, commandSettings); ftObservable1.subscribe(); assertEquals(1, supplierInvocationCount.get()); AstrixTestUtil.serviceInvocationException(() -> ftObservable2.toBlocking().first(), AstrixTestUtil.isExceptionOfType(ServiceUnavailableException.class)); assertEquals(1, supplierInvocationCount.get()); }