android - 如何在 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) 是返回 String 数组，那么这应该不是问题。你能帮帮我吗？

Answer 1

您只需将值传递给订阅者的onNext方法，不是doOnNext方法你的可观察性!

您还必须订阅该服务。为您的观察者尝试这样的事情:

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)方法:

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