如果有很大的延迟，如何在项目情感之间的 Observable 上运行函数？

Posted 2023-02-17

【中文标题】如果有很大的延迟，如何在项目情感之间的 Observable 上运行函数？

【英文标题】：How to run a function on an Observable between item emotion if there was a big delay?

【发布时间】：2017-06-06 14:21:23

【问题描述】：

我有一个 PublishSubject 接收来自 UI 的发射：

myPublishSubject
        .map 
            ...
        
        .doOnNext 
            // using emitted item
        
        .timeout (...) // wait for the gap!
        .doOnNext 
            // running a function after a specific gap between two item
        
        .subscribe()

我想在最后一次发射后等待一段特定的时间（不是 onComplete，因为它稍后会继续发射）并运行一个函数。可以理解为项目情感之间的差距。

我正在寻找类似 @​​987654322@ 的东西，但这个方法问题是它会因错误而杀死 Observable。

【问题讨论】：

您可以使用flatMap()/concatMap() 并将延迟放入其中。

@PhoenixWang 我不知道应该如何使用flatMap()/concatMap()！

【参考方案1】：

你必须对publish 和switchMap 有点创意，例如：

PublishSubject<Integer> ps = PublishSubject.create();

ps.publish(o -> 
    o.mergeWith(
        o.switchMap(e -> 
             Observable.just(1).delay(200, TimeUnit.MILLISECONDS)
            .ignoreElements()
            .doOnCompleted(() -> System.out.println("Timeout action: " + e))
        )
    )
).subscribe(System.out::println);

ps.onNext(1);
ps.onNext(2);

Thread.sleep(100);

ps.onNext(3);

Thread.sleep(250);

ps.onNext(4);

Thread.sleep(250);

它通过共享源和路由以两种方式工作，一种是直接发射，另一种是提供switchMap，当接收到新项目时，启动延迟的Observable并对其完成做出反应（忽略原始触发器元素，以避免由于mergeWith 引起的重复事件）。当宽限期内有新信号时，switchMap 将取消之前的延迟，并从新的延迟开始。