Sync two asynchronous API call with RxJava

Question

In what way can we sync two asynchronous calls using RxJava? In the example below, the method contentService.listContents which is a API call must first finish before the processSchema method to take place for each schema.

schemaService.listSchema()
    .toObservable()
    .flatMapIterable(schemas -> {
        schemas.forEach(schema -> {
            // async call
            contentService.listContents(schema.getName()).subscribe(contents -> {
                   doSomethingWithThe(contents); 
            });
        });
        // contentService.listContents` must complete first before 
        // processSchema should be called for each schema
        return schemas;
    }).subscribe(schema -> { processSchema(schema); }, 
                 error -> { Console.error(error.getMessage()); });

The problem with the code above the processSchema would not wait for the contentService.listContents since it is async not not synchronized with each other.

akarnokd · Accepted Answer · 2020-01-22 13:02:20Z

1

You have to use flatMap to process the schemas and since it is a list, you have to unroll it and flatMap again:

schemaService.listSchema()
.toObservable()
.flatMap(schemas -> 
     Observable.fromIterable(schemas)
     .flatMap(schema -> 
         contentService.listContents(schema.getName())
         .doOnNext(contents -> doSomethingWith(contents))
     )
     // probably you don't care about the inner contents
     .ignoreElements()
     // andThen will switch to this only when the sequence above completes
     .andThen(Observable.just(schemas))
)
.subscribe(
    schema -> processSchema(schema), 
    error -> Console.error(error.getMessage())
);

Note that you haven't defined the return types of the service calls so you may have to use flatMapSingle and doOnSuccess for example.

answered Jan 22, 2020 at 13:02

akarnokd

70.1k14 gold badges161 silver badges199 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Abdullah Mohammad Motiullah · Accepted Answer · 2020-01-21 05:34:17Z

0

You are probably looking for flatMap.

From the docs

Continuations

Sometimes, when an item has become available, one would like to perform some dependent computations on it. This is sometimes called continuations and, depending on what should happen and what types are involved, may involve various operators to accomplish.

Dependent

The most typical scenario is to given a value, invoke another service, await and continue with its result:

service.apiCall()
.flatMap(value -> service.anotherApiCall(value))
.flatMap(next -> service.finalCall(next))

It is often the case also that later sequences would require values from earlier mappings. This can be achieved by moving the outer flatMap into the inner parts of the previous flatMap for example:

service.apiCall()
.flatMap(value ->
    service.anotherApiCall(value)
    .flatMap(next -> service.finalCallBoth(value, next))
)

edited Jan 21, 2020 at 5:34

answered Jan 21, 2020 at 5:27

Abdullah Mohammad Motiullah

12.7k6 gold badges72 silver badges84 bronze badges

1 Comment

quarks Over a year ago

Note that the "value" in my case is an Iterable so your code does not show that

Collectives™ on Stack Overflow

Sync two asynchronous API call with RxJava

2 Answers 2

Comments

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related