CODEWITHSUNDEEP
javajava-8streamrx-javaobservable
Sandeep Tiwari
Sandeep Tiwari

Reputation: 2072

How to return Observable from subscriber?

I am using rxJava first time in my application I am trying to achieve following implementation :

  1. fetch account from 3rd party server
  2. fetch account from local DB
  3. compare account and filter out those account which is not in local DB
  4. save only those accounts in local db which is not in local DB.

Here is my code :- 

 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
    }

Any help is appreciated.

Upvotes: 2

Views: 250

Answers (1)

Ravindra Ranwala
Ravindra Ranwala

Reputation: 21124

You may do it like so,

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());
}

Update

The 2nd call to save returns an Observable<?> (just an assumption) and when it is wrapped in a map operator it returns Observable<Observable<?>>. But what you need as a return value is Observable<?>. So you need to flatten the Observable<Observable<?>> into Observable<?> and that's where flatMap is used. Here's the updated answer in case if needed.

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())));
    }

Upvotes: 2

Related Questions