Observable not repeating

Question

私はrepeat()に観測を試みていますが、それは複数回実行されていません。

Observable.create(new ObservableOnSubscribe<String>() { 
     @Override 
     public void subscribe(ObservableEmitter<String> e) throws Exception { 
      // **** Code Here is not repeating **** 
      e.onNext(pullMessagesFromServer()); 
     } 
    }) 
    .repeat() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .map(new Function<String, JSONArray>() { 
     @Override 
     public JSONArray apply(String s) throws Exception { 
      return JsonParser.parseString(s); 
     } 
    }) 
    .subscribe(new Observer<JSONArray>() { 
     @Override 
     public void onNext(JSONArray arr) { 
      // **** Code Here is not repeating **** 
     } 
    }); 
} 

何が得られますか？ありがとう！ Rxjavaを初めて使うと、やや混乱します。 :(私のコメントから

Answer 1

：。。ソースが完了した場合、繰り返しが動作e.onComplete()を追加

Observable.create(new ObservableOnSubscribe<String>() { 
     @Override 
     public void subscribe(ObservableEmitter<String> e) throws Exception { 
      // **** Code Here is not repeating **** 
      e.onNext(pullMessagesFromServer()); 
      e.onComplete(); 
     } 
    }) 
    .repeat() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .map(new Function<String, JSONArray>() { 
     @Override 
     public JSONArray apply(String s) throws Exception { 
      return JsonParser.parseString(s); 
     } 
    }) 
    .subscribe(new Observer<JSONArray>() { 
     @Override 
     public void onNext(JSONArray arr) { 
      // **** Code Here is not repeating **** 
     } 
    }); 
} 

この理由はObservableプロトコルはつまり、イベントが時間をかけて到着するかもしれない、潜在的に非同期であるということですので。あなたのコードがシグナリングイベントを止めたからといって、ライブラリーはそれ以上のイベントがないことを知ることはできません。create()を使用しても値が出力されなくなることを明示する必要があります。 fromCallableとして生成されます。これにより、単一の生成値が適切に通知されます。

Observable.fromCallable(new Callable<String>() { 
     @Override 
     public String call() throws Exception { 
      // **** Code Here is not repeating **** 
      return pullMessagesFromServer(); 
     } 
    }) 
    .repeat() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .map(new Function<String, JSONArray>() { 
     @Override 
     public JSONArray apply(String s) throws Exception { 
      return JsonParser.parseString(s); 
     } 
    }) 
    .subscribe(new Observer<JSONArray>() { 
     @Override 
     public void onNext(JSONArray arr) { 
      // **** Code Here is not repeating **** 
     } 
    }); 
}