android - RxJava - 获取列表，对其项目进行操作，再次分组到列表中。 toList() 等待 onCompleted()

Question

我想在每次出现时从我的数据库中获取所有未同步的项目，尝试将它们同步到后端，然后写入数据库，以确保它们已同步。为了执行此操作，我制作了以下操作链(使用 RxJava 可观察对象):

1. 每当数据库表发生变化时，获取未同步项目的列表。
2. 对于列表中的每一项 - 将其发送到后端。
3. 收集所有成功发送到后端的项目...
4. ...并在单个事务中将它们写入数据库。

我试过这样的:

newMsgsObservable //from realm database
            .flatMapIterable(new Func1<List<Message>, Iterable<Message>>() { //1
                @Override
                public Iterable<Message> call(List<Message> messages) {
                    return messages;
                }
            })
            .flatMap(new Func1<Message, Observable<Message>>() { //2
                @Override
                public Observable<Message> call(Message message) {
                    return restService.sendMessage(message);
                }
            })
            .toList() //3
            .flatMap(new Func1<List<Message>, Observable<?>>() { //4
                @Override
                public Observable<?> call(List<Message> messages) {
                    for(Message message : messages) {
                        message.setSynchronised(true);
                    }
                    persistence.saveMessages(messages);
                    return null; //yeah, I know, don't bother ;)
                }
            })
            .subscribe();

问题

newMsgsObservable 一直在等待数据库中的新更改(它永远不会完成)这很好。不幸的是，toList() 在再次返回项目列表之前等待可观察对象完成。我想要一个从数据库中获取单个列表的解决方案，对列表中的每个项目执行所有操作，然后获取相同的列表并将其返回到下一个可观察对象。每次从初始可观察到的新列表可用时都执行此操作。

或者也许我做错了，我应该换一种方式来解决这个问题？

Answer 1

主要思想是将您的toList() 移动到另一个级别。因此，您的代码将如下所示:

newMsgsObservable.flatMap(new Func1<List<Message>, Observable<List<Message>>>() {
    @Override
    public Observable<List<Message>> call(List<Message> messages) {
        return Observable.from(messages).flatMap(new Func1<Message, Observable<Message>>() {
            @Override
            public Observable<Message> call(Message message) {
                return restService.sendMessage(message);
            }
        }).toList();
    }
})
        .doOnNext(new Action1<List<Message>>() {
            @Override
            public void call(List<Message> messages) {
                for (Message message : messages) {
                    message.setSynchronised(true);
                }
                persistence.saveMessages(messages);
            }
        })
        .subscribe();