RxJava外卖直到操作员抛出IllegalStateException

Question

我正在尝试使用RxJava实现一个简单的自动完成视图，其中后端调用在按下cancel按钮时被取消。这是我的密码：

    Observable<String> searchButton$ = RxView
                                        .clicks(searchButton)
                                        .map(item -> mEditText.getText().toString());
    Observable<Integer> cancelRequest$ = RxView
                                        .clicks(cancelRequest)
                                        .scan(0,(counter,string) -> ++counter)
                                        .skip(1);

    Observable<String> search$ = searchButton$
                                            .distinctUntilChanged()
                                            .debounce(1,TimeUnit.SECONDS)
                                            .doOnNext(text-> {
                                                Log.i(TAG,"Searching ["+text+"]");
                                            });
    search$
        .observeOn(Schedulers.single())
        .doOnError(error -> {
            Log.d(TAG,"Error after ObserveOn");
            error.printStackTrace();
        })
        .switchMap(text ->
                fetchList(text)
                .doOnError(error -> {
                    Log.d(TAG,"Error after FetchList");
                    error.printStackTrace();
                })
                .takeUntil(cancelRequest$)
                .doOnError(error -> {
                    Log.d(TAG,"Error after TakeUntil");
                    error.printStackTrace();
                })
        )
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(list -> {
            Log.i(TAG,"Fetched:"+ list.toString());
            resultView.setText("");
            list.forEach(item -> resultView.append(item + "\n"));
        });

但在编辑文本中键入某些内容时，我将得到以下输出

07-19 13:42:14.881 13869-13907/personal.com.actrecog I/MainActivity: Searching [12]
07-19 13:42:14.883 13869-13909/personal.com.actrecog D/MainActivity: Error after TakeUntil
07-19 13:42:14.883 13869-13909/personal.com.actrecog W/System.err: java.lang.IllegalStateException: Expected to be called on the main thread but was RxSingleScheduler-1

我不明白为什么要在mainThread上叫外卖。我尝试在ObserveOn(AndroidSchedulers.mainThread())之前添加takeUntil，但是错误仍然是一样的。

有什么想法吗？

Answer 1

如果您在RxBinding代码中深入挖掘一下，您就会找到答案。

clicks定义：

 /**
   * Create an observable which emits on {@code view} click events. The emitted value is
   * unspecified and should only be used as notification.
   * <p>
   * <em>Warning:</em> The created observable keeps a strong reference to {@code view}. Unsubscribe
   * to free this reference.
   * <p>
   * <em>Warning:</em> The created observable uses {@link View#setOnClickListener} to observe
   * clicks. Only one observable can be used for a view at a time.
   */
  @CheckResult @NonNull
  public static Observable<Object> clicks(@NonNull View view) {
    checkNotNull(view, "view == null");
    return new ViewClickObservable(view);
  }

ViewClickObservable定义：

final class ViewClickObservable extends Observable<Object> {
  private final View view;

  ViewClickObservable(View view) {
    this.view = view;
  }

  @Override protected void subscribeActual(Observer<? super Object> observer) {
    if (!checkMainThread(observer)) {
      return;
    }
    Listener listener = new Listener(view, observer);
    observer.onSubscribe(listener);
    view.setOnClickListener(listener);
  }

  static final class Listener extends MainThreadDisposable implements OnClickListener {
    private final View view;
    private final Observer<? super Object> observer;

    Listener(View view, Observer<? super Object> observer) {
      this.view = view;
      this.observer = observer;
    }

    @Override public void onClick(View v) {
      if (!isDisposed()) {
        observer.onNext(Notification.INSTANCE);
      }
    }

    @Override protected void onDispose() {
      view.setOnClickListener(null);
    }
  }
}

在subscribeActual方法中，您将注意到具有以下定义的!checkMainThread(observer)：

public static boolean checkMainThread(Observer<?> observer) {
    if (Looper.myLooper() != Looper.getMainLooper()) {
      observer.onSubscribe(Disposables.empty());
      observer.onError(new IllegalStateException(
          "Expected to be called on the main thread but was " + Thread.currentThread().getName()));
      return false;
    }
    return true;
  }

换句话说，需要在主线程上订阅RxView.clicks，而不是takeUntil。observeOn()意味着您希望在给定的调度程序上观察结果。在您的情况下，您必须使您的实际工作发生在主线程上。在这种情况下，observeOn()不会对您有所帮助，您必须使用subscribeOn()

.switchMap(text ->
                fetchList(text)
                .doOnError(error -> {
                    Log.d(TAG,"Error after FetchList");
                    error.printStackTrace();
                })
                .takeUntil(cancelRequest$)
                .subscribeOn(AndroidSchedulers.mainThread())
                .doOnError(error -> {
                    Log.d(TAG,"Error after TakeUntil");
                    error.printStackTrace();
                })
        )