Angular2/Websocket：如何为传入的 websocket 消息返回一个 observable

Question

我将使用 Angular2 接收 websocket 传入消息并根据收到的消息更新网页。现在，我正在使用一个虚拟的 echo websocket 服务，并将替换它。

据我了解，接收 websocket 消息的函数必须返回一个由处理程序订阅的可观察对象，该处理程序将更新网页。但我不知道如何返回一个 observable。

代码sn-p附在下面。 MonitorService 创建一个 websocket 连接并返回一个包含接收到的消息的 observable。

@Injectable()
export class MonitorService {

    private actionUrl: string;
    private headers: Headers;
    private websocket: any;
    private receivedMsg: any;
    constructor(private http: Http, private configuration: AppConfiguration) {

        this.actionUrl = configuration.BaseUrl + 'monitor/';
        this.headers = new Headers();
        this.headers.append('Content-Type', 'application/json');
        this.headers.append('Accept', 'application/json');
    }

    public GetInstanceStatus = (): Observable<Response> => {
        this.websocket = new WebSocket("ws://echo.websocket.org/"); //dummy echo websocket service
        this.websocket.onopen =  (evt) => {
            this.websocket.send("Hello World");
        };

        this.websocket.onmessage = (evt) => { 
            this.receivedMsg = evt;
        };

        return new Observable(this.receivedMsg).share();
    }

}

下面是另一个组件，它订阅从上面返回的 observable 并相应地更新网页。

export class InstanceListComponent {
  private instanceStatus: boolean
  private instanceName: string
  private instanceIcon: string
  constructor(private monitor: MonitorService) { 
    this.monitor.GetInstanceStatus().subscribe((result) => {
        this.setInstanceProperties(result);
    });
  }

  setInstanceProperties(res:any) {
    this.instanceName = res.Instance.toUpperCase();
    this.instanceStatus = res.Status;
    if (res.Status == true)
    {
      this.instanceIcon = "images/icon/healthy.svg#Layer_1";
    } else {
      this.instanceIcon = "images/icon/cancel.svg#cancel";
    }
  }
}

现在，我在浏览器控制台中遇到了这个错误 TypeError: this._subscribe is not a function

Answer 1

我把它放在plunker 上，并添加了一个用于向 Websocket 端点发送消息的函数。这是重要的编辑：

public GetInstanceStatus(): Observable<any>{
    this.websocket = new WebSocket("ws://echo.websocket.org/"); //dummy echo websocket service
    this.websocket.onopen =  (evt) => {
        this.websocket.send("Hello World");
    };
    return Observable.create(observer=>{
        this.websocket.onmessage = (evt) => { 
            observer.next(evt);
        };
    })
    .share();
}

更新
正如您在评论中提到的，更好的替代方法是使用Observable.fromEvent()

websocket = new WebSocket("ws://echo.websocket.org/");
public GetInstanceStatus(): Observable<Event>{
    return Observable.fromEvent(this.websocket,'message');
}

plunker example 为Observable.fromEvent()；

此外，您可以使用WebSocketSubject 来完成，不过，它看起来还没有准备好（从 rc.4 开始）：

constructor(){
  this.websocket = WebSocketSubject.create("ws://echo.websocket.org/");
}

public sendMessage(text:string){
  let msg = {msg:text};
  this.websocket.next(JSON.stringify(msg));
} 

plunker example

Answer 2

从套接字获取 onMessage 数据。

import { Injectable } from '@angular/core';
import {Observable} from 'rxjs/Rx';


@Injectable()

export class HpmaDashboardService {
private socketUrl: any = 'ws://127.0.0.0/util/test/dataserver/ws';
private websocket: any;      

public GetAllInstanceStatus(objStr): Observable<any> {
      this.websocket = new WebSocket(this.socketUrl);
      this.websocket.onopen =  (evt) => {
        this.websocket.send(JSON.stringify(objStr));
      };
      return Observable.create(observer => {
        this.websocket.onmessage = (evt) => {
          observer.next(evt);
        };
      }).map(res => res.data).share();
 }

**Get only single mesage from socket.**

    public GetSingleInstanceStatus(objStr): Observable<any> {
      this.websocket = new WebSocket(this.socketUrl);
      this.websocket.onopen =  (evt) => {
        this.websocket.send(JSON.stringify(objStr));
      };
      return Observable.create(observer => {
        this.websocket.onmessage = (evt) => {
          observer.next(evt);
          this.websocket.close();
        };
      }).map(res => res.data).share();
  }

}