更新
我正在寻找一种方法来控制一个可以被另一个观察到的流量。例如，让我们有2个单数递增（重要）的整数可观察值：
source : 1----2-2---2--3--3--4----4--5---6----8---9---10--------11------
control : -1----3----------------5-----------6-------9-----------12------
我需要生成一个新的Observable，其元素与源完全匹配，但是它们的时序由控件Observable进行以下控制：源值应始终小于或等于控件值。这意味着只有所有大于最近发布的控件的源值都应等待，直到它们被控件“释放”为止
source : 1----2-2---2--3--3--4----4--5---6----8---9---10--------11------
control : -1----3----------------5-----------6-------9-----------12------
expected result: -1----2-2--2--3--3-----4-4--5------6-------8-9---------10-11---
请看下面的代码示例：
private static <T, C> Observable<T> combine(Observable<T> source, Observable<C> control, BiFunction<T, C, Boolean> predicate) {
// ???
}
@Test
public void testControl() throws InterruptedException {
Subject<Integer> control = PublishSubject.create();
Observable<Integer> source = Observable.fromArray(1, 2, 2, 2, 3, 3, 4, 4, 5, 6, 8, 10, 11);
Observable<Integer> combined = combine(source, control, (s, c) -> s <= c);
control.subscribe(val -> System.out.println("Control: " + val));
combined.observeOn(Schedulers.io()).subscribe(val -> System.out.println("Value: " + val));
control.onNext(3); // should release 1,2,2,2,3,3
Thread.sleep(1000);
control.onNext(6); // should release 4,4,5,6
Thread.sleep(1000);
control.onNext(11); // should release 8,10,11
Thread.sleep(1000);
}
由于找不到合适的解决方案，因此我最终自己实现了它。如果有人可以提出更优雅的解决方案，我会很高兴（在这种情况下，我将不接受此答案，并且会接受更好的答案）。以下是我的解决方案：
private static <T, C> Observable<T> combine(Observable<T> source, Observable<C> control, BiFunction<T, C, Boolean> predicate) {
return Observable.create(emitter -> {
Queue<T> buffer = new ArrayDeque<>();
AtomicReference<C> lastControl = new AtomicReference<>();
CompletableSubject sourceCompletable = CompletableSubject.create();
CompletableSubject controlCompletable = CompletableSubject.create();
Disposable disposable = new CompositeDisposable(
control.subscribe(
val -> {
lastControl.set(val);
synchronized (buffer) {
while (!buffer.isEmpty() && predicate.apply(buffer.peek(), val)) {
emitter.onNext(buffer.poll());
}
}
},
emitter::onError,
controlCompletable::onComplete),
source.subscribe(
val -> {
C lastControlVal = lastControl.get();
synchronized (buffer) {
if (lastControlVal != null && predicate.apply(val, lastControlVal)) {
emitter.onNext(val);
} else {
buffer.add(val);
}
}
},
emitter::onError,
sourceCompletable::onComplete),
controlCompletable.andThen(sourceCompletable).subscribe(emitter::onComplete));
emitter.setDisposable(disposable);
});
}
本文收集自互联网，转载请注明来源。
如有侵权，请联系 [email protected] 删除。
我来说两句