Как реализовать периодическую обработку пользовательского ввода?

Что-то вроде этого должно работать (на самом деле не пробовал)

 Single<String> firstTypeOnlyStream = RxTextView.textChanges(mEditText)
            .skipInitialValue()
            .map(CharSequence::toString)
            .firstOrError();

    Observable<CharSequence> restartTypingStream = RxTextView.textChanges(mEditText)
            .filter(charSequence -> charSequence.length() == 0);

    Single<String> latestTextStream = RxTextView.textChanges(mEditText)
            .map(CharSequence::toString)
            .firstOrError();

    Observable<TextViewEditorActionEvent> enterStream =
            RxTextView.editorActionEvents(mEditText, actionEvent -> actionEvent.actionId() == EditorInfo.IME_ACTION_DONE);

    firstTypeOnlyStream
            .flatMapObservable(__ ->
                    latestTextStream
                            .toObservable()
                            .doOnNext(text -> nextDelay = delayByLength(text.length()))
                            .repeatWhen(objectObservable -> objectObservable
                                    .flatMap(o -> Observable.timer(nextDelay, TimeUnit.MILLISECONDS)))
                            .distinctUntilChanged()
                            .flatMap(text -> {
                                if (text.length() > MINIMUM_TEXT_LENGTH) {
                                    return apiRequest(text);
                                } else {
                                    return Observable.empty();
                                }
                            })
            )
            .takeUntil(restartTypingStream)
            .repeat()
            .takeUntil(enterStream)
            .mergeWith(enterStream.flatMap(__ ->
                    latestTextStream.flatMapObservable(this::apiRequest)
            ))
            .subscribe(requestResult -> {
                //do your thing with each request result
            });

Идея состоит в том, чтобы создать поток на основе выборки, а не самих событий изменения текста, исходя из вашего требования выполнять выборку каждый X раз.

То, как я сделал это здесь, состоит в том, чтобы создать один поток (firstTypeOnlyStream для начального запуска событий (первый ввод текста пользователем), этот поток запустит весь поток обработки с первого ввода пользователя, затем, когда это При поступлении первого триггера мы будем периодически сэмплировать текст редактирования, используя latestTextStream. latestTextStream на самом деле не поток с течением времени, а скорее выборка текущего состояния EditText с использованием свойства InitialValueObservable RxBinding (он просто выдает при подписке текущий текст на EditText) другими словами, это причудливый способ получить текущий текст при подписке, и он эквивалентен:
Observable.fromCallable(() -> mEditText.getText().toString());
далее, для динамического тайм-аута/задержки, мы обновляем nextDelay на основе длины текста и использования repeatWhen с таймером ждать нужное время, вместе с distinctUntilChanged должно давать желаемую выборку на основе длины текста, далее мы будем запускать запрос на основе текста (если он достаточно длинный).

Stop by Enter — используйте takeUntil с enterStream, которые будут срабатывать при Enter, а также запускать окончательный запрос.

Перезапуск — когда пользователь «перезапускает» набор текста — т. е. текст пуст, .takeUntil(restartTypingStream) + repeat() остановит поток при вводе пустой строки и перезапустит его (повторная подписка).

Ну, вы могли бы использовать что-то вроде этого:

RxSearch.fromSearchView(searchView)
            .debounce(300, TimeUnit.MILLISECONDS)
            .filter(item -> item.length() > 1)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(query -> {
                adapter.setNamesList(namesAPI.searchForName(query));
                adapter.notifyDataSetChanged();
                apiCallsTextView.setText("API CALLS: " + apiCalls++);
            });    

public class RxSearch {
    public static Observable<String> fromSearchView(@NonNull final SearchView searchView) {
        final BehaviorSubject<String> subject = BehaviorSubject.create("");

        searchView.setOnQueryTextListener(new SearchView.OnQueryTextListener() {

            @Override
            public boolean onQueryTextSubmit(String query) {
                subject.onCompleted();
                return true;
            }

            @Override
            public boolean onQueryTextChange(String newText) {
                if (!newText.isEmpty()) {
                    subject.onNext(newText);
                }
                return true;
            }
        });

        return subject;
    }
}

ссылка на блог

ваш запрос можно легко решить с помощью методов RxJava2, прежде чем я опубликую код, я добавлю шаги того, что я делаю.

1. добавьте PublishSubject, который будет принимать ваши входные данные, и добавьте к нему фильтр, который будет проверять, больше ли входных данных два или нет.
2. добавьте метод debounce, чтобы все входные события, которые запускаются до 300 мс, игнорировались, а окончательный запрос, который запускается после 300 мс, принимался во внимание.
3. теперь добавьте карту переключения и добавьте в нее событие сетевого запроса,
4. Подпишитесь на ваше мероприятие.

Код выглядит следующим образом:

subject = PublishSubject.create(); //add this inside your oncreate
    getCompositeDisposable().add(subject
            .doOnEach(stringNotification -> {
                if(stringNotification.getValue().length() < 3) {
                    getMvpView().hideEditLoading();
                    getMvpView().onFieldError("minimum 3 characters required");
                }
            })
            .debounce(300,
                    TimeUnit.MILLISECONDS)
            .filter(s -> s.length() >= 3)
            .switchMap(s -> getDataManager().getHosts(
                    getDataManager().getDeviceToken(),
                    s).subscribeOn(Schedulers.io()))
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(hostResponses -> {
                getMvpView().hideEditLoading();
                if (hostResponses.size() != 0) {
                    if (this.hostResponses != null)
                        this.hostResponses.clear();
                    this.hostResponses = hostResponses;
                    getMvpView().setHostView(getHosts(hostResponses));
                } else {
                    getMvpView().onFieldError("No host found");
                }

            }, throwable -> {
                getMvpView().hideEditLoading();
                if (throwable instanceof HttpException) {
                    HttpException exception = (HttpException) throwable;
                    if (exception.code() == 401) {
                        getMvpView().onError(R.string.code_expired,
                                BaseUtils.TOKEN_EXPIRY_TAG);
                    }
                }

            })
);

это будет ваш текстовый наблюдатель:

searchView.addTextChangedListener(new TextWatcher() {
        @Override
        public void beforeTextChanged(CharSequence charSequence, int i, int i1, int i2) {

        }

        @Override
        public void onTextChanged(CharSequence charSequence, int i, int i1, int i2) {
            subject.onNext(charSequence.toString());
        }

        @Override
        public void afterTextChanged(Editable editable) {

        }
    });

P.S. Это работает для меня!!

Вы можете найти то, что вам нужно, в as оператор. Требуется ObservableConverter, который позволяет вам преобразовать ваш источник Observable в произвольный объект. Этот объект может быть другим Observable с произвольно сложным поведением.

public class MyConverter implements ObservableConverter<Foo, Observable<Bar>> {
    Observable<Bar> apply(Observable<Foo> upstream) {
        final PublishSubject<Bar> downstream = PublishSubject.create();
        // subscribe to upstream
        // subscriber publishes to downstream according to your rules
        return downstream;
    }
}

Затем используйте его следующим образом:

someObservableOfFoo.as(new MyConverter())... // more operators

Изменить: я думаю, что compose может быть более парадигматическим. Это менее мощная версия as специально для создания Observable вместо любого объекта. Использование по сути такое же. См. это руководство.