-
Notifications
You must be signed in to change notification settings - Fork 401
/
websocket-handler.ts
154 lines (133 loc) Β· 4.33 KB
/
websocket-handler.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
import { Injectable, Inject } from '@angular/core';
import { Actions, Store, getValue, ofActionDispatched } from '@ngxs/store';
import { WebSocketSubject, WebSocketSubjectConfig } from 'rxjs/webSocket';
import {
ConnectWebSocket,
DisconnectWebSocket,
SendWebSocketMessage,
NGXS_WEBSOCKET_OPTIONS,
NgxsWebsocketPluginOptions,
WebsocketMessageError,
WebSocketDisconnected,
TypeKeyPropertyMissingError,
WebSocketConnectionUpdated,
WebSocketConnected
} from './symbols';
@Injectable()
export class WebSocketHandler {
private socket: WebSocketSubject<any> | null = null;
private config: WebSocketSubjectConfig<any> = {
url: this.options.url!,
protocol: this.options.protocol,
// Default binary type is `blob` for the global `WebSocket`
binaryType: this.options.binaryType,
serializer: this.options.serializer,
deserializer: this.options.deserializer,
closeObserver: {
next: () => {
// ATTENTION!
// See https://github.com/ReactiveX/rxjs/blob/master/src/internal/observable/dom/WebSocketSubject.ts#L340
// RxJS socket emits `onComplete` event only if `event.wasClean` is truthy
// and doesn't complete socket subject if it's falsy
this.disconnect();
}
},
openObserver: {
next: () => this.store.dispatch(new WebSocketConnected())
}
};
private typeKey = this.options.typeKey!;
constructor(
private store: Store,
private actions$: Actions,
@Inject(NGXS_WEBSOCKET_OPTIONS) private options: NgxsWebsocketPluginOptions
) {
this.setupActionsListeners();
}
private setupActionsListeners(): void {
this.actions$.pipe(ofActionDispatched(ConnectWebSocket)).subscribe(({ payload }) => {
this.connect(payload);
});
this.actions$.pipe(ofActionDispatched(DisconnectWebSocket)).subscribe(() => {
this.disconnect();
});
this.actions$.pipe(ofActionDispatched(SendWebSocketMessage)).subscribe(({ payload }) => {
this.send(payload);
});
}
private connect(options?: NgxsWebsocketPluginOptions): void {
this.updateConnection();
// Users can pass the options in the connect method so
// if options aren't available at DI bootstrap they have access
// to pass them here
if (options) {
this.mergeConfigWithOptions(options);
}
this.socket = new WebSocketSubject(this.config);
this.socket.subscribe({
next: (message: any) => {
const type = getValue(message, this.typeKey);
if (!type) {
throw new TypeKeyPropertyMissingError(this.typeKey);
}
this.store.dispatch({ ...message, type });
},
error: (error: any) => {
if (error instanceof CloseEvent) {
this.dispatchWebSocketDisconnected();
} else {
this.store.dispatch(new WebsocketMessageError(error));
}
}
});
}
private disconnect(): void {
if (this.socket) {
// `socket.complete()` closes the connection
// also it doesn't invoke the `onComplete` callback that we passed
// into `socket.subscribe(...)`
this.socket.complete();
this.socket = null;
this.dispatchWebSocketDisconnected();
}
}
private send(data: any): void {
if (!this.socket) {
throw new Error('You must connect to the socket before sending any data');
}
this.socket.next(data);
}
/**
* Don't enlarge the `connect` method
*/
private mergeConfigWithOptions(options: NgxsWebsocketPluginOptions): void {
if (options.url) {
this.config.url = options.url;
}
if (options.serializer) {
this.config.serializer = options.serializer;
}
if (options.deserializer) {
this.config.deserializer = options.deserializer;
}
}
/**
* To ensure we don't have any memory leaks
* e.g. if the user occasionally dispatched `ConnectWebSocket` twice
* then the previous subscription will still live in the memory
* to prevent such behavior - we close the previous connection if it exists
*/
private updateConnection(): void {
if (this.socket) {
this.socket.complete();
this.socket = null;
this.store.dispatch(new WebSocketConnectionUpdated());
}
}
/**
* Used in many places so it's better to move the code into function
*/
private dispatchWebSocketDisconnected(): void {
this.store.dispatch(new WebSocketDisconnected());
}
}