### Question:
FlowableOnSubscribeExecuteAsBlockingOptional implements FlowableOnSubscribe<Optional<Result>> { @Override public void subscribe(@NonNull FlowableEmitter<Optional<Result>> emitter) throws Exception { try { final Result value = preparedOperation.executeAsBlocking(); emitter.onNext(Optional.of(value)); emitter.onComplete(); } catch (Exception e) { emitter.onError(e); } } FlowableOnSubscribeExecuteAsBlockingOptional(@NonNull PreparedOperation<Result, Optional<Result>, Data> preparedOperation); @Override void subscribe(@NonNull FlowableEmitter<Optional<Result>> emitter);  }### Answer:
@SuppressWarnings("CheckResult") @Test public void shouldExecuteAsBlockingAfterSubscription() { final PreparedOperation<String, Optional<String>, String> preparedOperation = mock(PreparedOperation.class); final String expectedResult = "expected_string"; when(preparedOperation.executeAsBlocking()).thenReturn(expectedResult); TestSubscriber<Optional<String>> testSubscriber = new TestSubscriber<Optional<String>>(); Flowable .create(new FlowableOnSubscribeExecuteAsBlockingOptional<String, String>(preparedOperation), MISSING) .subscribe(testSubscriber); verify(preparedOperation).executeAsBlocking(); testSubscriber.assertValue(Optional.of(expectedResult)); testSubscriber.assertNoErrors(); testSubscriber.assertComplete(); }
@Test public void shouldCallOnError() { Throwable throwable = new IllegalStateException("Test exception"); PreparedOperation<String, Optional<String>, String> preparedOperation = mock(PreparedOperation.class); when(preparedOperation.executeAsBlocking()).thenThrow(throwable); TestSubscriber<Optional<String>> testSubscriber = TestSubscriber.create(); Flowable .create(new FlowableOnSubscribeExecuteAsBlockingOptional<String, String>(preparedOperation), MISSING) .subscribe(testSubscriber); testSubscriber.assertError(throwable); testSubscriber.assertNoValues(); testSubscriber.assertNotComplete(); }