BehaviorSubject innerhalb eines BehaviorSubject

Question

Szenarios Umgang habe ich eine Liste von Territorien und ich werde alle Artikel in jedem holen mag und jede Territory vom Server. Also entschied ich mich, BehaviorSubject zu verwenden, um es zu versuchen. Ich weiß nicht, ob ich "verrückt" bin oder nicht. Das ist, was ich tat:

private Observable<ItemListResponse> getItemByTerritory() { 

    List<String> territories = PrefUtils.getUserTerritories(context); 

    BehaviorSubject<Integer> territoryControl = BehaviorSubject.<Integer>create(0); 
    Observable<ItemListResponse> territoryRet = territoryControl.asObservable().concatMap(integer -> { 

     if (integer != territories.size()) { 

      BehaviorSubject<Integer> ret = BehaviorSubject.create(1); 
      return ret.asObservable().concatMap(integer1 -> { 

       if (integer > 0) { 
        return bakeryAPI.getBakeries("Bearer " + PrefUtils.getToken(context), integer1, 21, territories.get(integer)) 
          .subscribeOn(Schedulers.io()) 
          .observeOn(AndroidSchedulers.mainThread()) 
          .doOnNext(itemListRespont -> { 
           if (itemListRespont.getItemResponse().getMeta().getPage() != itemListRespont.getItemResponse().getMeta().getPageCount() && itemListRespont.getItemResponse().getMeta().getPageCount() > 0) { 
            ret.onNext(integer1 + 1); 
           } else { 
            ret.onNext(-1); 
            territoryControl.onNext(integer + 1); 
           } 
          }); 
       } else { 
        return Observable.<ItemListResponse>empty().doOnCompleted(ret::onCompleted); 
       } 
      }); 
     } else { 
      return Observable.<ItemListResponse>empty().doOnCompleted(territoryControl::onCompleted); 
     } 
    }); 
    return Observable.defer(() -> territoryRet); 
} 

Alles, was es tut, ist einfach zu hängen, wie nichts ist überhaupt läuft, wenn ich dies tun

public void getItemByTerritory(APIRequestListener apiRequestListener){ 
    realm.executeTransaction(realm1 -> realm1.where(RealmItem.class).findAll().deleteAllFromRealm()); 

    getItemByTerritory().subscribe(new Subscriber<ItemListResponse>() { 
     @Override 
     public void onCompleted() { 
      Log.e(TAG, "Completed Bakery"); 
      apiRequestListener.completed(); 
      unsubscribe(); 
     } 

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

     @Override 
     public void onNext(ItemListResponse itemListRespont) { 
     } 
    }); 
} 

Answer 1

Wenn ich das richtig verstanden, und was sind Sie nach ist das Abrufen Seiten von getBakeries Antworten dann sollten die folgenden rekursiven Funktionsaufrufe der Trick:

private Observable<ItemListResponse> getItemByTerritory() { 
    List<String> territories = PrefUtils.getUserTerritories(context); 
    return Observable.from(territories).flatMap(territory -> fetchTerritoryPage(territory, 0)); 
} 

private Observable<ItemListResponse> fetchTerritoryPage(String territory, int page) { 
    System.out.println("Fetch territory page " + page); 
    return bakeryAPI.getBakeries("Bearer " + PrefUtils.getToken(context), page, 21, territory).flatMap(response 
      -> { 
     final ItemListResponse.ItemResponse.Meta meta = response.getItemResponse().getMeta(); 
     final Observable<ItemListResponse> thisPage = Observable.just(response); 
     if (page < meta.getPageCount()) { 
      final Observable<ItemListResponse> nextPage = fetchTerritoryPage(territory, page + 1); 
      return thisPage.concatWith(nextPage); 
     } else { 
      return thisPage; 
     } 
    }); 
} 

Beachten Sie, wie fetchTerritoryPage ruft sich selbst rekursiv nächste Seite zu holen.