java - 如果有很大的延迟，如何在可观察到的项目情绪之间运行函数？

Question

我有一个 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，因为它稍后会继续发出)并运行一个函数。可以理解为元素情感之间的差距。

我正在寻找类似timeout的东西，但这个方法的问题是它会错误地终止Observable。

Answer 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并对其完成使用react(忽略原始触发元素以避免由于 mergeWith 导致的重复事件)。当宽限期内有新信号时，switchMap 将取消之前的延迟并以较新的延迟开始。