Я новичок в RxJava, но я интегрирую его в проект, над которым я работаю, чтобы помочь мне изучить его. Я столкнулся с вопросом о лучших практиках.
У меня вопрос о том, как обрабатывать onError
от предотвращения остановки обработки Observable
.
Вот настройка:
У меня есть список userIds для каждого из них, который я хотел бы сделать 2 или более сетевых запросов. Если какой-либо из сетевых запросов завершился с ошибкой для userid, тогда этот идентификатор пользователя не будет обновлен и может быть пропущен. Это не должно препятствовать обработке других пользователей. У меня есть решение, но оно связано с вложенными подписчиками (см. Второй блок кода). Одна из проблем, которые я вижу, заключается в том, что если каждый вызов не выполняется, нет возможности короткого замыкания и прекращения останова от попадания сетевого ресурса даже после обнаружения определенного порогового числа.
Есть ли лучший способ сделать это?
В традиционном коде:
List<String> results = new ArrayList<String>();
for (String userId : userIds) {
try {
String info = getInfo(userId); // can throw an GetInfoException
String otherInfo = getOtherInfo(userId); // can throw an GetOtherInfoException
results.add(info + ", " + otherInfo);
} catch (GetInfoException e) {
log.error(e);
} catch (GetOtherInfoException e) {
log.error(e);
}
}
Проблема:
псевдокод:
userid -> network requests -> result
1 -> a, b -> onNext(1[a ,b])
2 -> a, onError -> onError
3 -> a, b -> onNext(3[a, b])
4 -> a, b -> onNext(4[a, b])
Ниже приведен рабочий пример списка userIds и для каждого 2 запроса информации. Если вы запустите его, вы увидите, что он потерпит неудачу (см. Ниже исходный код).
import rx.Observable;
import rx.Observable.OnSubscribeFunc;
import rx.Observer;
import rx.Subscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action0;
import rx.util.functions.Action1;
import rx.util.functions.Func1;
public class TestMergeDelayError {
public static Observable<String> getUserIds() {
return Observable.from(new String[]{"1", "2", "3", "4", "5", "6"});
}
public static Observable<String> getInfo(final String prefix, final String integer, final String errorNumber) {
Observable<String> observable = Observable.create(new OnSubscribeFunc<String>() {
public Subscription onSubscribe(Observer<? super String> t1) {
if (integer.contains(errorNumber)) {
t1.onError(new Exception());
} else {
t1.onNext(prefix + integer);
t1.onCompleted();
}
return Subscriptions.empty();
}
});
return observable;
}
public static void main(String[] args) {
Observable<String> userIdObservable = getUserIds();
Observable<String> t = userIdObservable.flatMap(new Func1<String, Observable<String>>() {
public Observable<String> call(final String t1) {
Observable<String> info1 = getInfo("1::: ", t1, "2");
Observable<String> info2 = getInfo("2::: ",t1, "3");
return Observable.mergeDelayError(info1, info2);
}
});
t.subscribe(new Action1<String>() {
public void call(String t1) {
System.out.println(t1);
}
}, new Action1<Throwable>() {
public void call(Throwable t1) {
t1.printStackTrace();
}
},
new Action0(){
public void call() {
System.out.println("onComplete");
}
});
}
}
Вывод:
1::: 1
2::: 1
2::: 2
java.lang.Exception
at TestMergeDelayError$1.onSubscribe(TestMergeDelayError.java:32)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMergeDelayError$MergeDelayErrorObservable$ParentObserver.onNext(OperationMergeDelayError.java:266)
at rx.operators.OperationMergeDelayError$MergeDelayErrorObservable$ParentObserver.onNext(OperationMergeDelayError.java:210)
at rx.operators.OperationMergeDelayError$2.onSubscribe(OperationMergeDelayError.java:77)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMergeDelayError$MergeDelayErrorObservable.onSubscribe(OperationMergeDelayError.java:171)
at rx.operators.OperationMergeDelayError$1.onSubscribe(OperationMergeDelayError.java:64)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:164)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:116)
at rx.operators.OperationMap$MapObservable$1.onNext(OperationMap.java:105)
at rx.operators.SafeObserver.onNext(SafeObserver.java:102)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMap$MapObservable.onSubscribe(OperationMap.java:102)
at rx.operators.OperationMap$2.onSubscribe(OperationMap.java:76)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable.onSubscribe(OperationMerge.java:106)
at rx.operators.OperationMerge$1.onSubscribe(OperationMerge.java:56)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:483)
Вложенное решение для подписки:
import rx.Observable;
import rx.Observable.OnSubscribeFunc;
import rx.Observer;
import rx.Subscription;
import rx.subscriptions.Subscriptions;
import rx.util.functions.Action0;
import rx.util.functions.Action1;
import rx.util.functions.Func1;
public class TestMergeDelayError {
public static Observable<String> getUserIds() {
return Observable.from(new String[]{"1", "2", "3", "4", "5", "6"});
}
public static Observable<String> getInfo(final String prefix, final String integer, final String errorNumber) {
Observable<String> observable = Observable.create(new OnSubscribeFunc<String>() {
public Subscription onSubscribe(Observer<? super String> t1) {
if (integer.contains(errorNumber)) {
t1.onError(new Exception());
} else {
t1.onNext(prefix + integer);
t1.onCompleted();
}
return Subscriptions.empty();
}
});
return observable;
}
public static void main(String[] args) {
Observable<String> userIdObservable = getUserIds();
userIdObservable.subscribe(new Action1<String>() {
public void call(String t1) {
Observable<String> info1 = getInfo("1::: ", t1, "2");
Observable<String> info2 = getInfo("2::: ", t1, "3");
Observable.merge(info1, info2).subscribe(new Action1<String>() {
public void call(String t1) {
System.out.println(t1);
}
}, new Action1<Throwable>() {
public void call(Throwable t1) {
t1.printStackTrace();
}
},
new Action0() {
public void call() {
System.out.println("onComplete");
}
});
}
});
}
}
Вывод:
1::: 1
2::: 1
onComplete
java.lang.Exception
at TestMergeDelayError$1.onSubscribe(TestMergeDelayError.java:28)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:164)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:116)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable.onSubscribe(OperationMerge.java:106)
at rx.operators.OperationMerge$1.onSubscribe(OperationMerge.java:56)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:483)
at TestMergeDelayError$2.call(TestMergeDelayError.java:47)
at TestMergeDelayError$2.call(TestMergeDelayError.java:42)
at rx.Observable$2.onNext(Observable.java:381)
at rx.operators.SafeObserver.onNext(SafeObserver.java:102)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:367)
at TestMergeDelayError.main(TestMergeDelayError.java:42)
1::: 3
java.lang.Exception
at TestMergeDelayError$1.onSubscribe(TestMergeDelayError.java:28)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:164)
at rx.operators.OperationMerge$MergeObservable$ParentObserver.onNext(OperationMerge.java:116)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.operators.OperationMerge$MergeObservable.onSubscribe(OperationMerge.java:106)
at rx.operators.OperationMerge$1.onSubscribe(OperationMerge.java:56)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:483)
at TestMergeDelayError$2.call(TestMergeDelayError.java:47)
at TestMergeDelayError$2.call(TestMergeDelayError.java:42)
at rx.Observable$2.onNext(Observable.java:381)
at rx.operators.SafeObserver.onNext(SafeObserver.java:102)
at rx.operators.OperationToObservableIterable$ToObservableIterable.onSubscribe(OperationToObservableIterable.java:94)
at rx.Observable.subscribe(Observable.java:241)
at rx.Observable.protectivelyWrapAndSubscribe(Observable.java:320)
at rx.Observable.subscribe(Observable.java:367)
at TestMergeDelayError.main(TestMergeDelayError.java:42)
1::: 4
2::: 4
onComplete
1::: 5
2::: 5
onComplete
1::: 6
2::: 6
onComplete
Как вы можете видеть, только отдельные пользовательские идентификаторы, которые не выполнили, остановили свою индивидуальную обработку, но остальные обработчики были обработаны.
Просто ищите совета, посмотрите, имеет ли это решение смысл, а если нет, то что лучше всего.
Спасибо, Alex