如何从订阅者返回 Observable？

Question

我第一次在我的应用程序中使用 rxJava 我试图实现以下实现：

1. 从第 3 方服务器获取帐户

2. 从本地数据库获取帐户

3. 比较帐户并过滤掉那些不在本地数据库中的帐户

4. 只保存那些不在本地数据库中的本地数据库中的帐户。

这是我的代码：-

 private Observable<List<Result<Account, IError>>> filterAccounts(Context context, List<Account> accounts){

         accountDAL.getByIds(context, accounts

                .stream()

                .map(a -> Long.valueOf(a.getAccountId()))

                .collect(Collectors.toList()))//return Observable<List<T>> getByIds(Context context, List<Long> ids)

                .map( a -> {

                    Map<Long, SearchConnectAccount> map = a.stream()

                            .collect(Collectors.toMap(a -> a.getId(), Function.identity())); // map ==> {id = Account}

                 return map;

                }).subscribe( seMap -> { // subscribe observable

                  List<Account> filteredList = accounts.stream()

                             .filter(a -> seMap.get(Long.valueOf(a.getAccountId())) == null)

                             .collect(Collectors.toList());

Observable<List<Result<Account, IError>>> o = accountDAL.save(context, filteredList).first();

                    return o;//accountDAL.save(context, filteredList).first();

         });

        // how to return Observable<List<Result<Account, IError>>> from here

    }

任何帮助表示赞赏。

Answer 1

你可以这样做，private Observable<List<Result<Account, IError>>> filterAccounts(Context context, List<Account> accounts){&nbsp; &nbsp; &nbsp;return accountDAL.getByIds(context, accounts&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .stream()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .map(a -> Long.valueOf(a.getAccountId()))&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .collect(Collectors.toList()))&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .map(a ->&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;a.stream()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .collect(Collectors.toMap(a -> a.getId(), Function.identity())) // map ==> {id = Account}&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ).map(seMap ->&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;accountDAL.save(context, accounts.stream()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;.filter(a -> seMap.get(Long.valueOf(a.getAccountId())) == null)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;.collect(Collectors.toList())).first());}更新第二次调用save返回一个Observable<?>（只是一个假设），当它被包装在一个map运算符中时，它返回Observable<Observable<?>>。但是你需要的返回值是Observable<?>. 所以，你需要拼合Observable<Observable<?>>到Observable<?>哪里，这就是flatMap被使用。如果需要，这里是更新的答案。private Observable<List<Result<Account, IError>>> filterAccounts(Context context, List<Account> accounts) {&nbsp; &nbsp; &nbsp; &nbsp; return accountDAL&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .getByIds(context,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; accounts.stream().map(a -> Long.valueOf(a.getAccountId())).collect(Collectors.toList()))&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .map(ar -> ar.stream().collect(Collectors.toMap(Account::getAccountId, Function.identity())) // map ==>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // {id =&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // Account}&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ).flatMap(seMap -> accountDAL.save(context, accounts.stream()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .filter(a -> seMap.get(Long.valueOf(a.getAccountId())) == null).collect(Collectors.toList())));&nbsp; &nbsp; }