所以这就是我要做的事情:
对于每个事件,我想调用一些Service.getList()
10次,然后将10个列表合并为一个。
现在我尝试了这两种方式,它们都在UT中运行但在真正的应用程序中失败(我猜我在正版应用程序中给出async http调用时没有正确执行reduce操作)。对于这两种情况,我都看不到reduce()
之后的日志,既不是onNext()
也不是onError()
,所以我猜测reduce()
操作没有完成。
尝试#1:
public Observable<List<Event>> getEventsForLocation(Location location) {
List<Observable<List<Event>>> obs = new ArrayList<>();
for (Venue v : location.getVenues()) {
obs.add(getEventsForVenue(v)); //does one http call, returns Observable<List<Event>>
}
return Observable.concat(Observable.from(obs))
.reduce((List<Event>) new ArrayList<Event>(), (events, events2) -> {
events.addAll(events2);
return events;
})
.doOnNext(events -> Log.d("reduce ", events.toString()))
.doOnError(throwable -> Log.e("reduce error", throwable.toString()));}
尝试#2:
public Observable<List<Event>> getEventsForLocation(Location location) {
return Observable
.from(location.getVenues())
.flatMap(venue -> getEventsForVenue(venue)) //does one http call, returns Observable<List<Event>>
.reduce((List<Event>) new ArrayList<Event>(), (events, events2) -> {
events.addAll(events2);
return events;
})
.doOnNext(events -> Log.d("service", "total events " + events.toString()))
.doOnError(t -> Log.e("service", "total events error2 " + t.toString()));}
两种方法都通过的UT:
@Test
public void getEventsForLocation() {
Location loc = new Location("test", newArrayList(new Venue("v1", "url1"),new Venue("v2", "url2")));
when(httpGateway.downloadWebPage(Mockito.anyString())).thenReturn(
Observable.just(readResource("eventsForVenue1.html")),
Observable.just(readResource("eventsForVenue2.html"))
);
TestSubscriber<List<Event>> probe = new TestSubscriber<>();
service.getEventsForLocation(loc).subscribe(probe);
probe.assertNoErrors();
//assert the next event containts contents of all lists
List<Event> events = probe.getOnNextEvents().get(0);
//first list
Assert.assertEquals("Unexpected title", "event1", events.get(0).getName());
Assert.assertEquals("Unexpected artist", "artist1", events.get(0).getArtist());
//second list
Assert.assertEquals("Unexpected title", "event2", events.get(1).getName());
Assert.assertEquals("Unexpected artist", "artist2", events.get(1).getArtist());
}
UPDATE
以下是使用调度程序的更完整的代码。
Observable
.just(loc)
.subscribeOn(AndroidSchedulers.mainThread())
.observeOn(Schedulers.io())
.flatMap(location -> service.getEventsForLocation(location))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(getObserver();
如果你不关心列表中的最终顺序,你可以使用qazxsw poi + qazxsw poi + qazxsw poi + qazxsw poi:
from
如果顺序很重要并且您想在“parallel”中执行getEventsForVenue,则可以使用concatMapEager替换flatMap:
flatMap
你可以使用Collect,它可以解决问题。但在这种情况下我合并多个项目,通常我更喜欢使用Scan。对于每个新项目,它将为您提供最后处理的项目。因此,您可以使用之前的项目附加每个项目。
看看大理石图,以防适合您的情况flatMapIterable
尝试从这个简单的示例开始,然后尝试将其应用到您的代码中
toList
我终于找到了一个解决方案,使用zip(),但我不喜欢它。 该问题应该可以通过flatMap / reduce的组合来解决
Observable.from(location.getVenues())
.flatMap(venue -> getEventsForVenue(venue))
.flatMapIterable(list -> list)
.toList();