Angular4 Websocket rxjs Reconnect and onError



看起来有几个类似的问题,但几天后我找不到正确的答案。我的问题是如何知道服务器是否已关闭 websocket 以及如何尝试重新连接。我已经看过几个示例,但是当我想在更改视图时实现从客户端关闭 websocket 的功能时,它们都没有正常工作。

然后我找到了这个例子,它是我迄今为止见过的最好的例子,通过一些小的修改,我能够添加一个运行良好的关闭函数。

从客户端关闭 websocket 不再是问题,但是,我无法知道服务器何时关闭websocket 以及如何再次重新连接

我的代码与这个问题非常相似,但我要求不同的东西。此外,我在重复使用 websocket 时遇到了问题,直到我在我放置的链接中看到他们谈论的共享功能,因此在我发布的类中,websocket 服务和使用 websocket 服务的服务合并为一个

我的网络套接字服务

import { Injectable } from '@angular/core';
import { Observable, Observer, Subscription } from 'rxjs';
import { Subject } from 'rxjs/Subject';
import { ConfigService } from "../config.service";
@Injectable()
export class WSEtatMachineService {
public messages: Subject<any>  = new Subject<any>();
private url: string = '';
static readonly ID = 'machine';
private _subject: Subject<MessageEvent>;
private _subjectData: Subject<number>;
private _ws: any;
constructor(private configService: ConfigService) {
console.log('constructyor ws machine service')
this.setUrl(WSEtatMachineService.ID)
}
public setUrl(id:string) {
const host = this.configService.getConfigReseau().ipServer;
const port = this.configService.getConfigReseau().portServer;
this.url = `ws://${host}:${port}/` + id 
}
public connect() {
console.log('connect ws machine service ', this.url)
this.messages = <Subject<any>>this._connect(this.url)
.map((response: any): any => {
console.log('ws etat machine service: ', response)
return JSON.parse(response.data);
})
}
public close() {
console.log('on closing WS');
this._ws.close()
this._subject = null
}
public send(msg: any) {
this.messages.next(JSON.stringify(msg));
}
// Private methods to create the websocket
public _connect(url: string): Subject<MessageEvent> {
if (!this._subject) {
this._subject = this._create(url);
}
return this._subject;
}
private _create(url: string): Subject<MessageEvent> {
this._ws = new WebSocket(url);
let observable = Observable.create(
(obs: Observer<MessageEvent>) => {
this._ws.onmessage = obs.next.bind(obs);
this._ws.onerror   = obs.error.bind(obs);
this._ws.onclose   = obs.complete.bind(obs);
return this._ws.close.bind(this._ws);
}).share();
let observer = {
next: (data: Object) => {
if (this._ws.readyState === WebSocket.OPEN) {
this._ws.send(JSON.stringify(data));
}
}
};
return Subject.create(observer, observable);
}
} // end class 

然后在组件中我做:

constructor( private wsMachineService: WSMachineService) { ... }
ngOnInit() { 
...
this.wsMachineService.connect();
// connexion du web socket
this.wsMachineService.messages.subscribe(
machine => {
console.log(" wsMachineService open and alive", machine);
},
err => {
// This code is never executed
console.log(" wsMachineService closed by server!!", err);
}
);
}
ngOnDestroy() {
//let tmp = this.synoptiqueSocketSubscription.unsubscribe();
this.wsMachineService.messages.unsubscribe();
this.wsMachineService.close()
}

我想我在_create函数中缺少一些东西,因为我尝试在连接函数的主题中捕获它不起作用。

关于我如何知道它是否被关闭并尝试再次重新连接的任何想法?

谢谢


编辑: 我认为我的问题与主题/可观察量有关,因为我无法完全控制它们。我有一个旧方法,我可以知道服务器何时死机,并且它试图每 X 秒重新连接一次,但不幸的是,我无法关闭来自客户端的连接,因为我无法访问 websocket 对象。我也添加了代码:

public messages: Observable<any>;
private ws: Subject<any>;
private url: string;
public onclose = new Subject();
public connect(urlApiWebSocket: string): Observable<any> {
if (this.messages && this.url === urlApiWebSocket) {
return this.messages;
}
this.url = urlApiWebSocket;
this.ws = Observable.webSocket({
url: urlApiWebSocket,
closeObserver: this.onclose
});
return this.messages = this.ws.retryWhen(errors => errors.delay(10000)).map(msg => msg).share();
}
send(msg: any) {
this.ws.next(JSON.stringify(msg));
}

让我们看看是否有任何方法可以将这两种解决方案结合起来。

嗯,我找到了一种方法。我正在使用Observable.websocket的旧方法

@Injectable()
export class WSMyService {
private url: string = '';
static readonly ID = 'mytestwebsocket';
readonly reload_time = 3000;
public messages: Observable<any>;
private ws: Subject<any>;
public onclose = new Subject();
constructor(private configService: ConfigService) {
console.log('constructor ws synop service')
this.setUrl(WSActionFrontService.ID)
}
public setUrl(id:string) {
...
this.url = `ws://${host}:${port}/` + id 
}
public connect(): Observable<any> {
this.ws = Observable.webSocket({
url: this.url,
closeObserver: this.onclose
});
return this.messages = this.ws.retryWhen(errors => errors.delay(this.reload_time)).map(msg => msg).share();
}
send(msg: any) {
this.ws.next(JSON.stringify(msg));
}
public close() {
console.log('on closing WS');
this.ws.complete();
}

当我使用它时:

constructor(
private wsMyService: WSMyService,
ngOnInit():
...
this.mySocketSubscription = this.wsMyService.connect().subscribe(message => {
... }
this.wsMyService.onclose.subscribe(evt => {
// the server has closed the connection
})
ngOnDestroy() {
this.wsMyService.close();
this.mySocketSubscription.unsubscribe();
...
}

看起来我所要做的就是调用函数 complete(),它告诉服务器客户端已经完成。 我相信有更好的方法,但这是我发现的唯一适合我的方法。

谢谢你的帮助。

您无法知道服务器何时关闭与客户端的连接。

正如您所注意到的,只有当客户端完成"关闭"操作时,才会触发this._ws.onclose = obs.complete.bind(obs);

常见的玩法:

  • 关闭服务器时,向所有客户端发送特殊消息以通知它。
  • 创建ping机制询问服务器是否还活着。

最新更新