Цепочка наблюдаемых на основе результата

Я полный новичок в rx-java и rx-android. Я слышал, что кривая обучения довольно крутая в начале.

Я пытаюсь заменить весь код на основе Eventbus на более безопасную альтернативу с помощью rx-android.

Я настроил этот фрагмент для создания наблюдаемых из событий изменения текста редактирования текста:

Основное действие

RxUtils.createEditTextChangeObservable(txtInput).throttleLast(200, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread()).subscribe(new Action1<EditText>() {
            @Override
            public void call(EditText editText) {
                searchStopResultFragment.query(editText.getText().toString());
            }
        });

RxUtils:

public static Observable<EditText> createEditTextChangeObservable(final EditText editText){
        return Observable.create(new Observable.OnSubscribe<EditText>() {
            @Override
            public void call(final Subscriber<? super EditText> subscriber) {
                editText.addTextChangedListener(new TextWatcher() {
                    @Override
                    public void beforeTextChanged(CharSequence s, int start, int count, int after) {

                    }

                    @Override
                    public void onTextChanged(CharSequence s, int start, int before, int count) {

                    }

                    @Override
                    public void afterTextChanged(Editable s) {
                        if (subscriber.isUnsubscribed()) return;
                        subscriber.onNext(editText);
                    }
                });
            }
        });
    }

SearchStopResultFragment:

public void query(String query){
        lastQuery = query;
        resultObservable = StopProvider.getStopResultObservable(getActivity().getContentResolver(),query);
        subscription = resultObservable.subscribeOn(Schedulers.newThread())
                    .observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<List<Stop>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(List<Stop> stops) {
                if(!lastQuery.equals("")) {

                    if(stops.size()>0) {

                        ArrayList<AdapterItem> items = adapter.getItems();
                        items.clear();

                        for (Stop stop : stops) {
                            SearchResultStopItem item = new SearchResultStopItem(stop, SearchResultStopItem.STOP);
                            items.add(item);

                        }

                        adapter.setItems(items);
                        adapter.notifyDataSetChanged();
                    }else{
                      //DO A NOTHER ASYNC QUERY TO FETCH RESULTS
                    }
                }else{
                    showStartItems();
                }
            }
        });
    }

Такое ощущение, что я делаю это неправильно. Я создаю новые наблюдаемые из метода запроса в моем фрагменте при каждом событии изменения текста. Я также хочу создать новую операцию асинхронного поиска на основе результата в StopProvider.getStopResultObservable (см. комментарий)

Любые мысли?


android reactive-programming rx-java rx-android
person Richard    schedule 23.08.2015    source источник
comment
сделайте так, чтобы ваш метод query() возвращал Observable и использовал flatMap для цепочки событий EditText с запросами   -  person krp    schedule 23.08.2015

Ответы (3)


arrow_upward
2
arrow_downward

Вот что я придумал:

RxUtils.createEditTextChangeObservable(txtInput)
.throttleLast(200, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())
.map(EXTRACT_STRING)
.filter(STRING_IS_NOT_EMPTY)
.concatMap(new Func1<EditText, Observable<Pair<String,List<Stop>>>>() {

    @Override
    public Observable<Pair<String, List<Stop>>> call(final String query) {

        return StopProvider.getStopResultObservable(getContentResolver(), query)
        .map(new Func1<List<Stop>, Pair<String, List<Stop>>>() {
            // I think this map is a bit more readable than the 
            // combineLatest, and since "query" should not be changing 
            // anyway, the result should be the same (you have to 
            // declare it as final in the method signature, though
            @Override
            public Pair<String, List<Stop>> call(List<Stop> stops) {
                return new Pair(query, stops);
            }
        });
    }
)
.concatMap(new Func1<Pair<String, List<Stop>>, Observable<List<Stop>>>() {

    @Override
    public Observable<List<Stop>> call(Pair<String, List<Stop>> queryAndStops) {
        if (queryAndStops.second.size() == 0) {
            return RestClient.service().locationName(queryAndStops.first)
                       .map(new Func1<LocationNameResponse, List<Stop>>() {

                            @Override
                            public List<Stop> call(LocationNameResponse locationNameResponse) {
                                // since there was no if-else in your original code (you were always
                                // just wrapping the List in an Observable) I removed that, too
                                return locationNameResponse.getAddresses();
                            }
            });
        } else {
            return Observable.just(queryAndStops.second);
        }
    }
)
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.compose(this.<List<Stop>>bindToLifecycle())
.subscribe(new Action1<List<Stop>>() {
    @Override
    public void call(List<Stop> stops) {
        // since I don't know what your API is returning I think
        // it's saver to keep this check in:
        if (stops != null) {
            searchStopResultFragment.showStops(stops);
        } else {
            searchStopResultFragment.showStartItems();
        }
    }
},
new Action1<Throwable>() {
    @Override
    public void call(Throwable throwable) {
        showError(throwable);
    }
});

куда:

public static final Func1<EditText, String> EXTRACT_STRING = new Func1<EditText, String>() {

    @Override
    public void String call(EditText editText) {
        return editText.getText().toString();
    }
};

public static final Func1<String, Boolean> STRING_IS_NOT_EMPTY = new Func1<String, Boolean>() {

    @Override
    public void String call(String string) {
        return !string.isEmpty();
    }
};

Таким образом, это, по крайней мере, устраняет необходимость возвращать Observable.just(null), а затем проверять это по цепочке.

person david.mihola    schedule 03.09.2015

arrow_upward
2
arrow_downward

Вы можете переместить свой второй concatMap в единственное место, где он вам нужен - после объединения последних

    RxUtils.createEditTextChangeObservable(txtInput)
            .throttleLast(200, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread())
            .concatMap(new Func1<EditText, Observable<Pair<String, List<Stop>>>>() {
                @Override
                public Observable<Pair<String, List<Stop>>> call(EditText editText) {
                    String query = editText.getText().toString();
                    //searchStopResultFragment.setLastQuery(query);
                    if (query.isEmpty()) {
                        return Observable.just(null);
                    }
                    return Observable
                            .combineLatest(StopProvider.getStopResultObservable(getContentResolver(), query), Observable.just(query), new Func2<List<Stop>, String, Pair<String, List<Stop>>>() {
                                @Override
                                public Pair<String, List<Stop>> call(List<Stop> stops, String s) {
                                    return new Pair(s, stops);
                                }
                            })
                            .concatMap(new Func1<R, Observable<? extends Pair<String, List<Stop>>>>() {
                                @Override
                                public Observable<? extends Pair<String, List<Stop>>> call(R r) {
                                    if (queryAndStops.second.size() == 0) {
                                        return RestClient.service().locationName(queryAndStops.first).concatMap(new Func1<LocationNameResponse, Observable<? extends List<Stop>>>() {
                                            @Override
                                            public Observable<? extends List<Stop>> call(LocationNameResponse locationNameResponse) {
                                                return Observable.just(locationNameResponse.getAddresses());
                                            }
                                        });
                                    } else {
                                        return Observable.just(queryAndStops.second);
                                    }
                                }
                            });
                }
            })
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread()).compose(this.<List<Stop>>bindToLifecycle())
            .subscribe(new Action1<List<Stop>>() {
                @Override
                public void call(List<Stop> stops) {
                    if (stops != null) {
                        searchStopResultFragment.showStops(stops);
                    } else {
                        searchStopResultFragment.showStartItems();
                    }

                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    showError(throwable);
                }
            });
person krp    schedule 02.09.2015

arrow_upward
0
arrow_downward

Решил это с помощью concatmap и объединил последние:

 RxUtils.createEditTextChangeObservable(txtInput).throttleLast(200, TimeUnit.MILLISECONDS, AndroidSchedulers.mainThread()).concatMap(new Func1<EditText, Observable<Pair<String,List<Stop>>>>() {
            @Override
            public Observable<Pair<String, List<Stop>>> call(EditText editText) {
                String query = editText.getText().toString();
                //searchStopResultFragment.setLastQuery(query);
                if(query.isEmpty()){
                    return Observable.just(null);
                }
                return Observable.combineLatest(StopProvider.getStopResultObservable(getContentResolver(), query), Observable.just(query), new Func2<List<Stop>, String, Pair<String, List<Stop>>>() {
                    @Override
                    public Pair<String, List<Stop>> call(List<Stop> stops, String s) {
                        return new Pair(s,stops);
                    }
                });
            }
        }).concatMap(new Func1<Pair<String, List<Stop>>, Observable<List<Stop>>>() {
            @Override
            public Observable<List<Stop>> call(Pair<String, List<Stop>> queryAndStops) {
                if(queryAndStops!=null) {
                    if (queryAndStops.second.size() == 0) {
                        return RestClient.service().locationName(queryAndStops.first).concatMap(new Func1<LocationNameResponse, Observable<? extends List<Stop>>>() {
                            @Override
                            public Observable<? extends List<Stop>> call(LocationNameResponse locationNameResponse) {
                                return Observable.just(locationNameResponse.getAddresses());
                            }
                        });
                    } else {
                        return Observable.just(queryAndStops.second);
                    }
                }
                return Observable.just(null);

            }
        }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread()).compose(this.<List<Stop>>bindToLifecycle()).subscribe(new Action1<List<Stop>>() {
            @Override
            public void call(List<Stop> stops) {
                if (stops != null) {
                    searchStopResultFragment.showStops(stops);
                }else{
                    searchStopResultFragment.showStartItems();
                }

            }
        }, new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
                showError(throwable);
            }
        });

Однако есть ли более приятный способ вырваться из цепочки без отправки Observable.just(null) и проверки нулей в следующем вызове?

person Richard    schedule 26.08.2015