Rxjava: 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

[georgious070]

@weituotian can you share the example, because i have the same issue

[sunshinfight]

Thanks a lot, I have puzzled by this problem much time!

[Zhuinden]

Actually, I should have posted my final code, maybe it helps someone

    private Flowable<State> traverseAfterChain(Flowable<State> stateFlowable, Action action, int index) {
         if(index < 0) {
             return stateFlowable;
         }
         final Middleware middleware = middlewares.get(index);
         if(stateFlowable == null) {
             stateFlowable = middleware.doAfter().reduce(state.getValue(), action);
         } else {
             stateFlowable = stateFlowable.concatMap(newState -> middleware.doAfter().reduce(newState, action));
         }
         return traverseAfterChain(stateFlowable, action, index - 1);
      }           
}

then later I seem to have changed it to something like

    Single<Pair<State, Action>> executeAfter(ReduxStore reduxStore, State state, Action action) {
        Interception interception = doAfter();
        if(interception == null) {
            return Single.just(Pair.with(state, action));
        } else {
            return interception.intercept(reduxStore, action)
                    .flatMap(newAction -> Single.just(Pair.with(state, newAction)));
        }
    }

It seems to have been 9 months ago and I haven't touched it since 😄

[Basel78]

Hello,
I am facing same case described by lexer, and the solution proposed
geocodeAddressRequest.flatMap( address -> {
return api.createRide(address)
})
is throwing the error :

geocodeAddressRequest.flatMap() is not a function

both functions have the pseudo code like:

function geocodeAddressRequest (text:string):Observable\ {
return this.http.post('url',body,options).map((res) => { res.josn()})
}

Any suggestion to fix this error

[akarnokd]

Did you forget the parenthesis after geocodeAddressRequest, i.e.: geocodeAddressRequest().flatMap(...)?

[Basel78]

Its there this.getAccessToken(returnUrl,code).flatMap(..)
Actually,when I use this.getAccessToken(returnUrl,code).SwitchMap its working, but the issue with SwitchMap I think it returns an empty value since the second function is not called, maybe coz unsubscribed after the first Observable occurred?

[Basel78]

After check and comparing to some old functions, I found the reason why it throw this error,which is missing the references for the function :(
import 'rxjs/add/operator/mergeMap';
import 'rxjs/add/operator/catch';

[scamexdotexe]

how about if I wanted to process first the result from first observable rather than sending it directly to second observable?

[Zhuinden]

@scamexdotexe probably something like

firstObservable
    .flatMap { firstItem -> secondObservable.map { secondItem -> firstItem to secondItem } }
    .doOnNext { (first, second) -> ...

[abersnaze]

This thread is closed, very old, and these questions about usage are probably better asked on stack overflow. If you find a bug or the question stumps SO then please open a new issue.