Angular 2 跨组件共享 websocket 服务 [英] Angular 2 share websocket service across components
问题描述
我正在使用 angular 2 构建一个 Web 应用程序，我想让多个组件监听同一个服务.此服务返回一个可观察的对象，该对象从 websocket 返回传入的数据.我根据 this 示例编写了代码.
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.
目前的问题是:数据从 home 组件通过服务发送到服务器(使用 websockets)并返回数据.然而，只有 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.
有人能告诉我为什么不叫两者吗?我也尝试将 messages$.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!
}
});
}
}
推荐答案
看来您的 observable create 函数被多次调用，很可能是两个组件 => 两个订阅 => 两个 observable create 函数调用.所以最新的 observable create fn 覆盖了之前对 websocket onmessage、onerror 和 onclose 的 observable 回调.您应该多播底层 observable 以防止这种情况发生(共享操作符应该这样做).
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.tshttps://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屋！
