rxjs - 如果在X时间内未观察到可观察值，则产生副作用

Question

我正在研究一个用例，该用例要求如果可观察对象在一定时间内没有发出值，那么我们应该做一些副作用。

给出一个实际的用例:

打开Web套接字连接

如果在X时间内未发送/接收任何消息，则关闭Web套接字连接并通知用户

这要求在每个发出的值上启动计时器，并在初始订阅observable时启动计时器，然后在分配的时间之后或直到发出复位计时器的值之前，该计时器将运行某些功能。我正在努力以Rx方式执行此操作。任何帮助，将不胜感激 :)

Answer 1

debounceTime 是您要查找的运算符:如果在特定的超时时间内没有其他跟随者，则它仅发出一个值。监听debounce d流的第一条消息将使您超时并清理Websocket连接。如果您需要从打开流开始开始超时，则只需startWith即可。具体来说:

messages$.startWith(null)
         .debounceTime(timeout)
         .take(1)
         .subscribe(() => { /* side effects */ });

编辑:如果相反，您希望在消息流超时时完全结束消息流(例如，在onComplete处理程序中进行清理)，只需将 debounceTime塞入 takeUntil即可:

messages$.takeUntil(
  messages$.startWith(null)
           .debounceTime(timeout)
).subscribe(timeout_observer);

使用 timeout_observable: Observer<TMessage>，其中包含您对onComplete进行的清理。