时间戳

首页 > 解决方案 > onNext of the Subscribe method not emitting items after using the ZIP WITH operator in RxJava?

android - onNext of the Subscribe method not emitting items after using the ZIP WITH operator in RxJava?

问题描述

Main POJO:

class VideoResponse{
    List<VideoFiles> videosFiles;
}

I have the following case where i combine the results from two database operation and return as Observable(List(VideoResponse)) .

##Update##

mDbHelper
  ===/* getVideoCategory() returns Observable<List<VideoResponse>> */=========                 
    .getVideoCategory()  
    .flatMapIterable(videoResponses -> videoResponses)
    .doOnNext(videoResponse -> {
              Timber.d("Flatmap Iterable Thread == > %s", Thread.currentThread().getName());})
    .concatMap((Function<VideoResponse, ObservableSource<VideoResponse>>) videoResponse -> {
                Integer videoId = videoResponse.getId();
                return Observable.fromCallable(() -> videoResponse)
                               .doOnNext(videoResponse1 -> {

                            Timber.d("Thread before ZIP WITH ===>  
                                  %s", Thread.currentThread().getName());
                                })

     ===/* getVideoFileList(int videoId) returns Observable<List<VideoFiles>> */====

                              .zipWith(getVideoFilesList(videoId)),
                                        (videoResponse1, videoFiles) -> {
                                            videoResponse1.setFiles(videoFiles);
                                            return videoResponse1;
                                        })
                              .doOnNext(vResponse -> {

                                 Timber.d("Thread After ZIP WITH ===>  
                                  %s",Thread.currentThread().getName());
                                })
           ======= /*This Gets printed*/ ======================
                           .doOnComplete(()->{
                                Timber.d(" OnComplete Thread for Video Files ===>  %s ",Thread.currentThread().getName());
                            });

                     })
                    .toList()
                    .toObservable()

     ===/* Below Print statement is not getting Executed */=================              
                     .doOnComplete(()->{
                    Timber.d(" Thread doOnComplete");
                })
                    .doOnNext(videoResponses -> {
                        Timber.d("Thread after loading from the LOCAL DB ====> %s", Thread.currentThread().getName());
            });

Below are the scheduler threads being executed:

 Flatmap Iterable Thread == >  RxCachedThreadScheduler-1
 Thread before ZIP WITH  ===>  RxCachedThreadScheduler-1
 Flatmap Iterable Thread == >  RxCachedThreadScheduler-1
 Thread After ZIP WITH   ===>  RxCachedThreadScheduler-2
 Thread before ZIP WITH  ===>  RxCachedThreadScheduler-2
 Thread After ZIP WITH   ===>  RxCachedThreadScheduler-2

The final onNext is never getting executed.I need to return the List in the OnNext. I have placed observeOn on different positions ,nothing seems to work..!! Any suggestions..

##Update## Using SqlBrite,

  @Override
public Observable<List<VideoResponse>> getVideoCategory() {
        return mDBHelper
                .createQuery(VideoEntry.TABLE_NAME,
                        DbUtils.getSelectAllQuery(VideoEntry.TABLE_NAME))
                .mapToOne(DbUtils::videosFromCursor);



  @Override
    public Observable<List<VideoFiles>> getVideoFilesList(int videoId) {
        return mDBHelper.createQuery(VideoDetailsEntry.TABLE_NAME,
                     DbUtils.getSelectFromId(VideoDetailsEntry.TABLE_NAME,VideoDetailsEntry.COLUMN_VIDEO_ID),
                String.valueOf(videoId))
                .mapToOne(DbUtils::videoDetailsFromCursor);
    }

标签： androidrx-javarx-java2android-mvpsqlbrite

解决方案

正如@BobDalgleish 所暗示的那样，在toList 之前而不是在ZipWith之后调用OnComplete。现在我进行了以下更改，并且我从 db 获得了完整列表。我使用concatMap来保留订单并等待完成。

mLocalDataSource
  .getVideoCategory()
  .compose(RxUtils.applySchedulers())
  .flatMap(new Function<List<VideoResponse>, 
              ObservableSource<List<VideoResponse>>>() {
                    @Override
                     public ObservableSource<List<VideoResponse>> apply(List<VideoResponse> videoResponses) throws Exception {
                             return Observable.just(videoResponses)
                                    .concatMap(videoResponses1 -> Observable.fromIterable(videoResponses1)
                                    .concatMap(videoResponse -> Observable.just(videoResponse)
                                    .concatMap(videoResponse1 -> {
                                                Integer videoId = videoResponse1.getId();
                                                return Observable.just(videoResponse1)
                                                      .zipWith(getVideoFilesList(videoId), new BiFunction<VideoResponse, List<VideoFiles>, VideoResponse>() {
                                                                @Override
                                                                public VideoResponse apply(VideoResponse videoResponse1, List<VideoFiles> videoFiles) throws Exception {
                                                                        videoResponse1.setFiles(videoFiles);
                                                                        Timber.d("Video Responses == >  %s",videoResponse1);
                                                                        return videoResponse1;
                                                                        }
                                                                    });
                                                     })))
                                      .toList()
                                      .toObservable()
                                     .observeOn(AndroidSchedulers.mainThread());
                            }
                        });

我知道这看起来有点乱！！任何建议或优化，请张贴..！！

推荐阅读