onNext метода Subscribe не генерирует элементы после использования оператора ZIP WITH в RxJava?

Question

Основной POJO:

class VideoResponse{
    List<VideoFiles> videosFiles;
}

У меня есть следующий случай, когда я объединяю результаты двух операций с базой данных и возвращаю как Наблюдаемый (Список (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());
            }); 

Ниже приведены выполняемые потоки планировщика:

 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

Окончательный onNext никогда не бываетвыполняется. Мне нужно вернуть List в OnNext.Я поместил Наблюдать за на разных позициях, кажется, ничего не работает .. !!Любые предложения ..

## Обновление ## Использование 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);
    }

Answer 1

Как намекнул @BobDalgleish, OnComplete вызывался до toList, а не после ZipWith . Теперь я внес следующие изменения и получаю полный список из 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());
                            }
                        }); 

Я знаю, это выглядит немного грязно !!. Любые предложения или оптимизации, пожалуйста, напишите .. !!