android - RxJava 和 Retrofit 处理多个平面图的分页

Question

我正在使用 Retrofit 和 RxJava，我需要执行以下登录场景:

1) 使用密码登录用户并获取访问 token

2) 获取用户资料，friends(需要分页假设15页)，conversations(需要分页假设10页)

第二步的所有请求都需要 token ，因此它们取决于第一步。

我需要做的是执行第一步，完成后执行第二步，并在所有 3 个作业完成时收到通知。

我现在拥有的:

首先，我使用 concatWith 运算符处理分页，并且效果很好。例如:

mApi.getConversationsByPage(accessToken,page,take)
.concatMap(convResponse -> {
  if(convResponse.body().getNextPageUrl().equals("NA")) {
      return Observable.just(convResponse);
  } else {
      return Observable.just(convResponse)
      .concatWith(getConversationsAndNext(accessToken,convResponse .body().getNextPage(),take));
  }
});

我的代码:

mService.loginUser(email,password)
     .subscribeOn(mSchedulerProvider.io())
     .observeOn(mSchedulerProvider.ui())
     .flatMap(tokenResponse -> {
            //login finished save token
            mService.saveUserToken(tokenResponse.body());
            //first step done go to second step
            return mService.getUserProfile(mService.getToken();
       })
     .flatMap(profileResponse -> {
         mService.saveUserProfile(profileResponse.body());
         return mService.getUserFriendsAndNext(mService.getToken(),FIRST_PAGE,TAKE_PERPAGE);
     }).doOnNext(friendResponse -> mService.handleFriendPaging(friendResponse.body().getData()))
     .flatMap(friendResponse -> {
         mService.saveAllFriends();
         return mService.getConversationsAndNext(mService.getToken(),FIRST_PAGE,TAKE_PERPAGE);
     })
     .subscribe(new Observer<Response<ConvResult>>() {
         @Override
         public void onSubscribe(Disposable d) {
             mDisposables.add(d);
         }

         @Override
         public void onNext(Response<ConvResult> convResponse) {
             mService.handleConvPaging(convResponse .body().getData());   
         }

         @Override
         public void onError(Throwable e) {
             //handle error
         }

         @Override
         public void onComplete() {
             //all requests done handle it
             mService.saveAllFriends();
         }
     });

一切正常。我正在保存用户 token 、用户个人资料、用户 friend ，但是当它进入对话时，mService.handleConvPaging 方法调用了 25 次而不是 10 次。

我的测试方法:Mockito.verify(mService,times(10)).handleConvPaging();

错误:

org.mockito.exceptions.verification.TooManyActualInvocations: 
mService.handleConvPaging();
Wanted 10 times:
-> at ........
But was 25 times. 

所以我想做的是:独立获取用户好友和对话，并在所有请求、分页完成时得到通知。

有没有什么方法可以不使用嵌套回调来实现这一点？提前致谢。

@编辑:

mService.loginUser(email,password)
            .flatMap(tokenResponse -> {
                if(tokenResponse.isSuccessful()) {
                    mService.saveUserToken(tokenResponse.body());
                    return mService.getUserProfile(mService.getUserAccessToken());
                } else {
                    return Observable.error(new Exception("login token error"));
                }
            })
            .flatMap(profileResponse -> {
                if(profileResponse.isSuccessful()) {
                    mService.saveUserProfile(profileResponse.body());
                    return mService.getUserFriendsAndNext(mService.getUserAccessToken(),1,Constants.TAKE_COUNT);
                }else {
                    return Observable.error(new Exception("login profile error"));
                }
            })
            .doOnNext(friendResponse -> {
                if(friendResponse.isSuccessful()) mService.handleFriendPaging(friendResponse.body().getData());
            })
            .ignoreElements()
            .andThen(mService.getConversationsAndNext(mService.getUserAccessToken(),1,Constants.TAKE_COUNT))
            .subscribe(onsubscribe,onnext,onerror,oncomplete) 

如果我在 tokenResponse 部分返回 Observable.error()，andThen 方法正在调用，我不希望这种情况发生

而且当 mService.loginUser 方法调用时我也返回这个:

Response<UserTokenResult> tokenResultResponse = Response.error(400,mResponseBody);

Answer 1

正如您在评论中提到的，您的问题是您的第三个 flatMap 应该在第二个 flatMap 完成后调用。所以我想提出这样的解决方案:

    mService.loginUser(email,password)
            .subscribeOn(mSchedulerProvider.io())
            .observeOn(mSchedulerProvider.ui())
            .flatMap(tokenResponse -> {
                mService.saveUserToken(tokenResponse.body());
                return mService.getUserProfile(mService.getToken();
            })
            .flatMap(profileResponse -> {
                mService.saveUserProfile(profileResponse.body());
                return mService.getUserFriendsAndNext(mService.getToken(),FIRST_PAGE,TAKE_PERPAGE);
            })

            // do All your work with your Friends in this operator
            .doOnNext(friendResponse -> mService.handleFriendPaging(friendResponse.body().getData()))

            //his will return you a Completable
            .ignoreElements()

            //You get the signal means the upStream works are done
            .andThen(mService.getConversationsAndNext(mService.getToken(),FIRST_PAGE,TAKE_PERPAGE))
            .subscribe(new Observer<Response<ConvResult>>() {
                @Override
                public void onSubscribe(Disposable d) {
                    mDisposables.add(d);
                }

                @Override
                public void onNext(Response<ConvResult> convResponse) {
                    mService.handleConvPaging(convResponse .body().getData());
                }

                @Override
                public void onError(Throwable e) {
                    //handle error
                }

                @Override
                public void onComplete() {
                    //This is somehow not necessary
                    mService.saveAllFriends();
                }
            });

因此，您只保留一个流并完成您的工作。

ignoreElements() 提供你只接收终端事件(onComplete() 和 onError())的能力，andThen() 将订阅 upStream Completable 并将继续从您放入 andThen() 的源发出项目。

我测试了类似的东西

我试过类似的东西

    Observable.just("A", "B", "C")
            .flatMap(x -> Observable.error(new Throwable("Eorror")))
            .doOnNext(x -> System.out.println(x))
            .ignoreElements()
            .andThen(Observable.just("New D"))
            .subscribe(x -> System.out.println("onNext" + x),
            error -> System.out.println("onError" + error),
            () -> System.out.println("END"));

它只是打印

System.out: onErrorjava.lang.Throwable: Eorror

使用 RxJava 版本 2.0.9