如何从订阅者返回 Observable？

Question

我第一次在我的应用程序中使用 rxJava 我试图实现以下实现：从第 3 方服务器获取帐户从本地数据库获取帐户比较帐户并过滤掉那些不在本地数据库中的帐户只保存那些不在本地数据库中的本地数据库中的帐户。这是我的代码：- 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){

     return accountDAL.getByIds(context, accounts

            .stream()

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

            .collect(Collectors.toList()))

            .map(a -> 

                 a.stream()

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

            ).map(seMap -> 

               accountDAL.save(context, accounts.stream()

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

                     .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) {

        return accountDAL

                .getByIds(context,

                        accounts.stream().map(a -> Long.valueOf(a.getAccountId())).collect(Collectors.toList()))

                .map(ar -> ar.stream().collect(Collectors.toMap(Account::getAccountId, Function.identity())) // map ==>

                                                                                                            // {id =

                // Account}

                ).flatMap(seMap -> accountDAL.save(context, accounts.stream()

                        .filter(a -> seMap.get(Long.valueOf(a.getAccountId())) == null).collect(Collectors.toList())));

    }