Как превратить EditText в поток данных для RxJava2?

#android #rxjava #listener #rxandroid #rxjava2

Начал постепенно переходить на реактивное программирование с замечательным фреймворком
RxJava 2 и заинтересовал простейший пример создания потока в динамическом стиле. К
сожалению, не считаю пример Observable.just(1,2,3) достаточно полным для себя, но вот
уже несколько часов мучаю себя мыслью, как превратить изменения в EditText в поток
данных? Если не сложно, опишите, пожалуйста, пример как сделать реальный Observable
из TextWatcher, чтобы можно было на него подписаться и подписчик реагировал на изменения
текста в EditText. 
    

Ответы

Ответ 1

Можно воспользоваться готовой библиотекой RxBinding. Также в ней можно посмотреть
конкретную реализацию.

Вообще для связки callback-методов с rx используется метод Observable.create, например:

Observable.create(emitter -> {
    TextWatcher watcher = 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) {
        }

        @Override
        public void afterTextChanged(Editable editable) {
            if (!emitter.isDisposed()) { //если еще не отписались
                emitter.onNext(editable.toString()); //отправляем текущее состояние
            }
        }
    };
    emitter.setCancellable(() -> editText.removeTextChangedListener(watcher)); //удаляем
листенер при отписке от observable
    editText.addTextChangedListener(watcher);
});


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

Что выполняют CallAdapterFactory в Retrofit?

#android #kotlin #retrofit #rxjava2

.addCallAdapterFactory(RxJava2CallAdapterFactory.create())
    

Ответы

Ответ 1

С помощью метода addCallAdapterFactory вы расширяете ваш retrofit сторонними библиотеками.
В данном случае вы добавляете поддержку RxJava 2. Список доступных адаптеров тут.

Как правильно заменить AsyncTask с помощью RxJava2 и RxAndroid2

#android #rxjava #rxandroid #rxjava2

У меня есть AsyncTask, который я хочу переделать в Rx 

Вот так выглядит мой AsyncTask

new AsyncTask()
    {
        ArrayList undoneServiceCodes = new ArrayList<>();
        HashMap undoneForms = new HashMap<>();
        boolean isHasAtLeastOneDoneServiceCode = false;

        @Override
        protected void onPreExecute()
        {
            super.onPreExecute();

            if (iCloseCallListener != null)
            {
                iCloseCallListener.onPreValidation();
            }
        }

        @Override
        protected Void doInBackground(Void... params)
        {
            undoneServiceCodes = getUndoneServiceCodes();
            undoneForms = getUndoneForms();
            isHasAtLeastOneDoneServiceCode = isHasAtLeastOneDoneServiceCode();

            return null;
        }

        @Override
        protected void onPostExecute(Void iVoid)
        {
            super.onPostExecute(iVoid);

            if (iCloseCallListener != null)
            {
                iCloseCallListener.onPostValidation(undoneServiceCodes, undoneForms,
isHasAtLeastOneDoneServiceCode);
            }
        }
    }.execute();


Мне не понятно как можно выполнить в бекграунде Rx , можно выполнить сразу 3 разных
метода как в примере с AsyncTask когда в бекграунде выполняется 3 метода


undoneServiceCodes = getUndoneServiceCodes();
undoneForms = getUndoneForms();
isHasAtLeastOneDoneServiceCode = isHasAtLeastOneDoneServiceCode();


Если это был бы один метод(допустим первый) я бы это сделал так

Flowable.fromIterable(getUndoneServiceCodes())//
            .subscribeOn(Schedulers.io())//
            .observeOn(AndroidSchedulers.mainThread())//
            .doOnSubscribe(iSubscription ->
            {
                if (iCloseCallListener != null)
                {
                    iCloseCallListener.onPreValidation();
                }
            }).toList()//
            .subscribe(resultList -> {
                if (iCloseCallListener != null)
                {
                    iCloseCallListener.onPostValidation(iCloseCallListener, ???, ???);
                }
            });


Но так я получу результат только для одного выполняемого в бекграунде метода, как
сделать так, чтоб можно было обрабоать 3?
    

Ответы

Ответ 1

Можно воспользоваться оператором zip как-то так (точность названий методов не гарантирую):

Flowable.zip(
    Flowable.fromCallable(method1()), 
    Flowable.fromCallable(method2()),
    Flowable.fromCallable(method3()),
    (result1, result2, result3) -> new Triple(result1, result2, result3) 
)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(triple -> iCloseCallListener.onPostValidation(triple.first, triple.second,
triple.third))

Возможен ли реактивный маршалинг (marshaling) xml в java-объект?

Я делаю запрос на определенный сервис который отдаем мне очень медленно большой xml
Мне нужно не дожидаясь полного ответа документа начать читать и преобразовывать некоторые элементы xml в java-объекты через Jaxb и производить с ними не который операции
Как такое реализовать с помощью reactive spring и возможно ли?

Ответ

Попытался сделать максимально простой пример. Использовал java 9.
Исходные данные:
Xml вида:
Jack Green Ivan Ivanov
Класс в который будет маппится содержимое тега person
@XmlAccessorType(XmlAccessType.FIELD) private static class Person { @XmlElement private String name; @XmlElement private String surname;
@Override public String toString() { return String.format("Person{name=%s, surname=%s}", name, surname); } }
Теперь нам необходимо написать publisher который будет производить анмаршаллинг:
private static class MyPublisher extends SubmissionPublisher {
void run(String fileName) { try { XMLInputFactory xif = XMLInputFactory.newFactory(); StreamSource xml = new StreamSource(Thread.currentThread().getContextClassLoader().getResourceAsStream(fileName)); XMLStreamReader xsr = xif.createXMLStreamReader(xml); xsr.nextTag(); int count = Integer.parseInt(xsr.getAttributeValue(null, "count"));
for (int i = 0; i < count; i++) { xsr.nextTag(); JAXBContext jc = JAXBContext.newInstance(Person.class); Unmarshaller unmarshaller = jc.createUnmarshaller(); JAXBElement jaxbElement = unmarshaller.unmarshal(xsr, Person.class); Person person = jaxbElement.getValue(); submit(person); } } catch (Exception e) { throw new RuntimeException(e); } } }
Ничего сложного - читаем первый тег, из него получаем значением атрибута count. Далее в цикле читаем последующие теги, преобразуем в объекты класса Person и отправляем подписчиками
Subscriber может выглядить следующим образом:
public static class PrintSubscriber implements Flow.Subscriber { private Flow.Subscription subscription;
@Override public void onSubscribe(Flow.Subscription subscription) { this.subscription = subscription; subscription.request(1); }
@Override public void onNext(Person item) { System.out.println("Received item: " + item); subscription.request(1); }
@Override public void onError(Throwable error) { System.out.println("Error occurred: " + error.getMessage()); }
@Override public void onComplete() { System.out.println("PrintSubscriber is complete"); } }
Как заставить все это работать вместе? Очень просто - создаем сначала publisher, потом subscriber Одного подписываем на другово и запускаем метод run для разбора xml
MyPublisher publisher = new MyPublisher(); publisher.subscribe(new PrintSubscriber()); publisher.run("demo.xml"); TimeUnit.MINUTES.sleep(1); publisher.close();
Запуск
Как известно в java 9 ввели модули, работа с xml отсутствует в дефолтном модуле java.base, поэтому в настройках своего модуля потребуется прописать зависимости. Выглядит это примерно так module-info.java :
open module ru.izebit { requires java.xml; requires java.xml.bind; }
PS java.xml.bind - deprecated, но до java 10 пользоваться еще можно, потом удалят ☹️

Как превратить EditText в поток данных для RxJava2?

Начал постепенно переходить на реактивное программирование с замечательным фреймворком RxJava 2 и заинтересовал простейший пример создания потока в динамическом стиле. К сожалению, не считаю пример Observable.just(1,2,3) достаточно полным для себя, но вот уже несколько часов мучаю себя мыслью, как превратить изменения в EditText в поток данных? Если не сложно, опишите, пожалуйста, пример как сделать реальный Observable из TextWatcher, чтобы можно было на него подписаться и подписчик реагировал на изменения текста в EditText.

Ответ

Можно воспользоваться готовой библиотекой RxBinding. Также в ней можно посмотреть конкретную реализацию
Вообще для связки callback-методов с rx используется метод Observable.create, например:
Observable.create(emitter -> { TextWatcher watcher = 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) { }
@Override public void afterTextChanged(Editable editable) { if (!emitter.isDisposed()) { //если еще не отписались emitter.onNext(editable.toString()); //отправляем текущее состояние } } }; emitter.setCancellable(() -> editText.removeTextChangedListener(watcher)); //удаляем листенер при отписке от observable editText.addTextChangedListener(watcher); });
Нужно не забыть подписаться, и самое главное, отписаться от такого источника данных, т.к. он держит ссылку на editText и может привести к утечке памяти.

Как правильно заменить AsyncTask с помощью RxJava2 и RxAndroid2

У меня есть AsyncTask, который я хочу переделать в Rx
Вот так выглядит мой AsyncTask
new AsyncTask() { ArrayList undoneServiceCodes = new ArrayList<>(); HashMap undoneForms = new HashMap<>(); boolean isHasAtLeastOneDoneServiceCode = false;
@Override protected void onPreExecute() { super.onPreExecute();
if (iCloseCallListener != null) { iCloseCallListener.onPreValidation(); } }
@Override protected Void doInBackground(Void... params) { undoneServiceCodes = getUndoneServiceCodes(); undoneForms = getUndoneForms(); isHasAtLeastOneDoneServiceCode = isHasAtLeastOneDoneServiceCode();
return null; }
@Override protected void onPostExecute(Void iVoid) { super.onPostExecute(iVoid);
if (iCloseCallListener != null) { iCloseCallListener.onPostValidation(undoneServiceCodes, undoneForms, isHasAtLeastOneDoneServiceCode); } } }.execute();
Мне не понятно как можно выполнить в бекграунде Rx , можно выполнить сразу 3 разных метода как в примере с AsyncTask когда в бекграунде выполняется 3 метода
undoneServiceCodes = getUndoneServiceCodes(); undoneForms = getUndoneForms(); isHasAtLeastOneDoneServiceCode = isHasAtLeastOneDoneServiceCode();
Если это был бы один метод(допустим первый) я бы это сделал так
Flowable.fromIterable(getUndoneServiceCodes())// .subscribeOn(Schedulers.io())// .observeOn(AndroidSchedulers.mainThread())// .doOnSubscribe(iSubscription -> { if (iCloseCallListener != null) { iCloseCallListener.onPreValidation(); } }).toList()// .subscribe(resultList -> { if (iCloseCallListener != null) { iCloseCallListener.onPostValidation(iCloseCallListener, ???, ???); } });
Но так я получу результат только для одного выполняемого в бекграунде метода, как сделать так, чтоб можно было обрабоать 3?

Ответ

Можно воспользоваться оператором zip как-то так (точность названий методов не гарантирую):
Flowable.zip( Flowable.fromCallable(method1()), Flowable.fromCallable(method2()), Flowable.fromCallable(method3()), (result1, result2, result3) -> new Triple(result1, result2, result3) ) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(triple -> iCloseCallListener.onPostValidation(triple.first, triple.second, triple.third))