AwsObservableExt { public static Completable asyncActionCompletable(Func1<CompletableHandlerSupplier, Future<?>> action) { return Completable.create(subscriber -> { try { final Future<?> result = action.call(new CompletableHandlerSupplier(subscriber)); subscriber.onSubscribe(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 asyncActionCompletable() throws Exception { AmazonWebServiceRequest someRequest = AmazonWebServiceRequest.NOOP; final MockAsyncClient<AmazonWebServiceRequest, String> client = new MockAsyncClient<>(someRequest, "some response"); final Completable completable = AwsObservableExt.asyncActionCompletable(factory -> client.someAsyncOperation(factory.handler( (req, resp) -> { assertEquals(someRequest, req); assertEquals("some response", resp); }, (t) -> { throw new IllegalStateException("Should never be here"); } ))); TestScheduler testScheduler = Schedulers.test(); final AssertableSubscriber<Void> subscriber = completable.subscribeOn(testScheduler).test(); testScheduler.triggerActions(); subscriber.assertNotCompleted(); client.run(); testScheduler.triggerActions(); subscriber.assertCompleted(); }
@Test public void asyncActionCompletableErrors() { 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 Completable completable = AwsObservableExt.asyncActionCompletable(factory -> client.throwException(exception)); TestScheduler testScheduler = Schedulers.test(); final AssertableSubscriber<Void> subscriber = completable.subscribeOn(testScheduler).test(); testScheduler.triggerActions(); subscriber.assertError(exception); }