RxJS 取消订阅钩子答案

【问题标题】：RxJS unsubscribe hookRxJS 取消订阅钩子
【发布时间】：2020-09-14 18:32:00
【问题描述】：

有一段代码可以将一些数据提取到缓存中，并使用 websocket 连接保持同步，并使用 Observable 提供对数据的访问接口，我怎么知道所述数据没有订阅者在某个时间点？

我想挂接到 unsubscribe() 对 observable 的调用（并检查订阅者数量是否为 0），但我不确定如何实现。

现实生活中的场景是一个 graphql 查询（apollo-angular）和一个 websocket 订阅以保持数据同步。一旦没有数据的订阅者（例如用户导航到另一个页面），可以关闭websocket连接以节省服务器资源，但需要同时从缓存中删除数据，否则可能会消失不同步，因为不再从服务器接收到推送事件，以防用户再次导航回来。我不愿意将过时的数据保存在缓存中。

那么...我如何才能挂钩可观察对象被销毁/所有订阅者都取消订阅以从缓存中删除数据的事件？该服务设置了一个 graphql 查询并设置了订阅以侦听更改，返回 Observable，但我不确定如何将其删除，因为该实例已通过用户代码传递。

我不是在寻找仅限网络、绕过缓存和类似的 apollo 客户端设置，我正在寻找一种实际使用的方法，但要保持 apollo 缓存中的数据干净和最新，因为服务器提供了一种方法来做到这一点。

【问题讨论】：

你能用 refCount 和一个 Subject 吗？你对 websocket observable 有多少控制权？

标签： angular rxjs graphql rxjs-observables apollo-angular

【解决方案1】：

如果您使用的是 RxJS，则可以利用 WebSocketSubject，它提供了一些不错的选择。

例如，WebSocketSubject 可以被赋予一个配置对象，其中一个有用的选项是closeObserver: NextObserver<T>，它的next 方法在套接字连接关闭时被调用。

当WebSocketSubject没有更多活跃订阅者时，连接自动关闭。

这是上述想法的代码：

const connectionClosed = new Subjet();
const unsubscribe = new Subject();

const webSocket = new WebSocketSubject({
  url: '',
  /* ... */,
  closeObserver: connectionClosed
});

connectionClosed.subscribe(() => {
  // Teardown logic
  // e.g clear cache 
});


// Registering subscribers
webSocket.pipe(takeUntil(unsubscribe)).subscribe(...);
webSocket.pipe(takeUntil(unsubscribe)).subscribe(...);
webSocket.pipe(takeUntil(unsubscribe)).subscribe(...);

// In `ngOnDestroy`, called maybe when the user navigates to another route
unsubscribe.next();
unsubscribe.complete();

【讨论】：