RxJava合并去抖动而不是去抖动可观察者

Question

我有两个可观察量：

Observable TO（open）：在textview中打开一些内容的文件

可观察的E（编辑）：在textview中编辑的文件内容

我想去掉E observable，并将它与O observable合并。

obs = Observable.merge(E.debounce(2000, TimeUnit.MILLISECONDS) , O)
                .subscribe(content->System.out.println("new content: " + content))

问题是如果E发出事件E1并且在O发出O1事件之后，我们有输出：

new content: O1
new content: E1 // this output is rebundant (cuz we already have newer content O1)

这是一个正在发生的事情的图表：

如何从去抖动的观察中摆脱这个过度的旧事件？

Answer 1

你可以试试

Observable.merge(O, O.switchMap(o -> E.debounce()))
          .subscribe()

switchMap的行为与flatMap非常相似，只是当源Observable发出新项时，它将取消订阅并停止镜像从先前发出的项生成的Observable，并开始仅镜像当前的项。

Answer 2

我看到两个主要选择。一种是使用时间戳，这很容易，但有理论上的竞争条件（但很可能是不可能的），另一种选择是使用与每个文件打开相关的唯一标识符，并且从编辑到该打开文件的文本发出的事件是附带文件打开的标识符。

使用时间戳：

obs = Observable.defer(() -> {
  AtomicBoolean first = new AtomicBoolean(true);
  e.timestamp()
   .debounce(2000, TimeUnit.MILLISECONDS))
   .mergeWith(o.timestamp())
   .buffer(2,1)
   .flatMap(list -> {
     Observable<Object> start;
     if (first.compareAndSet(true, false)) 
         start = Observable.just(list.get(0).getValue ());
     else 
         start = Observable.empty();
     if (list.size() == 1) 
         return start;
     else {
       Timestamped<Object> a = list.get(0);
       Timestamped<Object> b = list.get(1);
       if (a.getTimestampMillis() <= b.getTimestampMillis()) 
         return start.concatWith(Observable.just(b.getValue ()));
       else 
         return start; 
     }
   })
});

我怀疑带时间戳的版本就足够了。