angular - RxJS 取消订阅钩子(Hook)

Question

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

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

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

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

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

Answer 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();