Как использовать один наблюдаемый rxjava после предыдущего?

Question

У меня есть два наблюдателя RxJava, и я получаю arraylist от первого наблюдаемого, а затем использую его для получения данных из другого наблюдаемого, подобного этому.

Observable<KarobarTvVod> observable1 = youtubeDataHelper.getTVData(); 
    observable1.subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .unsubscribeOn(Schedulers.io()) 
      .subscribe(new Subscriber<KarobarTvVod>() { 
       @Override 
       public void onCompleted() { 

       } 

       @Override 
       public void onError(Throwable e) { 
        e.printStackTrace(); 
       } 

       @Override 
       public void onNext(KarobarTvVod karobarTvVod) { 
        Log.d(TAG, "onNext: size" + karobarTvVod.getEtag()); 
        tvObjArrayList = new ArrayList<TVObj>(); 
        for (int i = 0; i < karobarTvVod.getItems().size(); i++) { 
         TVObj tvObj = new TVObj(); 
         tvObj.setVideoDate(karobarTvVod.getItems().get(i).getSnippet().getPublishedAt()); 
         tvObj.setVideoIcon(karobarTvVod.getItems().get(i).getSnippet().getThumbnails().getHigh().getUrl()); 
         tvObj.setVideoTitle(karobarTvVod.getItems().get(i).getSnippet().getTitle()); 
         tvObj.setVideoID(karobarTvVod.getItems().get(i).getId().getVideoId()); 
         tvObjArrayList.add(tvObj); 
        } 


       } 
      }); 



    Observable<YoutubeViews> observable2 = youtubeDataHelper.getTVDataViews(tvObjArrayList); 
    observable2.subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .unsubscribeOn(Schedulers.io()) 
      .subscribe(new Subscriber<YoutubeViews>() { 
       @Override 
       public void onCompleted() { 

       } 

       @Override 
       public void onError(Throwable e) { 
        Log.d(TAG, "onError: in 2nd obs"); 
        e.printStackTrace(); 
       } 

       @Override 
       public void onNext(YoutubeViews youtubeViews) { 
        Log.d(TAG, "onNext: views" + youtubeViews.getEtag()); 
        viewsList = new ArrayList<String>(); 
        for (int i = 0; i < youtubeViews.getItems().size(); i++) { 

         viewsList.add(youtubeViews.getItems().get(i).getStatistics().getViewCount()); 
        } 
        tvView.displayList(tvObjArrayList, viewsList); 
       } 
      }); 

Это просто пример кода, мне нужно пройти tvObjArrayList, когда он получает сведения из первых Наблюдаемое на второй Observable, что лучшая практика, чтобы сделать это? А также я использую for-loop внутри 1-го наблюдаемого, есть ли лучший способ его достижения с помощью rxjava? Благодаря

Answer 1

Вы должны использовать оператор flatMap. Это не будет намного легче, чем это.

Observable<KarobarTvVod> observable1 = youtubeDataHelper.getTVData(); 
observable1.subscribeOn(Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .unsubscribeOn(Schedulers.io()) 
     .flatMap(new Func1<KarobarTvVod, Observable<YoutubeViews>>() { 
      @Override 
      public Observable<YoutubeViews> call(KarobarTvVod karobarTvVod) { 
       Log.d(TAG, "onNext: size" + karobarTvVod.getEtag()); 
       tvObjArrayList = new ArrayList<TVObj>(); 
       for (int i = 0; i < karobarTvVod.getItems().size(); i++) { 
        TVObj tvObj = new TVObj(); 
        tvObj.setVideoDate(karobarTvVod.getItems().get(i).getSnippet().getPublishedAt()); 
        tvObj.setVideoIcon(karobarTvVod.getItems().get(i).getSnippet().getThumbnails().getHigh().getUrl()); 
        tvObj.setVideoTitle(karobarTvVod.getItems().get(i).getSnippet().getTitle()); 
        tvObj.setVideoID(karobarTvVod.getItems().get(i).getId().getVideoId()); 
        tvObjArrayList.add(tvObj); 
       } 
       return youtubeDataHelper.getTVDataViews(tvObjArrayList); 
      } 
     }).subscribe(new Subscriber<YoutubeViews>() { 
      @Override 
      public void onCompleted() { 

      } 

      @Override 
      public void onError(Throwable e) { 
       Log.d(TAG, "onError: in 1st or 2nd obs"); 
       e.printStackTrace(); 
      } 

      @Override 
      public void onNext(YoutubeViews youtubeViews) { 
       Log.d(TAG, "onNext: views" + youtubeViews.getEtag()); 
       viewsList = new ArrayList<String>(); 
       for (int i = 0; i < youtubeViews.getItems().size(); i++) { 

        viewsList.add(youtubeViews.getItems().get(i).getStatistics().getViewCount()); 
       } 
       tvView.displayList(tvObjArrayList, viewsList); 
      } 
     }); 

Answer 2

Вы должны подписаться второй наблюдаемым в OnComplete первого одного

Observable<KarobarTvVod> observable1 = youtubeDataHelper.getTVData(); 
    Observable<YoutubeViews> observable2 = youtubeDataHelper.getTVDataViews(tvObjArrayList); 

    observable1.subscribeOn(Schedulers.io()) 
     .observeOn(AndroidSchedulers.mainThread()) 
     .unsubscribeOn(Schedulers.io()) 
     .subscribe(new Subscriber<KarobarTvVod>() { 
      @Override 
      public void onCompleted() { 
       observable2.subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .unsubscribeOn(Schedulers.io()) 
      .subscribe(new Subscriber<YoutubeViews>() { 
        @Override 
        public void onCompleted() { 

        } 

        @Override 
        public void onError(Throwable e) { 
         Log.d(TAG, "onError: in 2nd obs"); 
         e.printStackTrace(); 
        } 

        @Override 
        public void onNext(YoutubeViews youtubeViews) { 
         Log.d(TAG, "onNext: views" + youtubeViews.getEtag()); 
         viewsList = new ArrayList<String>(); 
         for (int i = 0; i < youtubeViews.getItems().size(); i++) { 

          viewsList.add(youtubeViews.getItems().get(i).getStatistics().getViewCount  ()); 
         } 
         tvView.displayList(tvObjArrayList, viewsList); 
        } 
       }); 
      } 

      @Override 
      public void onError(Throwable e) { 
       e.printStackTrace(); 
      } 

      @Override 
      public void onNext(KarobarTvVod karobarTvVod) { 
       Log.d(TAG, "onNext: size" + karobarTvVod.getEtag()); 
       tvObjArrayList = new ArrayList<TVObj>(); 
       for (int i = 0; i < karobarTvVod.getItems().size(); i++) { 
        TVObj tvObj = new TVObj(); 
        tvObj.setVideoDate(karobarTvVod.getItems().get(i).getSnippet().getPublishedAt()); 
        tvObj.setVideoIcon(karobarTvVod.getItems().get(i).getSnippet().getThumbnails().getHigh().getUrl()); 
        tvObj.setVideoTitle(karobarTvVod.getItems().get(i).getSnippet().getTitle()); 
        tvObj.setVideoID(karobarTvVod.getItems().get(i).getId().getVideoId()); 
        tvObjArrayList.add(tvObj); 
       } 


      } 
     }); 

Конечно, чтобы сделать этот код более читаемым Я хотел бы использовать функцию потребителей для observable2.subsriberOn по методу OnComplete

Answer 3

Вы можете использовать оператор

toList().flatMap() 

для наблюдаемых А, а в функции flatMap, работает для наблюдаемого В.

Например:

observableA 
    .toList() 
    .flatMap(observableB.subscribe()) 
    .subscribe() 

Answer 4

Я полагаю, что getTVData и getTVDataViews каждый испускает один элемент, а не звонки onComplete. Если это правда, то работает следующий пример. Нет loop «s, просто чисто rx :)

//getTVData should emit one item and then call obComplete 
//otherwise toList() will wait forever 
service.getTVData() 
     .flatMap(karobarTvVod -> Observable.from(karobarTvVod.getItems())) 
     .map(item -> { 
      TVObj tvObj = new TVObj(); 
      //set other fields 
      //by the way, I recommend you to use immutable objects 
      return tvObj; 
     }) 
     .toList() 
     //here we have List<TVObj> 
     .flatMap(
       objs -> { 
        //getTVDataViews should emit one item and then call onComplete 
        //otherwise toList will wait forever 
        return service.getTVDataViews(objs) 
          .flatMap(youtubeViews -> Observable.from(youtubeViews.getItems())) 
          .map(integer -> integer.toString()) 
          //after that we will have List<String> 
          .toList(); 
       }, 
       //a function that combines one item emitted by each of the source and collection Observables 
       // and returns an item to be emitted by the resulting Observable 
       new Func2<List<TVObj>,List<String>,Pair<List<TVObj>,List<String>>>() { 
        @Override 
        public Pair<List<TVObj>, List<String>> call(List<TVObj> objs, List<String> strings) { 
         return new Pair(objs, strings); 
        } 
       }) 
     .subscribe(pair -> tvView.displayList(pair.first, pair.second)); 

PS. Хотя этот подход более краткий, я считаю, что loop для создания list элементов более эффективен.