GsUtil { public static <T> Func1<List<AsyncResult<T>>, Observable<T>> asyncResultListToObservable() { return new Func1<List<AsyncResult<T>>, Observable<T>>() { @Override public Observable<T> call(final List<AsyncResult<T>> asyncRresults) { return Observable.create(new OnSubscribe<T>() { @Override public void call(Subscriber<? super T> subscriber) { for (AsyncResult<T> asyncInvocationResponse : asyncRresults) { if (asyncInvocationResponse.getException() != null) { subscriber.onError(asyncInvocationResponse.getException()); return; } subscriber.onNext(asyncInvocationResponse.getResult()); } subscriber.onCompleted(); } }); }}; }  @Deprecated static List<AstrixRemoteResult<T>> convertToAstrixRemoteResults(SpaceRemotingResult<T>[] results); static void subscribe(final AsyncFuture<T> asyncFuture, final Subscriber<? super T> t1); static Func1<List<AsyncResult<T>>, Observable<T>> asyncResultListToObservable();  }
@Test public void asyncResultList_ToFlatObservable_EmitsEachResultAsSingleItem() throws Exception { Func1<List<AsyncResult<String>>, Observable<String>> listToObservable = GsUtil.asyncResultListToObservable(); List<AsyncResult<String>> r = Arrays.<AsyncResult<String>> asList(new DefaultAsyncResult<String>("foo", null), new DefaultAsyncResult<String>("bar", null)); Observable<String> result = Observable.just(r).flatMap(listToObservable); assertEquals(Arrays.asList("foo", "bar"), result.toList().toBlocking().first()); }
@Test public void asyncResultList_ToFlatObservable_Error() throws Exception { Func1<List<AsyncResult<String>>, Observable<String>> listToObservable = GsUtil.asyncResultListToObservable(); List<AsyncResult<String>> r = Arrays.<AsyncResult<String>> asList(new DefaultAsyncResult<String>("foo", null), new DefaultAsyncResult<String>(null, new TestException())); Observable<String> result = Observable.just(r).flatMap(listToObservable); Iterator<String> resultIterator = result.toBlocking().getIterator(); resultIterator.next(); try { resultIterator.next(); fail("Expected TestException to be thrown"); } catch (TestException e) { } }