将Observable收集到List中似乎不会立即发出集合 [英] Collecting Observables to a List doesn't seem to emit the collection at once
问题描述
我正在使用RxJava基本上收集单独发出的Observable列表，并将它们组合成一个Observable列表（基本上与flatMap相反）。这是我的代码：
I'm using RxJava to essentially collect the list of individually emitted Observables and combine them into a list of Observables (essentially sort of the opposite of flatMap). Here's my code:
// myEvent.findMemberships() returns an Observable<List<Membership>>
myEvent.findMemberships()
.flatMap(new Func1<List<Membership>, Observable<User>>() {
@Override
public Observable<User> call(List<Membership> memberships) {
List<User> users = new ArrayList<User>();
for (Membership membership : memberships) {
users.add(membership.getUser());
}
return Observable.from(users);
}
})
.toList()
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<List<User>>() {
@Override
public void onCompleted() { }
@Override
public void onError(Throwable e) {
Timber.e(e, "Error when trying to get memberships");
}
@Override
public void onNext(List<User> users) {
Timber.d("%d users emitted", users.size());
}
})
我注意到我的onNext永远不会被调用。我似乎无法理解这一点。如果我删除.toList调用并基本输出各个用户（如下所示），它通过发出每个项目来工作。
I notice that my onNext is never called. I can't seem to understand this. If i remove the .toList call and basically output the individual Users (as shown below) it works by emitting each item.
subscriptions //
.add(currentEvent.findMemberships()
.flatMap(new Func1<List<Membership>, Observable<User>>() {
@Override
public Observable<User> call(List<Membership> memberships) {
List<User> users = new ArrayList<User>();
for (Membership membership : memberships) {
users.add(membership.getUser());
}
return Observable.from(users);
}
})
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<User>() {
@Override
public void onCompleted() { }
@Override
public void onError(Throwable e) {
Timber.e(e, "Error when trying to get memberships");
}
@Override
public void onNext(User user) {
Timber.d("%d users emitted", user.getName());
}
}));
Q1。我对.toList的理解不正确吗？
Q1. Is my understanding of .toList incorrect?
Q2。如何将单独发出的 Observable< Object> 流组合成一个 Observable< List< Object>> ？
Q2. How does one combing a stream of individually emitted Observable<Object>s into a single Observable<List<Object>> ?
**编辑
@kjones完全解决了这个问题。我没有通过我的findMemberships电话呼叫onComplete。我在下面添加了代码段。我真正的用例有一些更多的转换，这就是为什么我需要使用.toList调用。正如@zsxwing也正确地指出的那样，对于这个用例，一个简单的地图就足够了。
@kjones totally nailed the issue. I was not calling onComplete with my findMemberships call. I've added the code snippet below. My real use case was a little more convoluted with a bunch of more transformations which is why I needed to be using .toList call. As @zsxwing also rightly pointed out, for just this use case, a simple map will suffice.
public Observable<List<Membership>> findMemberships() {
return Observable.create(new Observable.OnSubscribe<List<Membership>>() {
@Override
public void call(Subscriber<? super List<Membership>> subscriber) {
try {
// .....
List<Membership> memberships = queryMyDb();
subscriber.onNext(memberships);
// BELOW STATEMENT FIXES THE PROBLEM ><
// subscriber.onCompleted();
} catch (SQLException e) {
// ...
}
}
});
}
推荐答案
看起来 myEvent.findMemberships（）调用返回的Observable永远不会调用onComplete。你能展示这段代码吗？
It looks like the Observable returned by the myEvent.findMemberships() call is never calling onComplete. Can you show this code?
如果是这种情况，它会解释你所看到的行为。 .toList（）在发出所有项目之前不会发出列表（由onComplete发出信号）。
If that is the case, it would explain the behavior you are seeing. The .toList() will not emit the list until all items have been emitted (signaled by onComplete).
没有 .toList（）的第二个版本将继续如下：
Your second version without .toList(), would proceed as follows:
.findMemberships()
emits a single List<Membership>
.flatMap()
transforms List<Membership> into a single List<User>
Observable.from(users) creates an observable that emits each user
.subscribe()
onNext() is called for each user
onCompleted() is never called.
您的原始版本：
.findMemberships()
emits a single List<Membership>
.flatMap()
transforms List<Membership> into a single List<User>
Observable.from(users) creates an observable that emits each user
.toList()
buffers each User waiting for onCompleted() to be called
onCompleted is never called because the .findMemberships Observable never completes
有几种解决方案：
1）使 findMemberShips（）可观察调用onComplete。如果由 findMemberShips（）是一个Rx主题（PublishSubject，BehaviorSubject等）
1) Make the findMemberShips() Observable call onComplete.This may not be desirable if the Observable returned by findMemberShips() is a Rx Subject (PublishSubject, BehaviorSubject, etc)
2）使用 Observable.just（）而不是 Observable.from（）。你已经在.flatMap（）中有一个 List< User> ，只需返回它。使用 Observable.from（users）创建一个发出每个用户的Observable。 Observable.just（用户）会创建一个Observable，它会发出一个 List< User> 。不需要 .toList（）。
2) Use Observable.just() instead of Observable.from(). You already have a List<User> in .flatMap(), just return it. Using Observable.from(users) creates an Observable that emits each user. Observable.just(users) would create an Observable that emits a single List<User>. No need for .toList().
3）使用 .map（）而不是 .flatMap（）。同样，不需要 .toList（）。由于每个列表< Membership> 都会转换为列表< User> ，您只需要使用 .map（）。
3) Use .map() instead of .flatMap(). Again, no need for .toList(). Since each List<Membership> gets transformed into a List<User> you only need to use .map().
myEvent
.findMemberships()
.map(new Func1<List<Membership>, List<User>>() {
@Override
public List<User> call(List<Membership> memberships) {
List<User> users = new ArrayList<User>();
for (Membership membership : memberships) {
users.add(membership.getUser());
}
return users;
}
})
这篇关于将Observable收集到List中似乎不会立即发出集合的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
