AwsObservableExt { public static <REQ extends AmazonWebServiceRequest, RES> Single<RES> asyncActionSingle(Func1<SingleHandlerSupplier<RES>, Future<RES>> action) { return Single.create(subscriber -> { try { final Future<RES> result = action.call(new SingleHandlerSupplier<>(subscriber)); subscriber.add(Subscriptions.create(() -> result.cancel(true))); } catch (Throwable t) { Exceptions.throwIfFatal(t); subscriber.onError(t); } }); }  static Completable asyncActionCompletable(Func1<CompletableHandlerSupplier, Future<?>> action); static Single<RES> asyncActionSingle(Func1<SingleHandlerSupplier<RES>, Future<RES>> action);  }
@Test public void asyncActionSingle() throws Exception { AmazonWebServiceRequest someRequest = AmazonWebServiceRequest.NOOP; final MockAsyncClient<AmazonWebServiceRequest, String> client = new MockAsyncClient<>(someRequest, "some response"); Single<String> single = AwsObservableExt.asyncActionSingle(supplier -> client.someAsyncOperation(supplier.handler())); TestScheduler testScheduler = Schedulers.test(); final AssertableSubscriber<String> subscriber = single.subscribeOn(testScheduler).test(); testScheduler.triggerActions(); subscriber.assertNoValues(); subscriber.assertNotCompleted(); client.run(); testScheduler.triggerActions(); subscriber.assertValueCount(1); subscriber.assertValue("some response"); subscriber.assertCompleted(); }
@Test public void asyncActionSingleErrors() { AmazonWebServiceRequest someRequest = AmazonWebServiceRequest.NOOP; final MockAsyncClient<AmazonWebServiceRequest, String> client = new MockAsyncClient<>(someRequest, "some response"); RuntimeException exception = new RuntimeException("error when initiating an async operation"); final Single<String> completable = AwsObservableExt.asyncActionSingle(supplier -> client.throwException(exception)); TestScheduler testScheduler = Schedulers.test(); final AssertableSubscriber<String> subscriber = completable.subscribeOn(testScheduler).test(); testScheduler.triggerActions(); subscriber.assertError(exception); }