I am writing an Angular Websocket service that can send messages to a server and then can subscribe to the response. I am grateful for every suggestion to improve this.
import { Injectable, OnDestroy } from '@angular/core';
import {webSocket, WebSocketSubject} from "rxjs/webSocket";
import {of, from, Observable, timer, BehaviorSubject, pipe, EMPTY} from "rxjs";
import {share, filter, timeout, switchMap, take, catchError, mergeMap} from "rxjs/operators";
export class WsMessage{
Id: number;
TimeStamp: Date;
ResponseId: number;
MessageData: string;
}
class MessageQueue{
Message: WsMessage;
TimeStamp: Date;
SyncId: number;
}
@Injectable({ providedIn: 'root'})
export class WebSocketServiceService implements OnDestroy {
// Each message has a maximum lifetime of 10 mimnutes.
private readonly MaxMessageQueueTime = 600;
private clearQueueTimer: any;
private messageQueue: Array<MessageQueue> = new Array<MessageQueue>();
private socket: WebSocketSubject<WsMessage>;
private idGenerator: number;
private messageReceiver = new BehaviorSubject<WsMessage>(null);
public OnMessages: Observable<WsMessage>;
constructor(){
this.OnMessages = this.messageReceiver.asObservable();
// Check every 30s for clearing old messages.
this.clearQueueTimer = setInterval(() =>{
if(this.messageQueue){
let maxTime = new Date();
maxTime = new Date(maxTime.setTime(maxTime.getTime() +
(this.MaxMessageQueueTime * 1000)));
this.messageQueue= this.messageQueue.filter(m => m.TimeStamp <= maxTime);
}
}, 30000);
}
public Init(url: string){
this.idGenerator = 0;
if(this.socket!= null) {
this.socket.unsubscribe();
this.socket = null;
}
this.socket = webSocket<WsMessage>({
url: url,
openObserver: {
next: value => {
console.log(value);
}
}
});
this.socket.subscribe(msg => {
if(msg != null && msg.ResponseId > 0){
let m = new MessageQueue();
m.SyncId = msg.Id;
m.Message = msg;
m.TimeStamp = new Date();
this.messageQueue.push(m);
}
this.messageReceiver.next(msg);
console.log("msg received: " + msg);
},
error1 => {
console.log("error received: " + error1);
},
() => {
console.log("connection closed!");
});
}
public sendAndWait(msg: WsMessage, wait: number): Observable<WsMessage>{
if(msg == null){
return null;
}
let id = this.idGenerator++;
msg.Id = id;
if(this.socket == null){
return null;
}
this.socket.next(msg);
let result = this.OnMessages.pipe(
filter(p => p.Id == id),
take(1),
timeout(wait),
catchError((error) => {
console.log("timeout: " + error);
return EMPTY;
})).pipe(share());
result.pipe(untilDestroyed(this)).subscribe(value => {
if(value != null) {
this.messageQueue.splice(this.messageQueue.findIndex(i => i.SyncId == id), 1);
}
});
return result;
}
ngOnDestroy(): void {
clearInterval(this.clearQueueTimer);
if(this.socket != null) {
this.socket.unsubscribe();
}
}
}