angular - 在 Angular 2 中重新订阅以前取消订阅的 Rx.Subject

Question

我不确定，如果我使用了错误的方法，或者我只是编码不好但无法解释我的问题。

我有一个后端组件，我在其中管理 websocket 数组(因为每个 websocket 都可以有不同的 WS url，或者它们需要单独关闭/打开)并且对于每个 websocket，我想知道它的状态和通过 Rx.Subject 将它发送到一个 View (如果它打开、关闭或者它没有建立连接所需的数据)

看起来

public connectDeviceTerminalWebSocket(server: string, port: string): void {
    if (!(server !== null) && !(port !== null)) {

        let websocket: WebSocket = null;

        let wsPosition: number = this.hardwareTerminalwebSockets.findIndex(ws => {
            if (ws.url.includes(server + ':' + port)) {
                websocket = ws;
                return true;
            }
        });

        if (websocket) {
            this.closeHardwareTerminalWebsocket(websocket.url);
        }

        websocket = new WebSocket(`${this.wsProtocol}://${server}:${port}/${this.getToken()}`);

        websocket.addEventListener('close', ws => {
            this.reconnectTerminalWebSocketAfterTimeout();
            this.hardwareTerminalState.next({ 'websocketUrl': websocket.url, 'isConnected': false, 'reason': 'conectionFailed' });
        });

        websocket.addEventListener('open', ws => {
            this.reconnectTerminalWebSocketAfterTimeout();
            this.hardwareTerminalState.next({ 'websocketUrl': websocket.url, 'isConnected': true, 'reason': 'connected' });
        });


        let opened = Rx.Observable
            .fromEvent<void>(websocket, 'open');
        let channelReceived = Rx.Observable
            .fromEvent<MessageEvent>(websocket, 'message')
            .map(event => {
                try {
                    return JSON.parse(event.data);
                } catch (e) {
                    console.error('Parse error: ', e);
                }
                return null;
            });
        channelReceived
            .filter(message => message.message_channel === 'hardware-logger')
            .subscribe(this.hardwareTerminal);


        opened.subscribe(open => this.sendWebSocketTerminalMessageQueue());

        if (wsPosition > -1) {

            this.hardwareTerminalwebSockets[wsPosition] = websocket;
        } else {

            this.hardwareTerminalwebSockets.push(websocket);
        }
    } else {
        this.hardwareTerminalState.next({ websocketUrl: null, isConnected: null, 'reason': 'cantConnect' });
        return;
    }
}

这行得通。

在我看来，我已经定义

hardwareTerminalStateWS: Rx.Subject<ITerminalWebsocketMessage>;

我只用这种方式订阅一次

this.hardwareTerminalStateWS.subscribe(msg => this.onStateMessage(msg));

在我看来在 ngOnDestroy 中

this.hardwareTerminalWS.unsubscribe();

取消订阅。

现在是主要问题如果我关闭/更改为不同的 View ，然后返回(例如重新打开并再次订阅)它会抛出错误

“./hwcomponent 类 hwComponent 中的错误 - 内联模板:47:28 由:对象取消订阅引起”

我应该将 Rx.Subject 更改为不同的类(观察者吗？)？或者使用不同的方法来关闭 hardwareTerminalStateWS？因为如果我不关闭它，下次我再次打开 View 时，它会打开第二个 hardwareTerminalStateWS 并读取每条消息两次。

Answer 1

如果您在 Subject 上调用 unsubscribe - 而不是在 Subscription 上调用 - 您将无法再使用 Subject.

有关详细信息，请参阅 this answer和 Ben Lesh 的 Medium 文章 On The Subject Of Subjects :

If you want the subject to loudly and angrily error when you next to it after it’s done being useful, you can call unsubscribe directly on the subject instance itself.

如果您打算重用一个主题，请改为对订阅(从调用订阅返回)调用取消订阅。

此外，需要注意的是，在 Subject 上调用 unsubscribe 与在 上调用 unsubscribe 效果不同订阅。通常，您会在 Subscription 上调用 unsubscribe，如果您想确保它不能被重复使用。