有没有办法知道 rxjs websocket 是否打开

Question

我在 Angular 4 项目中使用 RxJS。

我正在尝试启动一个 Websocket，尤其是要知道这个 Websocket 何时打开。

我目前正在使用来自 RxJS (v5) 的 WebSocket。 https://github.com/ReactiveX/rxjs/blob/master/src/observable/dom/WebSocketSubject.ts

我注意到 WebSocketSubjectConfig 中有一个 openObserver，但我找不到如何创建 Observer。我已经被锁定了几个小时。

到目前为止，这是我的代码摘录：

import { Injectable } from '@angular/core';
import { webSocket} from 'rxjs/observable/dom/webSocket';
import { WebSocketSubject, WebSocketSubjectConfig} from 'rxjs/observable/dom/WebSocketSubject';

@Injectable()
export class MzkWebsocketJsonRpcService {
    subject: WebSocketSubject<any>;
    jsonRpcId: number;

    constructor() {

        this.subject = webSocket('ws://localhost:12345');
        this.subject.openObserver =
            /// Find a way to create the openObserver


        this.subject.subscribe(
            this.onMessage,
            this.onError,
            this.onClose,
        );
        console.log('Socket connected');
        this.jsonRpcId = 1;
    }

    public send(method: string, params: any[]) {

        let jsonFrame: any = {id: this.jsonRpcId, 'json-rpc': '2.0', method: method};

        if (params) {
            jsonFrame['params'] = params;
        }
        this.subject.next(JSON.stringify(jsonFrame));
        this.jsonRpcId ++;
    }

    onMessage(data: string) {
        console.log('Websocket message: ', data);
    }

    onError(data: string) {
        console.log('Websocket error:', data);
    }

    onClose() {
        console.log('Websocket closing');
    }
}

Answer 1

观察者可以是至少部分实现Observer 接口的任何对象。见https://github.com/ReactiveX/rxjs/blob/master/src/Observer.ts

这意味着您可以编写一个自定义类：

MyObserver implements Observer {
  next(value: any): void {
    ...
  }
  complete(): void {
    ...
  }
}

let socket = new WebSocketSubject({
  url: 'ws://localhost:8081',
  openObserver: new MyObserver()
});

最终，在WebSocketSubject 的情况下，您可以使其更简单，并仅使用next 方法创建一个对象，因为openObserver 期望和对象具有NextObserver 接口https://github.com/ReactiveX/rxjs/blob/master/src/Observer.ts#L1。

let socket = new WebSocketSubject({
  url: 'ws://localhost:8081',
  openObserver: {
    next: (val: any) => {
      console.log('opened');
    }
  }
});