代码之家 › 专栏 › 技术社区 › THIAGO DE BONIS

如何在异步/等待之前取消订阅websocket主题?

websocket angular

THIAGO DE BONIS · 技术社区 · 2 年前

A. .on() 创建函数以连接到主题,并且 .off() 函数断开与主题的连接。

这个 在…上 函数是 async/await 并返回给我一个主题。

主题实际上是与websocket的通信。

我需要付出 关 这是在另一个时间取消订阅主题,但这不起作用。

只有我这样做才有效 关 之后 在…上 .

// DONT WORK
this.websocketTopicTest.off('FOO');
await this.websocketTopicTest.on('FOO');


// WORK
await this.websocketTopicTest.on('FOO');
this.websocketTopicTest.off('FOO');

1 回复 | 直到 2 年前

Ivan Ivanyuk 2 年前

您可以尝试使用来自的websocket包装器 rxjs https://rxjs.dev/api/webSocket/webSocket#examples

import { webSocket } from 'rxjs/webSocket';

const subject = webSocket('ws://localhost:8081');

const observableA = subject.multiplex(
  () => ({ subscribe: 'A' }), // When server gets this message, it will start sending messages for 'A'...
  () => ({ unsubscribe: 'A' }), // ...and when gets this one, it will stop.
  message => message.type === 'A' // If the function returns `true` message is passed down the stream. Skipped if the function returns false.
);

const observableB = subject.multiplex( // And the same goes for 'B'.
  () => ({ subscribe: 'B' }),
  () => ({ unsubscribe: 'B' }),
  message => message.type === 'B'
);

const subA = observableA.subscribe(messageForA => console.log(messageForA));
// At this moment WebSocket connection is established. Server gets '{"subscribe": "A"}' message and starts sending messages for 'A',
// which we log here.

const subB = observableB.subscribe(messageForB => console.log(messageForB));
// Since we already have a connection, we just send '{"subscribe": "B"}' message to the server. It starts sending messages for 'B',
// which we log here.

subB.unsubscribe();
// Message '{"unsubscribe": "B"}' is sent to the server, which stops sending 'B' messages.

subA.unsubscribe();
// Message '{"unsubscribe": "A"}' makes the server stop sending messages for 'A'. Since there is no more subscribers to root Subject,
// socket connection closes.

off 当我们取消订阅时,模拟应该正确实现并运行