java - 使用另一个 Observable<List<AnotherObject>> 过滤 Observable<List<Object>>

Question

有什么方法可以实现以下目标吗？

我有两个可观察对象，

• Observable<List<String>> broadcastIdsObservable .
• Observable<List<EventData>> eventDatasObservable .

有没有办法用broadcastIdsObsevable内部的List来过滤eventDatasObservable？我需要获取List<EventData>与 eventData满足条件eventData.getBroadcastId().equals(broadCastId)哪里broadCastId是broadcastIds的每一项。

我尝试了使用 flatMap、map 和 Observable.concat 的不同方法...但我无法实现它。

提前致谢。

Answer 1

您可以使用RxJavaJoins 。您需要包括

compile 'io.reactivex:rxjava-joins:0.22.0'

然后创建联接:

@NonNull
@SuppressWarnings("unchecked")
private Func0<Observable<List<EventData>>> loadAndFilterEvents() {
    return new Func0<Observable<List<EventData>>>() {

        @Override
        public Observable call() {

            Observable<List<String>> broadcastIdsObservable = ...;
            Observable<List<EventData>> eventDatasObservable = ...;

            Plan0<List<EventData>> plan = JoinObservable.from(broadcastIdsObservable)
                                                        .and(eventDatasObservable)
                                                        .then(filterJoin);
            return JoinObservable.when(plan).toObservable();
        }

        private Func2<List<String>, List<EventData>, List<EventData>> filterJoin = (broadcastIds, events) -> {
            List<EventData> filtered = new ArrayList<>();
            for (EventData event : events) {
                if (broadcastIds.contains(event.getBroadcastId())) {
                    filtered.add(event);
                }
            }
            return filtered;
        };

    };
}

并订阅它:

Observable.defer(loadAndFilterEvents())
              .subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe(filteredEvents -> Timber.d("Events: %s", filteredEvents),
                         Timber::e);