Angular 2跨组件共享Websocket服务 [英] Angular 2 share websocket service across components

问题描述

我正在使用angular 2构建一个Web应用程序，在该应用程序中我想让多个组件侦听同一服务.该服务返回一个可观察对象，该可观察对象返回来自websocket的传入数据.我基于此示例编写了代码.

I am building a web application using angular 2 in which I want to have multiple components listening to the same service. This service returns an observable that returns incoming data from a websocket. I wrote the code based on this example.

当前问题是: 数据通过服务从本地组件通过服务发送到服务器(使用websocket)，并返回数据.但是，只有home.component中的观察者被调用(ID:room.created和data)，而不是导航栏中的观察者.

The current problem is: The data is send from the home component through the service to the server (using websockets) and data is returned. However, only the observer in the home.component is getting called (with id: room.created and data), not the one in the navbar.

有人可以告诉我为什么不同时打电话给我吗?我还尝试将message $ .subscribe添加到app.component，但无济于事.

Could someone tell me why not both are called? I also tried to add the messages$.subscribe to the app.component but to no avail.

现在，让我们看一下代码.

Now, let's get to the code.

返回可观察值的消息服务.组件使用此服务从发送和接收消息.

A message service that returns an observable. This service is used by the components to send and receive messages from.

@Injectable()
export class MessageService {
    private _messages: Rx.Subject<Message>;
    messages$: Rx.Observable<Message>;

    constructor(wsService: SocketService, private configuration: Configuration) {
      console.log('messag eservice');
      this._messages = <Rx.Subject<Message>>wsService
        .connect()
        .map((response: MessageEvent): Message => {
            let data = JSON.parse(response.data);
            return {
                id: data.id,
                data: data.data,
            }
        });

      this.messages$ = this._messages.asObservable();
    }

    public send(message: Message): void {
      this._messages.next(message);
    }
} 

套接字服务，该服务创建一个websocket连接并将其自身绑定到此套接字的输入和输出.

A socket service which creates a websocket connection and binds itself to input and output of this socket.

import { Injectable } from '@angular/core';
import * as Rx from "rxjs/Rx";
import { Configuration } from '../app.constants';

@Injectable()
export class SocketService {
    private subject: Rx.Subject<MessageEvent>;

    constructor(private configuration: Configuration){};

    public connect(wsNamespace = ''): Rx.Subject<MessageEvent> {
        var url = this.configuration.wsUrl + wsNamespace;
        if(!this.subject) {
            this.subject = this.create(url);
        }
        return this.subject;
    }

    private create(url): Rx.Subject<MessageEvent> {
        let ws = new WebSocket(url);

        // bind ws events to observable (streams)
        let observable = Rx.Observable.create((obs: Rx.Observer<MessageEvent>) => {
            ws.onmessage = obs.next.bind(obs);
            ws.onerror = obs.error.bind(obs);
            ws.onclose = obs.complete.bind(obs);

            return ws.close.bind(ws);
        });

        // on obs next (send something in the stream) send it using ws.
        let observer = {
            next: (data: Object) => {
                if (ws.readyState === WebSocket.OPEN) {
                    ws.send(JSON.stringify(data));
                }
            },
        };

        return Rx.Subject.create(observer, observable);
    }
}

具有以下提供程序的应用程序组件:

An app component with the following providers:

  providers: [MessageService, SocketService, Configuration, AuthService]

我正在实例化我的主要app.component中的提供者，以确保消息和套接字服务不会被实例化两次.

I am instantiating the providers in my main app.component to make sure that the messages and socket services are not instantiated twice.

我的home.component看起来像这样(这是使用路由加载的页面):

My home.component looks like this (this is a page being loaded using routing):

import { Component, OnInit } from '@angular/core';
import { Subject } from 'rxjs';
import { Router }    from '@angular/router';
import { MessageService } from '../../services/message.service';

@Component({
  ...
  providers: []
})
export class HomeComponent implements OnInit {
  constructor(private router: Router, private messageService: MessageService) {}

  ngOnInit(): void {
    this.messageService.send({
      id: 'room.create',
      data: {'name': 'Blaat'}
    });

    this.messageService.messages$.subscribe(msg => {
      console.log(msg);
        if(msg.id == 'room.created') {
            // navigate naar games!
        }
    });
  }

}

我的导航栏组件如下所示(指令):

My navbar component looks like this (directive):

import { Component, OnInit } from '@angular/core';
import { MessageService } from '../../services/message.service';

@Component({
  moduleId: module.id,
  selector: 'navbar',
  templateUrl: 'navbar.component.html',
  styleUrls: ['navbar.component.css']
})
export class Navbar implements OnInit {

  constructor(private messageService: MessageService) { }

  ngOnInit() {

    this.messageService.messages$.subscribe(msg => {
      console.log(msg);
        if(msg.id == 'room.created') {
            // navigate naar games!
        }
    });
  }

}

推荐答案

似乎可观察的create函数被多次调用，很可能是两个组件=>两个订阅=>两个可观察的create函数调用.因此，最新的可观察的create fn会覆盖对websocket onmessage，onerror和onclose的先前可观察的回调.您应该多播底层的可观察对象，以防止这种情况发生(共享操作员应做到这一点).

It seems that your observable create function is called multiple times, most probably two components => two subscriptions => two observable create function invocations. So the latest observable create fn overrides previous observable callbacks to websocket onmessage, onerror and onclose. You should multicast the underlying observable to prevent that (share operator should do the trick).

        // bind ws events to observable (streams)
        let observable = Rx.Observable.create((obs: Rx.Observer<MessageEvent>) => {
            ws.onmessage = obs.next.bind(obs);
            ws.onerror = obs.error.bind(obs);
            ws.onclose = obs.complete.bind(obs);

            return ws.close.bind(ws);
        }).share();

有关如何正确执行此操作的更多有用资源 https://github.com/ReactiveX/rxjs/blob/master/src/observable/dom/WebSocketSubject.ts https://github.com/blesh/RxSocketSubject

More useful resources of how to do this properly https://github.com/ReactiveX/rxjs/blob/master/src/observable/dom/WebSocketSubject.ts https://github.com/blesh/RxSocketSubject

这篇关于Angular 2跨组件共享Websocket服务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！