如何在rxJava中手動調用observer.onNext

Question

我對RxJava/RxAndroid比較新。在此之前，我一直在使用AsyncTask來完成長時間運行的任務。 我已將大部分AsyncTask轉換爲RxJava，但是這一個。 我遇到的特殊問題是在後臺線程中調用類似於AsyncTask的publishProgress(params);的 。我需要這樣做來更新ProgressBar的進度。

首先這是的AsyncTask

private static class AddBooksToDatabase extends AsyncTask<String, String, String> { 
    //dependencies removed 

    AddBooksToDatabase(AddBooksDbParams params) { 
     //Removed assignment codes 
    } 

    @Override 
    protected String doInBackground(String... strings) { 
     //Initializing custom SQLiteOpenHelper and SQLite database 
     File mFile = new File(mFolderPath); 

     int booksSize = getFilesInFolder(mFile).size(); 
     String[] sizeList = {String.valueOf(booksSize)}; 
     //The first publishProgress is used to set the max of the progressbar 
     publishProgress(sizeList); 

     for (int i = 0; i < booksSize; i++) { 
      //publishProgress with current item, current file 
      publishProgress(String.valueOf(i), getFilesInFolder(mFile).get(i).getName()); 
      //Inserting current items in database. Code removed 
     } 
     return null; 
    } 

    @Override 
    protected void onPreExecute() { 
     //Show ProgressBar 
    } 

    @Override 
    protected void onPostExecute(String s) { 
     //Hide ProgressBar 
    } 

    @Override 
    protected void onProgressUpdate(String... values) { 
     super.onProgressUpdate(values); 
     if (values.length == 1) { 
      //The first call to publishProgress 
      mProgressBar.setMax(Integer.parseInt(values[0])); 
     } else { 
      //Subsequent calls to publish progress 
      Log.i(TAG, "Current item is " + values[0] + " and current file is " + values[1]); 
      infoText.setText(values[1]); 
      mProgressBar.setProgress(Integer.parseInt(values[0]), true); 

     } 
    } 

    @Override 
    protected void onCancelled() { 
     cancel(true); 
    } 
} 

代碼中使用RxJava

final Observable<String[]> addBooksObserver = Observable.create(new Observable.OnSubscribe<String[]>() { 
     @Override 
     public void call(Subscriber<? super String[]> subscriber) { 
      subscriber.onNext(setAddSubscription()); 
      subscriber.onCompleted(); 
     } 
    }) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()); 

private String[] setAddSubscription() { 
    S//Initializing custom SQLiteOpenHelper and SQLite database 
    File mFile = new File(mFolderPath); 

    int booksSize = getFilesInFolder(mFile).size(); 
    String[] sizeList = {String.valueOf(booksSize)}; 
    //The first publishProgress is used to set the max of the progressbar 
    addBooksObserver.doOnNext(addReturnParams(String.valueOf(sizeList.length), null, null)); 

    for (int i = 0; i < booksSize; i++) { 
     EpubReader reader = new EpubReader(); 
     //publishProgress with current item, current file* 
     addBooksObserver.doOnNext(addReturnParams(String.valueOf(sizeList.length), 
       String.valueOf(i), getFilesInFolder(mFile).get(i).getName())); 
     //Inserting current item in database. Code removed 
    } 
    return null; 
} 

private String[] addReturnParams(String totalItems, String currentItem, String currentFile) { 
    return new String[]{totalItems, currentItem, currentFile}; 
} 

代碼的問題是，線addBooksObserver.doOnNext(addReturnParams(正在顯示該錯誤 doOnNext（rx.functions。 Action1）不能應用於（java.lang.String []）

我不知道如何解決這個問題，因爲我認爲setAddSubscription()和addReturnParams(String totalItems, String currentItem, String currentFile)返回字符串數組，那麼這應該不成問題。請你幫我一下嗎？

Answer 1

你只需要值傳遞給onNext方法你subscriber的，不你觀察到的doOnNext方法！

您還必須訂閱該服務。嘗試這樣的事情你obserable：

Observable.create(new Observable.OnSubscribe<String[]>() { 
    @Override 
    public void call(Subscriber<? super String[]> subscriber) { 
    setAddSubscription(subscriber); 
    subscriber.onCompleted(); 
    } 
}) 
.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()) 
.subscribe(new Subscriber<String[]>() { 
    @Override 
    public void onCompleted() { 
    // handle 'oparation is done' 
    } 

    @Override 
    public void onError(Throwable e) { 

    } 

    @Override 
    public void onNext(String[] values) { 
    if (values.length == 1) { 
     //The first call to publishProgress 
     mProgressBar.setMax(Integer.parseInt(values[0])); 
    } else { 
     //Subsequent calls to publish progress 
     Log.i(TAG, "Current item is " + values[0] + " and current file is " + values[1]); 
     infoText.setText(values[1]); 
     mProgressBar.setProgress(Integer.parseInt(values[0]), true); 

    } 
    } 
}); 

你還需要修改你的私有方法一點點：

private void setAddSubscription(Subscriber<? super String[]> subscriber) { 
    //Initializing custom SQLiteOpenHelper and SQLite database 
    File mFile = new File(mFolderPath); 

    int booksSize = getFilesInFolder(mFile).size(); 
    String[] sizeList = {String.valueOf(booksSize)}; 
    //The first publishProgress is used to set the max of the progressbar 
    subscriber.onNext(addReturnParams(String.valueOf(sizeList.length), null, null)); 

    for (int i = 0; i < booksSize; i++) { 
    EpubReader reader = new EpubReader(); 
    //publishProgress with current item, current file* 
    subscriber.onNext(addReturnParams(String.valueOf(sizeList.length), 
      String.valueOf(i), getFilesInFolder(mFile).get(i).getName())); 
    //Inserting current item in database. Code removed 
    } 

} 

private String[] addReturnParams(String totalItems, String currentItem, String currentFile) { 
    return new String[]{totalItems, currentItem, currentFile}; 
} 

Answer 2

你Observer應該像Observable.create(new Observable.OnSubscribe<String>() &在call方法，你應該通過它的StringArray &傳球onNext例如循環：

@Override 
public void call(Subscriber<? super String> subscriber) { 
    for(String val : setAddSubscription()) { 
     subscriber.onNext(val); 
    } 
    subscriber.onCompleted(); 
} 

現在onNext應當返還您的個別項目& onCompleted會於被稱爲循環完成。

---

編輯

myObserver.subscribe(new Subscriber<String>() { 
    @Override 
    public void onCompleted() { 
    // handle completion. 
    } 

    @Override 
    public void onError(Throwable e) { 

    } 

    @Override 
    public void onNext(String value) { 
    // do whatever with each value passed to onNext 
    } 
});