How to execute two observables sequentially?

[lexer]

I've got to http request wrapped by observable

Observable<T1> request1 
Observable<T2> request2

I want to execute them sequentially.

[abersnaze]

This will do it.

Observable.concat(request1, request2)

[benjchristensen]

Yup, concat is the one to use.

[lexer]

I think concat actually don't work well for me.

Here is simplified problem:

Observable<String> geocodeAddressRequest 

returns address required to create Ride

Observable<Ride> createRideRequest = api.createRide(address)  

i need address from geoCodeAddressRequest to create it.

I want to have Observable geocodeAddresssThenCreateRideRequest

[benjchristensen]

If they are dependent on each other, then map/flatMap off the first into the second.

geocodeAddressRequest.flatMap( address -> {
   return api.createRide(address)
})

[lexer]

Thanks! Just realized same thing and you replied =)

[lexer]

We need to mark this as question.

[weituotian]

@benjchristensen @lexer @abersnaze

concat is not well !

        Observable<HistoryVideo> insert = historyVideoDao.insert(historyVideo);
        Observable<Long> count = historyVideoDao.count();
        Observable.concat(insert, count)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<Object>() {
                    @Override
                    public void call(Object o) {

                    }
                });

the example above have two Observable,
when concat two
the Action1 's param become Object
i can not figure out who is HistoryVideo and who is long

does rxjava have a solution to do that?

[akarnokd]

When you have such unrelated types, Java can't find any common ancestor other than Object. Why do you want to mix types on a stream?

[weituotian]

@akarnokd

i just want to excute Observals one after another

but i realize the style below is not so good

        historyVideoDao.insert(historyVideo)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<HistoryVideo>() {
                    @Override
                    public void call(HistoryVideo historyVideo1) {
                        
                        //do something handle historyVideo1
                        //and next i want to excute another observable

                        Observable<Long> count = historyVideoDao.count();
                        count.observeOn(AndroidSchedulers.mainThread())
                                .subscribe(new Action1<Long>() {
                                    @Override
                                    public void call(Long aLong) {
                                        // do something to handle along
     
                                    }
                                });

                    }
                });

[akarnokd]

concat is the right operator but due to limits of the Java language, you have to work with Objects and use instanceof + cast to get back the right data type.

[weituotian]

@akarnokd thanks to you!
i try instanceof and cast , it work fine to me!
but i am confused is there a way not have to mix types?

[Zhuinden]

I have a possibly silly but otherwise related question;

I have a single that is subscribed like this:

public void dispatch(Action action) {
        reducer.reduce(state.getValue(), action) // <-- is `Single<State>`
                .doOnSuccess(newState -> state.accept(newState) // <-- write to BehaviorRelay
                .subscribe();

However, I'd like to be able to force serial execution for these Singles, but I don't know the next Single at this time, so I cannot concat() them before subscribe().

How would I approach forcing a "one-at-a-time" execution, and buffer the Singles until this particular Single is executed?

I've been looking all over the net but I found nothing of value.

I imagine I might need to queue the Singles and "remove them from the queue" as side-effect, but then that needs to handle concurrency Rx-style? I dunno.

---

---

EDIT: Okay I've just realized that the solution is to use an event stream that has an unbounded buffer.

So my problem is that I am executing multiple independent Singles, instead of sending multiple events through an Observable stream, which is already buffered multiple events.

Turning Single into Observable seems to be working, thank you, apparently this is what concatMap is for

---

---

EDIT: Okay, Observables aren't chainable because it returns an ObservableSource instead of Observable, so you need to use Flowable instead for that.

[bugparty]

if you intent to execute one after another, use concatMap, if you need use both result, zip will be used too