Qfyilyi

Here is some RxJava code that I want to test:

public void triggerCancelOrderJob() {

        couchConnector()

            .findAbandonedOpenOrders()

            .flatMap(results -> results.rows())

            .flatMap(

                row ->

                    Observable.just(row)

                        .subscribeOn(Schedulers.io())

                        .map(

                            s -> return s.value())

                        .flatMap(

                            orderId -> {

                              return RxReactiveStreams.toObservable(

                                  serviceTokenCache

                                     .get(OrderApiConstants.SERVICE_TOKEN_CACHE_KEY)

                                      .flatMap(

                                          issueToken -> {

                                            return cancelOrderApiConnector()

                                                .invokeAPI(

                                                    RequestInputModel.builder().build(),

                                                    RequestInputModel.RequestBodyModel.builder().build());

                                          }));

                            }))

            .subscribe(//additional code)

So what's happening is that I run an async CB query, get an Observable< AsyncN1qlQueryResult >, then for each row I call call two external services one after the another (first call to the serviceTokenCache and second call to the cancelOrderApiConnector). Each row runs in a separate IO thread.

Note: serviceTokenCache.get() and cancelOrderApiConnector().invokeAPI() return a Mono respectively.

I cannot figure out how to test this code. What all components need to be tested? Since each row will run in its separate thread, I cannot wrap my head around how to test such asynchronous code.