Adapter Subscribing to Multiple Observables

Question

My question is What is the best way for an Adapter to subscribe to multiple observables

I have an Adapter that has a header and regular item types. The information for the adapter is fed from a database using Sqlbrite. Using Sqlbrite, I want to update the header and the list by having the adapter subscribe to an observable.

Not a problem for the first Observable. In my fragment's OnResume:

    mSubscription = TrackerDbUtils.getListSubmissionObservable(db)
            .map(Submission.MAP)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(mAdapter);

However, I then try to add a second Observable to the fragment

    mHeaderSubscription = TrackerDbUtils.getSummaryObservable(db)
            .map(Summary.MAP)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(mAdapter);

So, I need to implement Action1 in my Adapter:

public class MileageTrackerAdapter extends HeaderRecyclerViewAdapter
    implements Action1<List<Submission>>, Action1<Summary> { 
    .
    .
    .

}

However, I get a duplicate class error. So I tried to implement a generic type:

public class MileageTrackerAdapter extends HeaderRecyclerViewAdapter
    implements Action1<Object> { 
    .
    .
    .
    @Override
    public void call(Object o) {
        if(o instanceof Summary){
              // Logic
        } else {
              // Logic
        }
    }
}

But I get a runtime error that I am unable to Map a Summary to Submission. Even then, I am not sure that casting a generic is the best approach to the problem.

Based on feedback from dwursteisen:

I changed my Summary class to map to Submission. I also created a new field called "isSummary." I changed my observable to the following as suggested:

TrackerDbUtils.getListSubmissionObservable(db)
    .map(Submission.MAP)
    .mergeWith(TrackerDbUtils.getSummaryObservable(db)
                              .map(Summary.MAP))
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(mAdapter);

This creates an Observable that will fire twice; once for the bulk list of submissions and the other as a List with a size of 1. This list is actually the Summary. In my adapter:

@Override
public void call(List<Submission> mSubmissions) {

    if(mSubmissions.size() == 1 && mSubmissions.get(0).isSummary()){
        mSummary = mSubmissions.get(0);
    } else {
        mArray = new ArrayList<>(mSubmissions);
    }

    notifyDataSetChanged();

}

score 2 · Accepted Answer · answered Jun 07 '15 at 22:22

2

You should avoid using the same adapter for two different stream, as it will have to deal concurrent issue, that you can avoid using RxJava.

Why not try to merge your streams into a single one, and, then, use only one adapter ?

TrackerDbUtils.getListSubmissionObservable(db)
        .map(Submission.MAP)
        .mergeWith(TrackerDbUtils.getSummaryObservable(db)
                                  .map(Summary.MAP))
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(mAdapter);

You may need to map List<Submission> and Summary to a same common type, or deal with Object into your adapter

answered Jun 07 '15 at 22:22

dwursteisen

11,435
2
39
36

Could you elaborate on the concurrent issue? I'm new to RxJava, so I was under the impression that subscribing to multiple observables was a perk of RxJava. Is this not the case? – Tykin Jun 09 '15 at 03:46
The onNext method of your subscriber/observer may be called in the same time as each observable that share the same subscriber/observer may observe on different thread context. – dwursteisen Jun 09 '15 at 07:32

score 1 · Answer 2 · answered Jun 07 '15 at 22:05

Maybe someone will give a better answer, but in situations like this, I usually create public final anonymous instance of these interfaces. So the code looks something like this:

public class Adapter extends HeaderRecyclerViewAdapter {

    public final Action1<Summary> summary = new Action1<Summary> { ... }
    public final Action1<List<Submission>> submissions = new Action1<List<Submission>> { ... }

}

This way, the methods inside both Action1 implementations can access everything in the Adapter class as if they were normal methods and I can distinguish between them at the same time by using mAdapter.summary or mAdapter.submissions

Adapter Subscribing to Multiple Observables

2 Answers2