비동기 작업/rx Java 코드를 rxjava2로 변환하려면 어떻게해야합니까?

Question

Rxjava2에 내 AsyncTask 코드를 변환하려고했지만 분명히 Rxjava2은 null 값을 처리하지 않으므로 내 응용 프로그램이 충돌합니다.

new AsyncTask<Void, Void, Void>() { 
      @Override 
      protected Void doInBackground(Void... params) { 
       Set<Map.Entry<String, Participant>> entries = pool.entrySet(); 
       for (Map.Entry<String, Participant> entry : entries) { 
        Participant participant = entry.getValue(); 
        participant.release(); 
       } 
       return null; 
      } 

      @Override 
      protected void onPostExecute(Void aVoid) { 
       cb.event(new Spin.Event<Void>()); 
      } 
     }.execute(); 

을 그리고 여기 Rxjava (NOT Rxjava2)로 변환 된 코드입니다 : 여기 AsyncTask 내 코드의

Observable.defer(new Func0<Observable<Void>>() { 
     @Override 
     public Observable<Void> call() { 
      Set<Map.Entry<String, Participant>> entries = pool.entrySet(); 
      for (Map.Entry<String, Participant> entry : entries) { 
       Participant participant = entry.getValue(); 
       participant.release(); 
      } 
      return Observable.just(null); 
     } 
    }).doOnCompleted(new Action0() { 
     @Override 
     public void call() { 
      cb.event(new Spin.Event<Void>()); 
     } 
    }) 
    .subscribeOn(Schedulers.computation()) 
    .subscribe(); 

가 null의 반환에 충돌하지 않고 Rxjava로 변환하는 가장 좋은 방법 일 것입니다 무엇. 또한 .execute()는 Rxjava2과 관련하여 어떻게 재생됩니까? 그것이 심지어 Rxjava에서 작동하는지 확실하지 않은가요?

FATAL EXCEPTION: RxComputationThreadPool-3 

                      io.reactivex.exceptions.OnErrorNotImplementedException: null ObservableSource supplied 
                       at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704) 
                       at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701) 
                       at io.reactivex.internal.observers.LambdaObserver.onError(LambdaObserver.java:74) 
                       at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeOnObserver.onError(ObservableSubscribeOn.java:63) 
                       at io.reactivex.internal.disposables.EmptyDisposable.error(EmptyDisposable.java:63) 
                       at io.reactivex.internal.operators.observable.ObservableDefer.subscribeActual(ObservableDefer.java:35) 
                       at io.reactivex.Observable.subscribe(Observable.java:10842) 
                       at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96) 
                       at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38) 
                       at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26) 
                       at java.util.concurrent.FutureTask.run(FutureTask.java:237) 
                       at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:154) 
                       at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:269) 
                       at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1113) 
                       at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:588) 
                       at java.lang.Thread.run(Thread.java:818) 
                      Caused by: java.lang.NullPointerException: null ObservableSource supplied 
                       at io.reactivex.internal.functions.ObjectHelper.requireNonNull(ObjectHelper.java:39) 
                       at io.reactivex.internal.operators.observable.ObservableDefer.subscribeActual(ObservableDefer.java:32) 
                       at io.reactivex.Observable.subscribe(Observable.java:10842)  
                       at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)  
                       at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:38)  
                       at io.reactivex.internal.schedulers.ScheduledDirectTask.call(ScheduledDirectTask.java:26)  
                       at java.util.concurrent.FutureTask.run(FutureTask.java:237)  
                       at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:154)  
                       at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:269)  
                       at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1113)  
                       at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:588)  
                       at java.lang.Thread.run(Thread.java:818)  

Answer 1

Observable.defer(new Callable<ObservableSource<?>>() { 
      @Override 
      public ObservableSource<?> call() throws Exception { 
       Set<Map.Entry<String, Participant>> entries = pool.entrySet(); 
       for (Map.Entry<String, Participant> entry : entries) { 
        Participant participant = entry.getValue(); 
        participant.release(); 
       } 
       return Completable.complete().toObservable(); 
      } 
     }).doOnComplete(new Action() { 
      @Override 
      public void run() throws Exception { 
       Log.d("Complete", "Complete"); 
      } 
     }) 
      .subscribeOn(Schedulers.computation()) 
      .observeOn(AndroidSchedulers.mainThread()).subscribe(); 

이 코드도 작동합니다

다음은 충돌 로그입니다. subscribe() 메서드를 호출하면 작업이 시작됩니다. 다시 메인 스레드에 게시 할 값을 가지고 있지 않기 때문에

Answer 2

, 당신은 Completable를 사용할 수 있습니다

Completable.fromAction(() -> { 
    Set<Map.Entry<String, Participant>> entries = pool.entrySet(); 
    for (Map.Entry<String, Participant> entry : entries) { 
     Participant participant = entry.getValue(); 
     participant.release(); 
    } 
}) 
.subscribeOn(Schedulers.computation()) 
.observeOn(AndroidSchedulers.mainThread()) 
.subscribe(
    () -> { 
     cb.event(new Spin.Event<Void>()); 
    }, 
    error -> { /* show error toast */ } 
); 

Answer 3

Observable.defer(new Callable<ObservableSource<?>>() { 

//This method is replacing doInBackground 
     @Override 
     public ObservableSource<?> call() throws Exception { 
      Set<Map.Entry<String, Participant>> entries = pool.entrySet(); 
      for (Map.Entry<String, Participant> entry : entries) { 
       Participant participant = entry.getValue(); 
       participant.release(); 
      } 
      return Completable.complete().toObservable(); 
     } 
    }).doOnComplete(new Action() { 
    //This is onPostExecute 
     @Override 
     public void run() throws Exception { 
      Log.d("Complete", "Complete"); 
     } 
    }) 
     .subscribeOn(Schedulers.computation()) 
     .observeOn(AndroidSchedulers.mainThread()).subscribe()