Disconnect for WebSocketSubject

I have an Angular project in which I want to implement receiving data from the server via a websocket.

To implement the websocket service, I followed the instructions angular-websocket-starter

But this websocket implementation does not have a method to disconnect the socket. I need the socket to be closed when the component in which it is called is destroyed. I added a disconnect method for the service, but when the component is destroyed, the socket reconnects and starts working again.

websocket.service.ts:

import { Injectable, OnDestroy, Inject } from '@angular/core';
import { Observable, SubscriptionLike, Subject, Observer, interval } from 'rxjs';
import { filter, map } from 'rxjs/operators';
import { WebSocketSubject, WebSocketSubjectConfig } from 'rxjs/webSocket';

import { share, distinctUntilChanged, takeWhile } from 'rxjs/operators';
import { IWebsocketService, IWsMessage, WebSocketConfig } from './websocket.interfaces';
import { config } from './websocket.config';

export const wsUrl = 'wss://example.com:1234';


@Injectable({
providedIn: 'root'
})
export class WebsocketService implements IWebsocketService, OnDestroy {

private config: WebSocketSubjectConfig<IWsMessage<any>>;

private websocketSub: SubscriptionLike;
private statusSub: SubscriptionLike;

private reconnection$: Observable<number>;
private websocket$: WebSocketSubject<IWsMessage<any>>;
private connection$: Observer<boolean>;
private wsMessages$: Subject<IWsMessage<any>>;

private reconnectInterval: number;
private reconnectAttempts: number;
private isConnected: boolean;

public status: Observable<boolean>;

constructor(@Inject(config) private wsConfig: WebSocketConfig) {
this.wsMessages$ = new Subject<IWsMessage<any>>();

this.reconnectInterval = wsConfig.reconnectInterval || 5000; // pause between connections
this.reconnectAttempts = wsConfig.reconnectAttempts || 10; // number of connection attempts

this.config = {
url: wsConfig.url,
closeObserver: {
next: (event: CloseEvent) => {
console.log('WebSocket disconnected!');
this.websocket$ = null;
this.connection$.next(false);
}
},
openObserver: {
next: (event: Event) => {
console.log('socket event:',event);
console.log('WebSocket connected!');
this.connection$.next(true);
}
}
};

// connection status
this.status = new Observable<boolean>((observer) => {
this.connection$ = observer;
}).pipe(share(), distinctUntilChanged());

// run reconnect if not connection
this.statusSub = this.status
.subscribe((isConnected) => {
this.isConnected = isConnected;

if (!this.reconnection$ && typeof(isConnected) === 'boolean' && !isConnected) {
this.reconnect();
}
});

this.websocketSub = this.wsMessages$.subscribe(
null, (error: ErrorEvent) => console.error('WebSocket error!', error)
);

this.connect();
}

disconnect(): void { // <----I have added this method
this.websocket$ = null;
this.wsMessages$.complete();
this.connection$.next(false);
this.connection$.complete();
}

ngOnDestroy(): void {
this.websocketSub.unsubscribe();
this.statusSub.unsubscribe();
}


/*
* connect to WebSocked
* */
private connect(): void {
this.websocket$ = new WebSocketSubject(this.config);

this.websocket$.subscribe(
(message) => this.wsMessages$.next(message),
(error: Event) => {
if (!this.websocket$) {
// run reconnect if errors
this.reconnect();
}
});
}


/*
* reconnect if not connecting or errors
* */
private reconnect(): void {
this.reconnection$ = interval(this.reconnectInterval)
.pipe(takeWhile((v, index) => index < this.reconnectAttempts && !this.websocket$));

this.reconnection$.subscribe(
() => this.connect(),
null,
() => {
// Subject complete if reconnect attemts ending
this.reconnection$ = null;

if (!this.websocket$) {
this.wsMessages$.complete();
this.connection$.complete();
}
});
}

/*
* on message event
* */
public on<T>(event: string): Observable<T> {
if (event) {
return this.wsMessages$.pipe(
filter((message: IWsMessage<T>) => message.event === event),
map((message: IWsMessage<T>) => message.data)
);
}
}
}


The component in which the socket is called:

ngOnInit(): void {
this.websocketService.on<any[]>('messages').subscribe(res => {
console.log('response from server:',res);
});
});

ngOnDestroy(): void {
this.websocketService.disconnect(); <---this method doesn't work
}


What do I need to change in the service so that the disconnect occurs along with the destruction of the component?

Continue reading...