我正在寻找一种方法来定义观察者的顺序。
@GET("/get_user_msgs")
Observable<PrivateMessagesResponse> getPrivateMessages(@QueryMap Map<String, String> params);
例如，我给出了一个由Retrofit创建的Rest API中的Observable。
在我的ListView中，我正在观察这个Observable。
api.getPrivateMessages(params).subscribe(new Observer());
我也有一个API包装我的Espresso测试，我订阅相同的Observable那里。这样，首先调用API包装器中的观察者，然后才调用ListView中的观察者被称为.
public class IdlingWrapper implements Api, IdlingResource {
....
public IdlingWrapper(Api realApi) {
this.realApi = realApi;
}
...
public Observable<PrivateMessagesResponse> getPrivateMessages(@QueryMap Map<String, String> params); {
counter.incrementAndGet();
return wrapObservable(realApi.getPrivateMessages(params));
}
protected <T> Observable<T> wrapObservable(final Observable<PrivateMessagesResponse> observable) {
//what to do here?
}
}
是否有一种方法可以在所有其他观察者完成后强制通知某些观察者?或者类似的事情?
之类的Observable observable = getObservable();
observable.subscribeAsLast(new LastObserver());
observable.subscribe(new ObserverA());
observable.subscribe(new ObserverB());
因此ObserverA将首先被通知，然后是ObserverB，然后是LastObserver。
或任何其他方法，我可以找到所有注册的观察者何时被通知并完成。
我不太确定你想在IdlingWrapper中做什么，但我认为当前的实现非常脆弱。
我认为需要做的最重要的事情是保证可观察对象只能被调用一次。
下面是一个快速的实现来演示我的wrapObservable的实现。
public class Test {
private static int counter = 0;
private static final List<Observable<?>> list = Collections.synchronizedList(new ArrayList<>());
protected static <T> Observable<T> wrapObservable(final Observable<T> original) {
// run atleast once???
synchronized (list) {
list.add(original);
}
return Observable.create(new Observable.OnSubscribe<Void>() {
@Override
public void call(Subscriber<? super Void> subscriber) {
synchronized (list) {
counter++;
if (!list.contains(original)) {
subscriber.onError(new Exception("You can only subscribe once!"));
return;
}
list.remove(original);
}
// Sleep to make it easier to see things happening...
try {
Thread.sleep(3000);
} catch (InterruptedException ignored) {
}
subscriber.onCompleted();
}
}).flatMap(new Func1<Void, Observable<? extends T>>() {
@Override
public Observable<? extends T> call(Void o) {
return original;
}
}).finallyDo(new Action0() {
@Override
public void call() {
synchronized (list) {
counter--;
if (list.size() == 0 && counter == 0) {
System.err.println("finally");
}
}
}
});
}
public static void main(String[] args) throws InterruptedException {
for(int i = 0; i < 10; i++) {
// running in io thread for simulating async call.
Observable<String> test = wrapObservable(Observable.from("TEST!!!!!!")).subscribeOn(Schedulers.io());
test.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
System.err.println("completed");
}
@Override
public void onError(Throwable e) {
System.err.println("error");
}
@Override
public void onNext(String s) {
System.err.println("next");
}
});
// example of calling the same observable twice.
test.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
System.err.println("completed");
}
@Override
public void onError(Throwable e) {
System.err.println("error");
}
@Override
public void onNext(String s) {
System.err.println("next");
}
});
}
Thread.sleep(10000);
}
}
看起来，这工作得很好。
protected <T> Observable<T> wrapObservable(final Observable<T> original) {
return Observable.create(new Observable.OnSubscribeFunc<T>() {
@Override
public Subscription onSubscribe(final Observer<? super T> t1) {
original.subscribe(new Observer<T>() {
@Override
public void onCompleted() {
t1.onCompleted();
uiThreadHandler.post(new Runnable() {
@Override
public void run() {
counter.decrementAndGet();
notifyIdle();
}
});
}
@Override
public void onError(Throwable e) {
t1.onError(e);
uiThreadHandler.post(new Runnable() {
@Override
public void run() {
counter.decrementAndGet();
notifyIdle();
}
});
}
@Override
public void onNext(T args) {
t1.onNext(args);
}
});
return Subscriptions.empty();
}
});
}
如果您只想使用内置的RxJava方法来排序您的观察者，您可以使用flatMap和range将每个项转换为多个项，每个项都有一个优先级，然后根据优先级进行筛选。观察器是根据它们的过滤方式排序的。
这里有一个简单的例子:
Observable<Pair<Integer, Object>> shared = RxView.clicks(findViewById(R.id.textView))
.flatMap(c -> Observable.range(0, 2).map(i -> Pair.create(i, c)))
.share();
shared.filter(p -> p.first == 1)
.map(p -> p.second)
.doOnSubscribe(c -> Log.d(TAG, "first subscribed doOnSubscribe"))
.subscribe(c -> Log.d(TAG, "first subscribed onNext"));
shared.filter(p -> p.first == 0)
.map(p -> p.second)
.doOnSubscribe(c -> Log.d(TAG, "second subscribed doOnSubscribe"))
.subscribe(c -> Log.d(TAG, "second subscribed onNext"));
如果你到处都这样做