RxJS:从嵌套订阅取消订阅

Question

我使用SockJS和StompJS，当我在浏览器中打开我的应用程序时，有时它会在连接到websocket之前尝试订阅一些主题。我希望主题订阅等待应用程序连接到websocket。

export class SocksService {
...

public subscribe<T>(destination: string, callback?: (body: T) => void, headers?: object): Observable<Subscription> {
const subscription = new Subject<Subscription>();
headers = headers || {};

this.status.pipe(
  first(status => status === ConnectionStatus.CONNECTED)
).subscribe((status: ConnectionStatus) => {
  subscription.next(this.stompClient.subscribe(destination, (message: Message) => {
    if (callback) {
      callback(JSON.parse(message.body) as T);
    }
  }, headers));
});

return subscription.asObservable();
  }

...
}

这就是我实现此代码的原因，我将其称为：

this.socksService.subscribe<User>('/topic/user', (user: User) => {
  console.log('user received', user);
}).subscribe(subscription => this.userSubscription = subscription);

因此，我只在连接状态为connected时才订阅该主题，并且只有在客户端首次成功连接时才会调用该主题。

我想稍后从主题中取消订阅，所以我需要内部订阅返回的订阅对象，我还需要内部订阅的消息。

我所实现的很好，但我认为一定有更好的方法来实现它。

（我试过rx stomp，但它有很多错误。）

Answer 1

你说过你试过rx-践踏，但是你试过rx-践踏的Angular2版本吗？它将关键类公开为角注入服务。

看到它在rx-践踏下的提及，在这里：https://github.com/stomp-js/ng2-stompjs#documentation

...它的实现（角7）通过一个很好的一步一步，这里：https://stomp-js.github.io/guide/ng2-stompjs/ng2-stomp-with-angular7.html#receiving-messages

例如。：

this.rxStompService.watch('/topic/demo').subscribe((message: Message) => {
      this.receivedMessages.push(message.body);
    });