WebSocketManager.js
13.1 KB
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
'use strict';
const EventEmitter = require('events');
const WebSocketShard = require('./WebSocketShard');
const PacketHandlers = require('./handlers');
const { Error: DJSError } = require('../../errors');
const Collection = require('../../util/Collection');
const { Events, ShardEvents, Status, WSCodes, WSEvents } = require('../../util/Constants');
const Util = require('../../util/Util');
const BeforeReadyWhitelist = [
WSEvents.READY,
WSEvents.RESUMED,
WSEvents.GUILD_CREATE,
WSEvents.GUILD_DELETE,
WSEvents.GUILD_MEMBERS_CHUNK,
WSEvents.GUILD_MEMBER_ADD,
WSEvents.GUILD_MEMBER_REMOVE,
];
const UNRECOVERABLE_CLOSE_CODES = Object.keys(WSCodes).slice(1).map(Number);
const UNRESUMABLE_CLOSE_CODES = [1000, 4006, 4007];
/**
* The WebSocket manager for this client.
* <info>This class forwards raw dispatch events,
* read more about it here {@link https://discord.com/developers/docs/topics/gateway}</info>
* @extends EventEmitter
*/
class WebSocketManager extends EventEmitter {
constructor(client) {
super();
/**
* The client that instantiated this WebSocketManager
* @type {Client}
* @readonly
* @name WebSocketManager#client
*/
Object.defineProperty(this, 'client', { value: client });
/**
* The gateway this manager uses
* @type {?string}
*/
this.gateway = null;
/**
* The amount of shards this manager handles
* @private
* @type {number}
*/
this.totalShards = this.client.options.shards.length;
/**
* A collection of all shards this manager handles
* @type {Collection<number, WebSocketShard>}
*/
this.shards = new Collection();
/**
* An array of shards to be connected or that need to reconnect
* @type {Set<WebSocketShard>}
* @private
* @name WebSocketManager#shardQueue
*/
Object.defineProperty(this, 'shardQueue', { value: new Set(), writable: true });
/**
* An array of queued events before this WebSocketManager became ready
* @type {object[]}
* @private
* @name WebSocketManager#packetQueue
*/
Object.defineProperty(this, 'packetQueue', { value: [] });
/**
* The current status of this WebSocketManager
* @type {number}
*/
this.status = Status.IDLE;
/**
* If this manager was destroyed. It will prevent shards from reconnecting
* @type {boolean}
* @private
*/
this.destroyed = false;
/**
* If this manager is currently reconnecting one or multiple shards
* @type {boolean}
* @private
*/
this.reconnecting = false;
/**
* The current session limit of the client
* @private
* @type {?Object}
* @property {number} total Total number of identifies available
* @property {number} remaining Number of identifies remaining
* @property {number} reset_after Number of milliseconds after which the limit resets
*/
this.sessionStartLimit = null;
}
/**
* The average ping of all WebSocketShards
* @type {number}
* @readonly
*/
get ping() {
const sum = this.shards.reduce((a, b) => a + b.ping, 0);
return sum / this.shards.size;
}
/**
* Emits a debug message.
* @param {string} message The debug message
* @param {?WebSocketShard} [shard] The shard that emitted this message, if any
* @private
*/
debug(message, shard) {
this.client.emit(Events.DEBUG, `[WS => ${shard ? `Shard ${shard.id}` : 'Manager'}] ${message}`);
}
/**
* Connects this manager to the gateway.
* @private
*/
async connect() {
const invalidToken = new DJSError(WSCodes[4004]);
const {
url: gatewayURL,
shards: recommendedShards,
session_start_limit: sessionStartLimit,
} = await this.client.api.gateway.bot.get().catch(error => {
throw error.httpStatus === 401 ? invalidToken : error;
});
this.sessionStartLimit = sessionStartLimit;
const { total, remaining, reset_after } = sessionStartLimit;
this.debug(`Fetched Gateway Information
URL: ${gatewayURL}
Recommended Shards: ${recommendedShards}`);
this.debug(`Session Limit Information
Total: ${total}
Remaining: ${remaining}`);
this.gateway = `${gatewayURL}/`;
let { shards } = this.client.options;
if (shards === 'auto') {
this.debug(`Using the recommended shard count provided by Discord: ${recommendedShards}`);
this.totalShards = this.client.options.shardCount = recommendedShards;
shards = this.client.options.shards = Array.from({ length: recommendedShards }, (_, i) => i);
}
this.totalShards = shards.length;
this.debug(`Spawning shards: ${shards.join(', ')}`);
this.shardQueue = new Set(shards.map(id => new WebSocketShard(this, id)));
await this._handleSessionLimit(remaining, reset_after);
return this.createShards();
}
/**
* Handles the creation of a shard.
* @returns {Promise<boolean>}
* @private
*/
async createShards() {
// If we don't have any shards to handle, return
if (!this.shardQueue.size) return false;
const [shard] = this.shardQueue;
this.shardQueue.delete(shard);
if (!shard.eventsAttached) {
shard.on(ShardEvents.ALL_READY, unavailableGuilds => {
/**
* Emitted when a shard turns ready.
* @event Client#shardReady
* @param {number} id The shard ID that turned ready
* @param {?Set<string>} unavailableGuilds Set of unavailable guild IDs, if any
*/
this.client.emit(Events.SHARD_READY, shard.id, unavailableGuilds);
if (!this.shardQueue.size) this.reconnecting = false;
this.checkShardsReady();
});
shard.on(ShardEvents.CLOSE, event => {
if (event.code === 1000 ? this.destroyed : UNRECOVERABLE_CLOSE_CODES.includes(event.code)) {
/**
* Emitted when a shard's WebSocket disconnects and will no longer reconnect.
* @event Client#shardDisconnect
* @param {CloseEvent} event The WebSocket close event
* @param {number} id The shard ID that disconnected
*/
this.client.emit(Events.SHARD_DISCONNECT, event, shard.id);
this.debug(WSCodes[event.code], shard);
return;
}
if (UNRESUMABLE_CLOSE_CODES.includes(event.code)) {
// These event codes cannot be resumed
shard.sessionID = null;
}
/**
* Emitted when a shard is attempting to reconnect or re-identify.
* @event Client#shardReconnecting
* @param {number} id The shard ID that is attempting to reconnect
*/
this.client.emit(Events.SHARD_RECONNECTING, shard.id);
this.shardQueue.add(shard);
if (shard.sessionID) {
this.debug(`Session ID is present, attempting an immediate reconnect...`, shard);
this.reconnect(true);
} else {
shard.destroy({ reset: true, emit: false, log: false });
this.reconnect();
}
});
shard.on(ShardEvents.INVALID_SESSION, () => {
this.client.emit(Events.SHARD_RECONNECTING, shard.id);
});
shard.on(ShardEvents.DESTROYED, () => {
this.debug('Shard was destroyed but no WebSocket connection was present! Reconnecting...', shard);
this.client.emit(Events.SHARD_RECONNECTING, shard.id);
this.shardQueue.add(shard);
this.reconnect();
});
shard.eventsAttached = true;
}
this.shards.set(shard.id, shard);
try {
await shard.connect();
} catch (error) {
if (error && error.code && UNRECOVERABLE_CLOSE_CODES.includes(error.code)) {
throw new DJSError(WSCodes[error.code]);
// Undefined if session is invalid, error event for regular closes
} else if (!error || error.code) {
this.debug('Failed to connect to the gateway, requeueing...', shard);
this.shardQueue.add(shard);
} else {
throw error;
}
}
// If we have more shards, add a 5s delay
if (this.shardQueue.size) {
this.debug(`Shard Queue Size: ${this.shardQueue.size}; continuing in 5 seconds...`);
await Util.delayFor(5000);
await this._handleSessionLimit();
return this.createShards();
}
return true;
}
/**
* Handles reconnects for this manager.
* @param {boolean} [skipLimit=false] IF this reconnect should skip checking the session limit
* @private
* @returns {Promise<boolean>}
*/
async reconnect(skipLimit = false) {
if (this.reconnecting || this.status !== Status.READY) return false;
this.reconnecting = true;
try {
if (!skipLimit) await this._handleSessionLimit();
await this.createShards();
} catch (error) {
this.debug(`Couldn't reconnect or fetch information about the gateway. ${error}`);
if (error.httpStatus !== 401) {
this.debug(`Possible network error occurred. Retrying in 5s...`);
await Util.delayFor(5000);
this.reconnecting = false;
return this.reconnect();
}
// If we get an error at this point, it means we cannot reconnect anymore
if (this.client.listenerCount(Events.INVALIDATED)) {
/**
* Emitted when the client's session becomes invalidated.
* You are expected to handle closing the process gracefully and preventing a boot loop
* if you are listening to this event.
* @event Client#invalidated
*/
this.client.emit(Events.INVALIDATED);
// Destroy just the shards. This means you have to handle the cleanup yourself
this.destroy();
} else {
this.client.destroy();
}
} finally {
this.reconnecting = false;
}
return true;
}
/**
* Broadcasts a packet to every shard this manager handles.
* @param {Object} packet The packet to send
* @private
*/
broadcast(packet) {
for (const shard of this.shards.values()) shard.send(packet);
}
/**
* Destroys this manager and all its shards.
* @private
*/
destroy() {
if (this.destroyed) return;
this.debug(`Manager was destroyed. Called by:\n${new Error('MANAGER_DESTROYED').stack}`);
this.destroyed = true;
this.shardQueue.clear();
for (const shard of this.shards.values()) shard.destroy({ closeCode: 1000, reset: true, emit: false, log: false });
}
/**
* Handles the timeout required if we cannot identify anymore.
* @param {number} [remaining] The amount of remaining identify sessions that can be done today
* @param {number} [resetAfter] The amount of time in which the identify counter resets
* @private
*/
async _handleSessionLimit(remaining, resetAfter) {
if (typeof remaining === 'undefined' && typeof resetAfter === 'undefined') {
const { session_start_limit } = await this.client.api.gateway.bot.get();
this.sessionStartLimit = session_start_limit;
remaining = session_start_limit.remaining;
resetAfter = session_start_limit.reset_after;
this.debug(`Session Limit Information
Total: ${session_start_limit.total}
Remaining: ${remaining}`);
}
if (!remaining) {
this.debug(`Exceeded identify threshold. Will attempt a connection in ${resetAfter}ms`);
await Util.delayFor(resetAfter);
}
}
/**
* Processes a packet and queues it if this WebSocketManager is not ready.
* @param {Object} [packet] The packet to be handled
* @param {WebSocketShard} [shard] The shard that will handle this packet
* @returns {boolean}
* @private
*/
handlePacket(packet, shard) {
if (packet && this.status !== Status.READY) {
if (!BeforeReadyWhitelist.includes(packet.t)) {
this.packetQueue.push({ packet, shard });
return false;
}
}
if (this.packetQueue.length) {
const item = this.packetQueue.shift();
this.client.setImmediate(() => {
this.handlePacket(item.packet, item.shard);
});
}
if (packet && PacketHandlers[packet.t]) {
PacketHandlers[packet.t](this.client, packet, shard);
}
return true;
}
/**
* Checks whether the client is ready to be marked as ready.
* @private
*/
async checkShardsReady() {
if (this.status === Status.READY) return;
if (this.shards.size !== this.totalShards || this.shards.some(s => s.status !== Status.READY)) {
return;
}
this.status = Status.NEARLY;
if (this.client.options.fetchAllMembers) {
try {
const promises = this.client.guilds.cache.map(guild => {
if (guild.available) return guild.members.fetch();
// Return empty promise if guild is unavailable
return Promise.resolve();
});
await Promise.all(promises);
} catch (err) {
this.debug(`Failed to fetch all members before ready! ${err}\n${err.stack}`);
}
}
this.triggerClientReady();
}
/**
* Causes the client to be marked as ready and emits the ready event.
* @private
*/
triggerClientReady() {
this.status = Status.READY;
this.client.readyAt = new Date();
/**
* Emitted when the client becomes ready to start working.
* @event Client#ready
*/
this.client.emit(Events.CLIENT_READY);
this.handlePacket();
}
}
module.exports = WebSocketManager;