as

dd

/**
 * Port from {@link https://github.com/ReactiveX/RxSwift}
 */
RxTextView.textChanges(mSearchbar.getEditTextSearch())
  .subscribeOn(AndroidSchedulers.mainThread())
  //delay 500ms
  //debounce and throttle will use different thread after
  .throttleWithTimeout(300, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())
  .distinct()
  .filter(new Func1() {
  @UiThread @Override public Boolean call(CharSequence charSequence) {
  //void unnecessary request
  return charSequence.length() != 0;
  }
  })
  .map(new Func1() {
  @UiThread @Override public String call(CharSequence charSequence) {
  //fit network api doc require
  return charSequence + "*";
  }
  })
  .doOnNext(new Action1() {
  @UiThread @Override public void call(CharSequence charSequence) {
  progressBar.setVisibility(View.VISIBLE);
  arrayList.clear();
  adapter.notifyDataSetChanged();
  }
  })
  .observeOn(Schedulers.io())
  .switchMap(new Func1>>() {
  @WorkerThread @Override public Observable> call(String s) {
  return repo.getTags(20, s);
  }
  })
  .retry(new Func2() {
  //fix InterruptedIOException bugs on Retrofit
  // when stop old search
  @WorkerThread @Override public Boolean call(Integer integer, Throwable throwable) {
  return throwable instanceof InterruptedIOException;
  }
  })
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(new Subscriber>() {
  @Override public void onCompleted() {

  }

  @Override public void onError(Throwable e) {
  progressBar.setVisibility(View.INVISIBLE);
  e.printStackTrace();
  Toast.makeText(SearchActivity.this, e.getMessage(), Toast.LENGTH_SHORT).show();
  }

  @Override public void onNext(List tags) {
  progressBar.setVisibility(View.INVISIBLE);

  arrayList.clear();
  arrayList.addAll(tags);
  adapter.notifyDataSetChanged();
  }
  });