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()을 사용할 때 더 이상 값이 출력되지 않는다는 것을 명시해야합니다. 일반적으로 다른 팩토리 메소드를 사용할 수 있습니다 으로 설정하면 하나의 생성 된 값이 올바르게 표시됩니다.

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 **** 
     } 
    }); 
}