RxAndroid observable looping

Question

私は他の人たちによって開発された古いプロジェクトに取り組んでおり、RxAndroidがプロジェクトで使用されています。
私はアプリケーションの要求のいくつかにキャッシュ処理を追加しなければならず、必要なときにOkHttpキャッシュはうまく機能しませんでした。だから私は、インターネット接続をチェックしてから要求を出したり、接続に応じてキャッシュから読み込んだりするためのソリューションを思いついた。私はRxAndroidと経験を持っていないが、私は少しを読んで、私はこの思い付いた：

@Override 
public Observable<List<SocialNews>> getStream(@Path("stream") String stream, @Query("networks") String networks, @Query("page") Integer page, @Query("limit") Integer limit) { 
    return Observable.create(new AsyncOnSubscribe<Object, List<SocialNews>>() { 
     @Override 
     protected Object generateState() { 
      return null; 
     } 

     @Override 
     protected Object next(Object state, long requested, Observer<Observable<? extends List<SocialNews>>> observer) { 
      if (NetworkUtils.hasActiveInternetConnection(context)) { 
       observer.onNext(APIClient.getService(context).getStream(stream, networks, page, limit)); 
      } else { 
       observer.onNext(Observable.just(NetworkUtils.getCachedStream())); 
      } 
      return state; 
     } 
    }); 
} 

そして、私はこのようにそれに加入する：

final Observable<List<SocialNews>> observable = APIClient.getCacheService(getActivity()).getStream(stream, networks, page, SportFiveAPI.DEFAULT_ITEM_LIMIT); 
    LifecycleObservable.bindFragmentLifecycle(lifecycle(), 
      AppObservable.bindFragment(this, observable) 
        .subscribeOn(Schedulers.io()) 
        .observeOn(AndroidSchedulers.mainThread())) 
      .subscribeOn(Schedulers.io()) 
      .subscribe(
        socialNews -> { 
         // do stuff 
        }, 
        error -> { 
         // do stuff 
        } 
      ); 

しかし、どういうわけか、それがループ内で立ち往生私のAsyncOnSubscribe内の次のメソッドを呼び出し続けます。誰でも私がここで間違っていることを知っていますか？

Answer 1

あなたはあなたの観察者にonCompletedを呼び出していません。すなわちonCompletedを呼び出していない -

Answer 2

まさにあなたのコードを持っているObservable契約の違反の種類を避けるために、Observable.fromCallableそのObservable.createを使用してはるかに優れています。 concatとfirstを使用してフォールバック型データ検索のこの種を行うための本当にすてきなパターンがあり、また、

@Override 
public Observable<List<SocialNews>> getStream(@Path("stream") String stream, @Query("networks") String networks, @Query("page") Integer page, @Query("limit") Integer limit) { 
Observable.fromCallable(() -> NetworkUtils.hasActiveInternetConnection(context)) 
      .flatMap(hasConnection -> { 
      if (hasConnection) { 
       return APIClient.getService(context).getStream(stream, networks, page, limit); 
      } else { 
       return Observable.just(NetworkUtils.getCachedStream()) 
      } 
      }) 
} 

：私はこのようなものをお勧めします。詳細はDan Lewのblog postを参照してください。