mirror of
https://github.com/discordjs/discord.js.git
synced 2026-03-09 16:13:31 +01:00
refactor(WebSocketManager): use /ws package internally (#9099)
This commit is contained in:
@@ -39,7 +39,6 @@ pnpm add discord.js
|
||||
### Optional packages
|
||||
|
||||
- [zlib-sync](https://www.npmjs.com/package/zlib-sync) for WebSocket data compression and inflation (`npm install zlib-sync`)
|
||||
- [erlpack](https://github.com/discord/erlpack) for significantly faster WebSocket data (de)serialisation (`npm install discord/erlpack`)
|
||||
- [bufferutil](https://www.npmjs.com/package/bufferutil) for a much faster WebSocket connection (`npm install bufferutil`)
|
||||
- [utf-8-validate](https://www.npmjs.com/package/utf-8-validate) in combination with `bufferutil` for much faster WebSocket processing (`npm install utf-8-validate`)
|
||||
- [@discordjs/voice](https://www.npmjs.com/package/@discordjs/voice) for interacting with the Discord Voice API (`npm install @discordjs/voice`)
|
||||
|
||||
@@ -55,6 +55,7 @@
|
||||
"@discordjs/formatters": "workspace:^",
|
||||
"@discordjs/rest": "workspace:^",
|
||||
"@discordjs/util": "workspace:^",
|
||||
"@discordjs/ws": "workspace:^",
|
||||
"@sapphire/snowflake": "^3.4.2",
|
||||
"@types/ws": "^8.5.4",
|
||||
"discord-api-types": "^0.37.41",
|
||||
|
||||
@@ -1,39 +0,0 @@
|
||||
'use strict';
|
||||
|
||||
let erlpack;
|
||||
const { Buffer } = require('node:buffer');
|
||||
|
||||
try {
|
||||
erlpack = require('erlpack');
|
||||
if (!erlpack.pack) erlpack = null;
|
||||
} catch {} // eslint-disable-line no-empty
|
||||
|
||||
exports.WebSocket = require('ws');
|
||||
|
||||
const ab = new TextDecoder();
|
||||
|
||||
exports.encoding = erlpack ? 'etf' : 'json';
|
||||
|
||||
exports.pack = erlpack ? erlpack.pack : JSON.stringify;
|
||||
|
||||
exports.unpack = (data, type) => {
|
||||
if (exports.encoding === 'json' || type === 'json') {
|
||||
if (typeof data !== 'string') {
|
||||
data = ab.decode(data);
|
||||
}
|
||||
return JSON.parse(data);
|
||||
}
|
||||
if (!Buffer.isBuffer(data)) data = Buffer.from(new Uint8Array(data));
|
||||
return erlpack.unpack(data);
|
||||
};
|
||||
|
||||
exports.create = (gateway, query = {}, ...args) => {
|
||||
const [g, q] = gateway.split('?');
|
||||
query.encoding = exports.encoding;
|
||||
query = new URLSearchParams(query);
|
||||
if (q) new URLSearchParams(q).forEach((v, k) => query.set(k, v));
|
||||
const ws = new exports.WebSocket(`${g}?${query}`, ...args);
|
||||
return ws;
|
||||
};
|
||||
|
||||
for (const state of ['CONNECTING', 'OPEN', 'CLOSING', 'CLOSED']) exports[state] = exports.WebSocket[state];
|
||||
@@ -1,10 +1,16 @@
|
||||
'use strict';
|
||||
|
||||
const EventEmitter = require('node:events');
|
||||
const process = require('node:process');
|
||||
const { setImmediate } = require('node:timers');
|
||||
const { setTimeout: sleep } = require('node:timers/promises');
|
||||
const { Collection } = require('@discordjs/collection');
|
||||
const { GatewayCloseCodes, GatewayDispatchEvents, Routes } = require('discord-api-types/v10');
|
||||
const {
|
||||
WebSocketManager: WSWebSocketManager,
|
||||
WebSocketShardEvents: WSWebSocketShardEvents,
|
||||
CompressionMethod,
|
||||
CloseCodes,
|
||||
} = require('@discordjs/ws');
|
||||
const { GatewayCloseCodes, GatewayDispatchEvents } = require('discord-api-types/v10');
|
||||
const WebSocketShard = require('./WebSocketShard');
|
||||
const PacketHandlers = require('./handlers');
|
||||
const { DiscordjsError, ErrorCodes } = require('../../errors');
|
||||
@@ -12,6 +18,12 @@ const Events = require('../../util/Events');
|
||||
const Status = require('../../util/Status');
|
||||
const WebSocketShardEvents = require('../../util/WebSocketShardEvents');
|
||||
|
||||
let zlib;
|
||||
|
||||
try {
|
||||
zlib = require('zlib-sync');
|
||||
} catch {} // eslint-disable-line no-empty
|
||||
|
||||
const BeforeReadyWhitelist = [
|
||||
GatewayDispatchEvents.Ready,
|
||||
GatewayDispatchEvents.Resumed,
|
||||
@@ -22,15 +34,17 @@ const BeforeReadyWhitelist = [
|
||||
GatewayDispatchEvents.GuildMemberRemove,
|
||||
];
|
||||
|
||||
const unrecoverableErrorCodeMap = {
|
||||
[GatewayCloseCodes.AuthenticationFailed]: ErrorCodes.TokenInvalid,
|
||||
[GatewayCloseCodes.InvalidShard]: ErrorCodes.ShardingInvalid,
|
||||
[GatewayCloseCodes.ShardingRequired]: ErrorCodes.ShardingRequired,
|
||||
[GatewayCloseCodes.InvalidIntents]: ErrorCodes.InvalidIntents,
|
||||
[GatewayCloseCodes.DisallowedIntents]: ErrorCodes.DisallowedIntents,
|
||||
};
|
||||
const WaitingForGuildEvents = [GatewayDispatchEvents.GuildCreate, GatewayDispatchEvents.GuildDelete];
|
||||
|
||||
const UNRESUMABLE_CLOSE_CODES = [1000, GatewayCloseCodes.AlreadyAuthenticated, GatewayCloseCodes.InvalidSeq];
|
||||
const UNRESUMABLE_CLOSE_CODES = [
|
||||
CloseCodes.Normal,
|
||||
GatewayCloseCodes.AlreadyAuthenticated,
|
||||
GatewayCloseCodes.InvalidSeq,
|
||||
];
|
||||
|
||||
const reasonIsDeprecated = 'the reason property is deprecated, use the code property to determine the reason';
|
||||
let deprecationEmittedForInvalidSessionEvent = false;
|
||||
let deprecationEmittedForDestroyedEvent = false;
|
||||
|
||||
/**
|
||||
* The WebSocket manager for this client.
|
||||
@@ -56,27 +70,12 @@ class WebSocketManager extends EventEmitter {
|
||||
*/
|
||||
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[]}
|
||||
@@ -99,11 +98,11 @@ class WebSocketManager extends EventEmitter {
|
||||
this.destroyed = false;
|
||||
|
||||
/**
|
||||
* If this manager is currently reconnecting one or multiple shards
|
||||
* @type {boolean}
|
||||
* The internal WebSocketManager from `@discordjs/ws`.
|
||||
* @type {WSWebSocketManager}
|
||||
* @private
|
||||
*/
|
||||
this.reconnecting = false;
|
||||
this._ws = null;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -119,11 +118,14 @@ class WebSocketManager extends EventEmitter {
|
||||
/**
|
||||
* Emits a debug message.
|
||||
* @param {string} message The debug message
|
||||
* @param {?WebSocketShard} [shard] The shard that emitted this message, if any
|
||||
* @param {?number} [shardId] The id of the shard that emitted this message, if any
|
||||
* @private
|
||||
*/
|
||||
debug(message, shard) {
|
||||
this.client.emit(Events.Debug, `[WS => ${shard ? `Shard ${shard.id}` : 'Manager'}] ${message}`);
|
||||
debug(message, shardId) {
|
||||
this.client.emit(
|
||||
Events.Debug,
|
||||
`[WS => ${typeof shardId === 'number' ? `Shard ${shardId}` : 'Manager'}] ${message}`,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -132,11 +134,37 @@ class WebSocketManager extends EventEmitter {
|
||||
*/
|
||||
async connect() {
|
||||
const invalidToken = new DiscordjsError(ErrorCodes.TokenInvalid);
|
||||
const { shards, shardCount, intents, ws } = this.client.options;
|
||||
if (this._ws && this._ws.options.token !== this.client.token) {
|
||||
await this._ws.destroy({ code: CloseCodes.Normal, reason: 'Login with differing token requested' });
|
||||
this._ws = null;
|
||||
}
|
||||
if (!this._ws) {
|
||||
const wsOptions = {
|
||||
intents: intents.bitfield,
|
||||
rest: this.client.rest,
|
||||
token: this.client.token,
|
||||
largeThreshold: ws.large_threshold,
|
||||
version: ws.version,
|
||||
shardIds: shards === 'auto' ? null : shards,
|
||||
shardCount: shards === 'auto' ? null : shardCount,
|
||||
initialPresence: ws.presence,
|
||||
retrieveSessionInfo: shardId => this.shards.get(shardId).sessionInfo,
|
||||
updateSessionInfo: (shardId, sessionInfo) => {
|
||||
this.shards.get(shardId).sessionInfo = sessionInfo;
|
||||
},
|
||||
compression: zlib ? CompressionMethod.ZlibStream : null,
|
||||
};
|
||||
if (ws.buildStrategy) wsOptions.buildStrategy = ws.buildStrategy;
|
||||
this._ws = new WSWebSocketManager(wsOptions);
|
||||
this.attachEvents();
|
||||
}
|
||||
|
||||
const {
|
||||
url: gatewayURL,
|
||||
shards: recommendedShards,
|
||||
session_start_limit: sessionStartLimit,
|
||||
} = await this.client.rest.get(Routes.gatewayBot()).catch(error => {
|
||||
} = await this._ws.fetchGatewayInformation().catch(error => {
|
||||
throw error.status === 401 ? invalidToken : error;
|
||||
});
|
||||
|
||||
@@ -152,156 +180,130 @@ class WebSocketManager extends EventEmitter {
|
||||
|
||||
this.gateway = `${gatewayURL}/`;
|
||||
|
||||
let { shards } = this.client.options;
|
||||
this.client.options.shardCount = await this._ws.getShardCount();
|
||||
this.client.options.shards = await this._ws.getShardIds();
|
||||
this.totalShards = this.client.options.shards.length;
|
||||
for (const id of this.client.options.shards) {
|
||||
if (!this.shards.has(id)) {
|
||||
const shard = new WebSocketShard(this, id);
|
||||
this.shards.set(id, shard);
|
||||
|
||||
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)));
|
||||
|
||||
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(WebSocketShardEvents.AllReady, unavailableGuilds => {
|
||||
/**
|
||||
* Emitted when a shard turns ready.
|
||||
* @event Client#shardReady
|
||||
* @param {number} id The shard id that turned ready
|
||||
* @param {?Set<Snowflake>} unavailableGuilds Set of unavailable guild ids, if any
|
||||
*/
|
||||
this.client.emit(Events.ShardReady, shard.id, unavailableGuilds);
|
||||
|
||||
if (!this.shardQueue.size) this.reconnecting = false;
|
||||
this.checkShardsReady();
|
||||
});
|
||||
|
||||
shard.on(WebSocketShardEvents.Close, event => {
|
||||
if (event.code === 1_000 ? this.destroyed : event.code in unrecoverableErrorCodeMap) {
|
||||
shard.on(WebSocketShardEvents.AllReady, unavailableGuilds => {
|
||||
/**
|
||||
* 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
|
||||
* Emitted when a shard turns ready.
|
||||
* @event Client#shardReady
|
||||
* @param {number} id The shard id that turned ready
|
||||
* @param {?Set<Snowflake>} unavailableGuilds Set of unavailable guild ids, if any
|
||||
*/
|
||||
this.client.emit(Events.ShardDisconnect, event, shard.id);
|
||||
this.debug(GatewayCloseCodes[event.code], shard);
|
||||
return;
|
||||
}
|
||||
this.client.emit(Events.ShardReady, shard.id, unavailableGuilds);
|
||||
|
||||
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.ShardReconnecting, shard.id);
|
||||
|
||||
this.shardQueue.add(shard);
|
||||
|
||||
if (shard.sessionId) this.debug(`Session id is present, attempting an immediate reconnect...`, shard);
|
||||
this.reconnect();
|
||||
});
|
||||
|
||||
shard.on(WebSocketShardEvents.InvalidSession, () => {
|
||||
this.client.emit(Events.ShardReconnecting, shard.id);
|
||||
});
|
||||
|
||||
shard.on(WebSocketShardEvents.Destroyed, () => {
|
||||
this.debug('Shard was destroyed but no WebSocket connection was present! Reconnecting...', shard);
|
||||
|
||||
this.client.emit(Events.ShardReconnecting, 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?.code && error.code in unrecoverableErrorCodeMap) {
|
||||
throw new DiscordjsError(unrecoverableErrorCodeMap[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;
|
||||
this.checkShardsReady();
|
||||
});
|
||||
shard.status = Status.Connecting;
|
||||
}
|
||||
}
|
||||
// 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 sleep(5_000);
|
||||
return this.createShards();
|
||||
}
|
||||
|
||||
return true;
|
||||
await this._ws.connect();
|
||||
|
||||
this.shards.forEach(shard => {
|
||||
if (shard.listenerCount(WebSocketShardEvents.InvalidSession) > 0 && !deprecationEmittedForInvalidSessionEvent) {
|
||||
process.emitWarning(
|
||||
'The WebSocketShard#invalidSession event is deprecated and will never emit.',
|
||||
'DeprecationWarning',
|
||||
);
|
||||
|
||||
deprecationEmittedForInvalidSessionEvent = true;
|
||||
}
|
||||
if (shard.listenerCount(WebSocketShardEvents.Destroyed) > 0 && !deprecationEmittedForDestroyedEvent) {
|
||||
process.emitWarning(
|
||||
'The WebSocketShard#destroyed event is deprecated and will never emit.',
|
||||
'DeprecationWarning',
|
||||
);
|
||||
|
||||
deprecationEmittedForDestroyedEvent = true;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Handles reconnects for this manager.
|
||||
* Attaches event handlers to the internal WebSocketShardManager from `@discordjs/ws`.
|
||||
* @private
|
||||
* @returns {Promise<boolean>}
|
||||
*/
|
||||
async reconnect() {
|
||||
if (this.reconnecting || this.status !== Status.Ready) return false;
|
||||
this.reconnecting = true;
|
||||
try {
|
||||
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 sleep(5_000);
|
||||
this.reconnecting = false;
|
||||
return this.reconnect();
|
||||
attachEvents() {
|
||||
this._ws.on(WSWebSocketShardEvents.Debug, ({ message, shardId }) => this.debug(message, shardId));
|
||||
this._ws.on(WSWebSocketShardEvents.Dispatch, ({ data, shardId }) => {
|
||||
this.client.emit(Events.Raw, data, shardId);
|
||||
const shard = this.shards.get(shardId);
|
||||
this.handlePacket(data, shard);
|
||||
if (shard.status === Status.WaitingForGuilds && WaitingForGuildEvents.includes(data.t)) {
|
||||
shard.gotGuild(data.d.id);
|
||||
}
|
||||
// If we get an error at this point, it means we cannot reconnect anymore
|
||||
if (this.client.listenerCount(Events.Invalidated)) {
|
||||
});
|
||||
|
||||
this._ws.on(WSWebSocketShardEvents.Ready, ({ data, shardId }) => {
|
||||
this.shards.get(shardId).onReadyPacket(data);
|
||||
});
|
||||
|
||||
this._ws.on(WSWebSocketShardEvents.Closed, ({ code, shardId }) => {
|
||||
const shard = this.shards.get(shardId);
|
||||
shard.emit(WebSocketShardEvents.Close, { code, reason: reasonIsDeprecated, wasClean: true });
|
||||
if (UNRESUMABLE_CLOSE_CODES.includes(code) && this.destroyed) {
|
||||
shard.status = Status.Disconnected;
|
||||
/**
|
||||
* 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
|
||||
* 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.Invalidated);
|
||||
// Destroy just the shards. This means you have to handle the cleanup yourself
|
||||
this.destroy();
|
||||
} else {
|
||||
this.client.destroy();
|
||||
this.client.emit(Events.ShardDisconnect, { code, reason: reasonIsDeprecated, wasClean: true }, shardId);
|
||||
this.debug(GatewayCloseCodes[code], shardId);
|
||||
return;
|
||||
}
|
||||
} finally {
|
||||
this.reconnecting = false;
|
||||
}
|
||||
return true;
|
||||
|
||||
this.shards.get(shardId).status = Status.Connecting;
|
||||
/**
|
||||
* 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.ShardReconnecting, shardId);
|
||||
});
|
||||
this._ws.on(WSWebSocketShardEvents.Hello, ({ shardId }) => {
|
||||
const shard = this.shards.get(shardId);
|
||||
if (shard.sessionInfo) {
|
||||
shard.closeSequence = shard.sessionInfo.sequence;
|
||||
shard.status = Status.Resuming;
|
||||
} else {
|
||||
shard.status = Status.Identifying;
|
||||
}
|
||||
});
|
||||
|
||||
this._ws.on(WSWebSocketShardEvents.Resumed, ({ shardId }) => {
|
||||
const shard = this.shards.get(shardId);
|
||||
shard.status = Status.Ready;
|
||||
/**
|
||||
* Emitted when the shard resumes successfully
|
||||
* @event WebSocketShard#resumed
|
||||
*/
|
||||
shard.emit(WebSocketShardEvents.Resumed);
|
||||
});
|
||||
|
||||
this._ws.on(WSWebSocketShardEvents.HeartbeatComplete, ({ heartbeatAt, latency, shardId }) => {
|
||||
this.debug(`Heartbeat acknowledged, latency of ${latency}ms.`, shardId);
|
||||
const shard = this.shards.get(shardId);
|
||||
shard.lastPingTimestamp = heartbeatAt;
|
||||
shard.ping = latency;
|
||||
});
|
||||
|
||||
this._ws.on(WSWebSocketShardEvents.Error, err => {
|
||||
/**
|
||||
* Emitted whenever a shard's WebSocket encounters a connection error.
|
||||
* @event Client#shardError
|
||||
* @param {Error} error The encountered error
|
||||
* @param {number} shardId The shard that encountered this error
|
||||
*/
|
||||
this.client.emit(Events.ShardError, err, err.shardId);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -310,7 +312,7 @@ class WebSocketManager extends EventEmitter {
|
||||
* @private
|
||||
*/
|
||||
broadcast(packet) {
|
||||
for (const shard of this.shards.values()) shard.send(packet);
|
||||
for (const shardId of this.shards.keys()) this._ws.send(shardId, packet);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -322,8 +324,7 @@ class WebSocketManager extends EventEmitter {
|
||||
// TODO: Make a util for getting a stack
|
||||
this.debug(`Manager was destroyed. Called by:\n${new Error().stack}`);
|
||||
this.destroyed = true;
|
||||
this.shardQueue.clear();
|
||||
for (const shard of this.shards.values()) shard.destroy({ closeCode: 1_000, reset: true, emit: false, log: false });
|
||||
this._ws.destroy({ code: CloseCodes.Normal });
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1,22 +1,13 @@
|
||||
'use strict';
|
||||
|
||||
const EventEmitter = require('node:events');
|
||||
const { setTimeout, setInterval, clearTimeout, clearInterval } = require('node:timers');
|
||||
const { GatewayDispatchEvents, GatewayIntentBits, GatewayOpcodes } = require('discord-api-types/v10');
|
||||
const WebSocket = require('../../WebSocket');
|
||||
const Events = require('../../util/Events');
|
||||
const process = require('node:process');
|
||||
const { setTimeout, clearTimeout } = require('node:timers');
|
||||
const { GatewayIntentBits } = require('discord-api-types/v10');
|
||||
const Status = require('../../util/Status');
|
||||
const WebSocketShardEvents = require('../../util/WebSocketShardEvents');
|
||||
|
||||
const STATUS_KEYS = Object.keys(Status);
|
||||
const CONNECTION_STATE = Object.keys(WebSocket.WebSocket);
|
||||
|
||||
let zlib;
|
||||
|
||||
try {
|
||||
zlib = require('zlib-sync');
|
||||
} catch {} // eslint-disable-line no-empty
|
||||
|
||||
let deprecationEmittedForImportant = false;
|
||||
/**
|
||||
* Represents a Shard's WebSocket connection
|
||||
* @extends {EventEmitter}
|
||||
@@ -43,13 +34,6 @@ class WebSocketShard extends EventEmitter {
|
||||
*/
|
||||
this.status = Status.Idle;
|
||||
|
||||
/**
|
||||
* The current sequence of the shard
|
||||
* @type {number}
|
||||
* @private
|
||||
*/
|
||||
this.sequence = -1;
|
||||
|
||||
/**
|
||||
* The sequence of the shard after close
|
||||
* @type {number}
|
||||
@@ -57,20 +41,6 @@ class WebSocketShard extends EventEmitter {
|
||||
*/
|
||||
this.closeSequence = 0;
|
||||
|
||||
/**
|
||||
* The current session id of the shard
|
||||
* @type {?string}
|
||||
* @private
|
||||
*/
|
||||
this.sessionId = null;
|
||||
|
||||
/**
|
||||
* The resume url for this shard
|
||||
* @type {?string}
|
||||
* @private
|
||||
*/
|
||||
this.resumeURL = null;
|
||||
|
||||
/**
|
||||
* The previous heartbeat ping of the shard
|
||||
* @type {number}
|
||||
@@ -83,81 +53,6 @@ class WebSocketShard extends EventEmitter {
|
||||
*/
|
||||
this.lastPingTimestamp = -1;
|
||||
|
||||
/**
|
||||
* If we received a heartbeat ack back. Used to identify zombie connections
|
||||
* @type {boolean}
|
||||
* @private
|
||||
*/
|
||||
this.lastHeartbeatAcked = true;
|
||||
|
||||
/**
|
||||
* Used to prevent calling {@link WebSocketShard#event:close} twice while closing or terminating the WebSocket.
|
||||
* @type {boolean}
|
||||
* @private
|
||||
*/
|
||||
this.closeEmitted = false;
|
||||
|
||||
/**
|
||||
* Contains the rate limit queue and metadata
|
||||
* @name WebSocketShard#ratelimit
|
||||
* @type {Object}
|
||||
* @private
|
||||
*/
|
||||
Object.defineProperty(this, 'ratelimit', {
|
||||
value: {
|
||||
queue: [],
|
||||
total: 120,
|
||||
remaining: 120,
|
||||
time: 60e3,
|
||||
timer: null,
|
||||
},
|
||||
});
|
||||
|
||||
/**
|
||||
* The WebSocket connection for the current shard
|
||||
* @name WebSocketShard#connection
|
||||
* @type {?WebSocket}
|
||||
* @private
|
||||
*/
|
||||
Object.defineProperty(this, 'connection', { value: null, writable: true });
|
||||
|
||||
/**
|
||||
* @external Inflate
|
||||
* @see {@link https://www.npmjs.com/package/zlib-sync}
|
||||
*/
|
||||
|
||||
/**
|
||||
* The compression to use
|
||||
* @name WebSocketShard#inflate
|
||||
* @type {?Inflate}
|
||||
* @private
|
||||
*/
|
||||
Object.defineProperty(this, 'inflate', { value: null, writable: true });
|
||||
|
||||
/**
|
||||
* The HELLO timeout
|
||||
* @name WebSocketShard#helloTimeout
|
||||
* @type {?NodeJS.Timeout}
|
||||
* @private
|
||||
*/
|
||||
Object.defineProperty(this, 'helloTimeout', { value: null, writable: true });
|
||||
|
||||
/**
|
||||
* The WebSocket timeout.
|
||||
* @name WebSocketShard#wsCloseTimeout
|
||||
* @type {?NodeJS.Timeout}
|
||||
* @private
|
||||
*/
|
||||
Object.defineProperty(this, 'wsCloseTimeout', { value: null, writable: true });
|
||||
|
||||
/**
|
||||
* If the manager attached its event handlers on the shard
|
||||
* @name WebSocketShard#eventsAttached
|
||||
* @type {boolean}
|
||||
* @private
|
||||
*/
|
||||
Object.defineProperty(this, 'eventsAttached', { value: false, writable: true });
|
||||
|
||||
/**
|
||||
* A set of guild ids this shard expects to receive
|
||||
* @name WebSocketShard#expectedGuilds
|
||||
@@ -175,12 +70,17 @@ class WebSocketShard extends EventEmitter {
|
||||
Object.defineProperty(this, 'readyTimeout', { value: null, writable: true });
|
||||
|
||||
/**
|
||||
* Time when the WebSocket connection was opened
|
||||
* @name WebSocketShard#connectedAt
|
||||
* @type {number}
|
||||
* @external SessionInfo
|
||||
* @see {@link https://discord.js.org/#/docs/ws/main/typedef/SessionInfo}
|
||||
*/
|
||||
|
||||
/**
|
||||
* The session info used by `@discordjs/ws` package.
|
||||
* @name WebSocketShard#sessionInfo
|
||||
* @type {?SessionInfo}
|
||||
* @private
|
||||
*/
|
||||
Object.defineProperty(this, 'connectedAt', { value: 0, writable: true });
|
||||
Object.defineProperty(this, 'sessionInfo', { value: null, writable: true });
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -189,161 +89,7 @@ class WebSocketShard extends EventEmitter {
|
||||
* @private
|
||||
*/
|
||||
debug(message) {
|
||||
this.manager.debug(message, this);
|
||||
}
|
||||
|
||||
/**
|
||||
* Connects the shard to the gateway.
|
||||
* @private
|
||||
* @returns {Promise<void>} A promise that will resolve if the shard turns ready successfully,
|
||||
* or reject if we couldn't connect
|
||||
*/
|
||||
connect() {
|
||||
const { client } = this.manager;
|
||||
|
||||
if (this.connection?.readyState === WebSocket.OPEN && this.status === Status.Ready) {
|
||||
return Promise.resolve();
|
||||
}
|
||||
|
||||
const gateway = this.resumeURL ?? this.manager.gateway;
|
||||
|
||||
return new Promise((resolve, reject) => {
|
||||
const cleanup = () => {
|
||||
this.removeListener(WebSocketShardEvents.Close, onClose);
|
||||
this.removeListener(WebSocketShardEvents.Ready, onReady);
|
||||
this.removeListener(WebSocketShardEvents.Resumed, onResumed);
|
||||
this.removeListener(WebSocketShardEvents.InvalidSession, onInvalidOrDestroyed);
|
||||
this.removeListener(WebSocketShardEvents.Destroyed, onInvalidOrDestroyed);
|
||||
};
|
||||
|
||||
const onReady = () => {
|
||||
cleanup();
|
||||
resolve();
|
||||
};
|
||||
|
||||
const onResumed = () => {
|
||||
cleanup();
|
||||
resolve();
|
||||
};
|
||||
|
||||
const onClose = event => {
|
||||
cleanup();
|
||||
reject(event);
|
||||
};
|
||||
|
||||
const onInvalidOrDestroyed = () => {
|
||||
cleanup();
|
||||
// eslint-disable-next-line prefer-promise-reject-errors
|
||||
reject();
|
||||
};
|
||||
|
||||
this.once(WebSocketShardEvents.Ready, onReady);
|
||||
this.once(WebSocketShardEvents.Resumed, onResumed);
|
||||
this.once(WebSocketShardEvents.Close, onClose);
|
||||
this.once(WebSocketShardEvents.InvalidSession, onInvalidOrDestroyed);
|
||||
this.once(WebSocketShardEvents.Destroyed, onInvalidOrDestroyed);
|
||||
|
||||
if (this.connection?.readyState === WebSocket.OPEN) {
|
||||
this.debug('An open connection was found, attempting an immediate identify.');
|
||||
this.identify();
|
||||
return;
|
||||
}
|
||||
|
||||
if (this.connection) {
|
||||
this.debug(`A connection object was found. Cleaning up before continuing.
|
||||
State: ${CONNECTION_STATE[this.connection.readyState]}`);
|
||||
this.destroy({ emit: false });
|
||||
}
|
||||
|
||||
const wsQuery = { v: client.options.ws.version };
|
||||
|
||||
if (zlib) {
|
||||
this.inflate = new zlib.Inflate({
|
||||
chunkSize: 65535,
|
||||
flush: zlib.Z_SYNC_FLUSH,
|
||||
to: WebSocket.encoding === 'json' ? 'string' : '',
|
||||
});
|
||||
wsQuery.compress = 'zlib-stream';
|
||||
}
|
||||
|
||||
this.debug(
|
||||
`[CONNECT]
|
||||
Gateway : ${gateway}
|
||||
Version : ${client.options.ws.version}
|
||||
Encoding : ${WebSocket.encoding}
|
||||
Compression: ${zlib ? 'zlib-stream' : 'none'}`,
|
||||
);
|
||||
|
||||
this.status = this.status === Status.Disconnected ? Status.Reconnecting : Status.Connecting;
|
||||
this.setHelloTimeout();
|
||||
|
||||
this.connectedAt = Date.now();
|
||||
|
||||
// Adding a handshake timeout to just make sure no zombie connection appears.
|
||||
const ws = (this.connection = WebSocket.create(gateway, wsQuery, { handshakeTimeout: 30_000 }));
|
||||
ws.onopen = this.onOpen.bind(this);
|
||||
ws.onmessage = this.onMessage.bind(this);
|
||||
ws.onerror = this.onError.bind(this);
|
||||
ws.onclose = this.onClose.bind(this);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Called whenever a connection is opened to the gateway.
|
||||
* @private
|
||||
*/
|
||||
onOpen() {
|
||||
this.debug(`[CONNECTED] Took ${Date.now() - this.connectedAt}ms`);
|
||||
this.status = Status.Nearly;
|
||||
}
|
||||
|
||||
/**
|
||||
* Called whenever a message is received.
|
||||
* @param {MessageEvent} event Event received
|
||||
* @private
|
||||
*/
|
||||
onMessage({ data }) {
|
||||
let raw;
|
||||
if (data instanceof ArrayBuffer) data = new Uint8Array(data);
|
||||
if (zlib) {
|
||||
const l = data.length;
|
||||
const flush =
|
||||
l >= 4 && data[l - 4] === 0x00 && data[l - 3] === 0x00 && data[l - 2] === 0xff && data[l - 1] === 0xff;
|
||||
|
||||
this.inflate.push(data, flush && zlib.Z_SYNC_FLUSH);
|
||||
if (!flush) return;
|
||||
raw = this.inflate.result;
|
||||
} else {
|
||||
raw = data;
|
||||
}
|
||||
let packet;
|
||||
try {
|
||||
packet = WebSocket.unpack(raw);
|
||||
} catch (err) {
|
||||
this.manager.client.emit(Events.ShardError, err, this.id);
|
||||
return;
|
||||
}
|
||||
this.manager.client.emit(Events.Raw, packet, this.id);
|
||||
if (packet.op === GatewayOpcodes.Dispatch) this.manager.emit(packet.t, packet.d, this.id);
|
||||
this.onPacket(packet);
|
||||
}
|
||||
|
||||
/**
|
||||
* Called whenever an error occurs with the WebSocket.
|
||||
* @param {ErrorEvent} event The error that occurred
|
||||
* @private
|
||||
*/
|
||||
onError(event) {
|
||||
const error = event?.error ?? event;
|
||||
if (!error) return;
|
||||
|
||||
/**
|
||||
* Emitted whenever a shard's WebSocket encounters a connection error.
|
||||
* @event Client#shardError
|
||||
* @param {Error} error The encountered error
|
||||
* @param {number} shardId The shard that encountered this error
|
||||
*/
|
||||
this.manager.client.emit(Events.ShardError, error, this.id);
|
||||
this.manager.debug(message, this.id);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -351,43 +97,11 @@ class WebSocketShard extends EventEmitter {
|
||||
* @see {@link https://developer.mozilla.org/en-US/docs/Web/API/CloseEvent}
|
||||
*/
|
||||
|
||||
/**
|
||||
* @external ErrorEvent
|
||||
* @see {@link https://developer.mozilla.org/en-US/docs/Web/API/ErrorEvent}
|
||||
*/
|
||||
|
||||
/**
|
||||
* @external MessageEvent
|
||||
* @see {@link https://developer.mozilla.org/en-US/docs/Web/API/MessageEvent}
|
||||
*/
|
||||
|
||||
/**
|
||||
* Called whenever a connection to the gateway is closed.
|
||||
* @param {CloseEvent} event Close event that was received
|
||||
* @private
|
||||
*/
|
||||
onClose(event) {
|
||||
this.closeEmitted = true;
|
||||
if (this.sequence !== -1) this.closeSequence = this.sequence;
|
||||
this.sequence = -1;
|
||||
this.setHeartbeatTimer(-1);
|
||||
this.setHelloTimeout(-1);
|
||||
// Clearing the WebSocket close timeout as close was emitted.
|
||||
this.setWsCloseTimeout(-1);
|
||||
// If we still have a connection object, clean up its listeners
|
||||
if (this.connection) {
|
||||
this._cleanupConnection();
|
||||
// Having this after _cleanupConnection to just clean up the connection and not listen to ws.onclose
|
||||
this.destroy({ reset: !this.sessionId, emit: false, log: false });
|
||||
}
|
||||
this.status = Status.Disconnected;
|
||||
this.emitClose(event);
|
||||
}
|
||||
|
||||
/**
|
||||
* This method is responsible to emit close event for this shard.
|
||||
* This method helps the shard reconnect.
|
||||
* @param {CloseEvent} [event] Close event that was received
|
||||
* @deprecated
|
||||
*/
|
||||
emitClose(
|
||||
event = {
|
||||
@@ -410,93 +124,35 @@ class WebSocketShard extends EventEmitter {
|
||||
}
|
||||
|
||||
/**
|
||||
* Called whenever a packet is received.
|
||||
* Called when the shard receives the READY payload.
|
||||
* @param {Object} packet The received packet
|
||||
* @private
|
||||
*/
|
||||
onPacket(packet) {
|
||||
onReadyPacket(packet) {
|
||||
if (!packet) {
|
||||
this.debug(`Received broken packet: '${packet}'.`);
|
||||
return;
|
||||
}
|
||||
|
||||
switch (packet.t) {
|
||||
case GatewayDispatchEvents.Ready:
|
||||
/**
|
||||
* Emitted when the shard receives the READY payload and is now waiting for guilds
|
||||
* @event WebSocketShard#ready
|
||||
*/
|
||||
this.emit(WebSocketShardEvents.Ready);
|
||||
/**
|
||||
* Emitted when the shard receives the READY payload and is now waiting for guilds
|
||||
* @event WebSocketShard#ready
|
||||
*/
|
||||
this.emit(WebSocketShardEvents.Ready);
|
||||
|
||||
this.sessionId = packet.d.session_id;
|
||||
this.resumeURL = packet.d.resume_gateway_url;
|
||||
this.expectedGuilds = new Set(packet.d.guilds.map(d => d.id));
|
||||
this.status = Status.WaitingForGuilds;
|
||||
this.debug(`[READY] Session ${this.sessionId} | Resume url ${this.resumeURL}.`);
|
||||
this.lastHeartbeatAcked = true;
|
||||
this.sendHeartbeat('ReadyHeartbeat');
|
||||
break;
|
||||
case GatewayDispatchEvents.Resumed: {
|
||||
/**
|
||||
* Emitted when the shard resumes successfully
|
||||
* @event WebSocketShard#resumed
|
||||
*/
|
||||
this.emit(WebSocketShardEvents.Resumed);
|
||||
this.expectedGuilds = new Set(packet.guilds.map(d => d.id));
|
||||
this.status = Status.WaitingForGuilds;
|
||||
}
|
||||
|
||||
this.status = Status.Ready;
|
||||
const replayed = packet.s - this.closeSequence;
|
||||
this.debug(`[RESUMED] Session ${this.sessionId} | Replayed ${replayed} events.`);
|
||||
this.lastHeartbeatAcked = true;
|
||||
this.sendHeartbeat('ResumeHeartbeat');
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (packet.s > this.sequence) this.sequence = packet.s;
|
||||
|
||||
switch (packet.op) {
|
||||
case GatewayOpcodes.Hello:
|
||||
this.setHelloTimeout(-1);
|
||||
this.setHeartbeatTimer(packet.d.heartbeat_interval);
|
||||
this.identify();
|
||||
break;
|
||||
case GatewayOpcodes.Reconnect:
|
||||
this.debug('[RECONNECT] Discord asked us to reconnect');
|
||||
this.destroy({ closeCode: 4_000 });
|
||||
break;
|
||||
case GatewayOpcodes.InvalidSession:
|
||||
this.debug(`[INVALID SESSION] Resumable: ${packet.d}.`);
|
||||
// If we can resume the session, do so immediately
|
||||
if (packet.d) {
|
||||
this.identifyResume();
|
||||
return;
|
||||
}
|
||||
// Reset the sequence
|
||||
this.sequence = -1;
|
||||
// Reset the session id as it's invalid
|
||||
this.sessionId = null;
|
||||
// Set the status to reconnecting
|
||||
this.status = Status.Reconnecting;
|
||||
// Finally, emit the INVALID_SESSION event
|
||||
/**
|
||||
* Emitted when the session has been invalidated.
|
||||
* @event WebSocketShard#invalidSession
|
||||
*/
|
||||
this.emit(WebSocketShardEvents.InvalidSession);
|
||||
break;
|
||||
case GatewayOpcodes.HeartbeatAck:
|
||||
this.ackHeartbeat();
|
||||
break;
|
||||
case GatewayOpcodes.Heartbeat:
|
||||
this.sendHeartbeat('HeartbeatRequest', true);
|
||||
break;
|
||||
default:
|
||||
this.manager.handlePacket(packet, this);
|
||||
if (this.status === Status.WaitingForGuilds && packet.t === GatewayDispatchEvents.GuildCreate) {
|
||||
this.expectedGuilds.delete(packet.d.id);
|
||||
this.checkReady();
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Called when a GuildCreate or GuildDelete for this shard was sent after READY payload was received,
|
||||
* but before we emitted the READY event.
|
||||
* @param {Snowflake} guildId the id of the Guild sent in the payload
|
||||
* @private
|
||||
*/
|
||||
gotGuild(guildId) {
|
||||
this.expectedGuilds.delete(guildId);
|
||||
this.checkReady();
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -543,7 +199,6 @@ class WebSocketShard extends EventEmitter {
|
||||
);
|
||||
|
||||
this.readyTimeout = null;
|
||||
|
||||
this.status = Status.Ready;
|
||||
|
||||
this.emit(WebSocketShardEvents.AllReady, this.expectedGuilds);
|
||||
@@ -552,190 +207,6 @@ class WebSocketShard extends EventEmitter {
|
||||
).unref();
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the HELLO packet timeout.
|
||||
* @param {number} [time] If set to -1, it will clear the hello timeout
|
||||
* @private
|
||||
*/
|
||||
setHelloTimeout(time) {
|
||||
if (time === -1) {
|
||||
if (this.helloTimeout) {
|
||||
this.debug('Clearing the HELLO timeout.');
|
||||
clearTimeout(this.helloTimeout);
|
||||
this.helloTimeout = null;
|
||||
}
|
||||
return;
|
||||
}
|
||||
this.debug('Setting a HELLO timeout for 20s.');
|
||||
this.helloTimeout = setTimeout(() => {
|
||||
this.debug('Did not receive HELLO in time. Destroying and connecting again.');
|
||||
this.destroy({ reset: true, closeCode: 4009 });
|
||||
}, 20_000).unref();
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the WebSocket Close timeout.
|
||||
* This method is responsible for detecting any zombie connections if the WebSocket fails to close properly.
|
||||
* @param {number} [time] If set to -1, it will clear the timeout
|
||||
* @private
|
||||
*/
|
||||
setWsCloseTimeout(time) {
|
||||
if (this.wsCloseTimeout) {
|
||||
this.debug('[WebSocket] Clearing the close timeout.');
|
||||
clearTimeout(this.wsCloseTimeout);
|
||||
}
|
||||
if (time === -1) {
|
||||
this.wsCloseTimeout = null;
|
||||
return;
|
||||
}
|
||||
this.wsCloseTimeout = setTimeout(() => {
|
||||
this.setWsCloseTimeout(-1);
|
||||
this.debug(`[WebSocket] Close Emitted: ${this.closeEmitted}`);
|
||||
// Check if close event was emitted.
|
||||
if (this.closeEmitted) {
|
||||
this.debug(
|
||||
`[WebSocket] was closed. | WS State: ${CONNECTION_STATE[this.connection?.readyState ?? WebSocket.CLOSED]}`,
|
||||
);
|
||||
// Setting the variable false to check for zombie connections.
|
||||
this.closeEmitted = false;
|
||||
return;
|
||||
}
|
||||
|
||||
this.debug(
|
||||
`[WebSocket] did not close properly, assuming a zombie connection.\nEmitting close and reconnecting again.`,
|
||||
);
|
||||
|
||||
// Cleanup connection listeners
|
||||
if (this.connection) this._cleanupConnection();
|
||||
|
||||
this.emitClose({
|
||||
code: 4009,
|
||||
reason: 'Session time out.',
|
||||
wasClean: false,
|
||||
});
|
||||
}, time);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sets the heartbeat timer for this shard.
|
||||
* @param {number} time If -1, clears the interval, any other number sets an interval
|
||||
* @private
|
||||
*/
|
||||
setHeartbeatTimer(time) {
|
||||
if (time === -1) {
|
||||
if (this.heartbeatInterval) {
|
||||
this.debug('Clearing the heartbeat interval.');
|
||||
clearInterval(this.heartbeatInterval);
|
||||
this.heartbeatInterval = null;
|
||||
}
|
||||
return;
|
||||
}
|
||||
this.debug(`Setting a heartbeat interval for ${time}ms.`);
|
||||
// Sanity checks
|
||||
if (this.heartbeatInterval) clearInterval(this.heartbeatInterval);
|
||||
this.heartbeatInterval = setInterval(() => this.sendHeartbeat(), time).unref();
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a heartbeat to the WebSocket.
|
||||
* If this shard didn't receive a heartbeat last time, it will destroy it and reconnect
|
||||
* @param {string} [tag='HeartbeatTimer'] What caused this heartbeat to be sent
|
||||
* @param {boolean} [ignoreHeartbeatAck] If we should send the heartbeat forcefully.
|
||||
* @private
|
||||
*/
|
||||
sendHeartbeat(
|
||||
tag = 'HeartbeatTimer',
|
||||
ignoreHeartbeatAck = [Status.WaitingForGuilds, Status.Identifying, Status.Resuming].includes(this.status),
|
||||
) {
|
||||
if (ignoreHeartbeatAck && !this.lastHeartbeatAcked) {
|
||||
this.debug(`[${tag}] Didn't process heartbeat ack yet but we are still connected. Sending one now.`);
|
||||
} else if (!this.lastHeartbeatAcked) {
|
||||
this.debug(
|
||||
`[${tag}] Didn't receive a heartbeat ack last time, assuming zombie connection. Destroying and reconnecting.
|
||||
Status : ${STATUS_KEYS[this.status]}
|
||||
Sequence : ${this.sequence}
|
||||
Connection State: ${this.connection ? CONNECTION_STATE[this.connection.readyState] : 'No Connection??'}`,
|
||||
);
|
||||
|
||||
this.destroy({ reset: true, closeCode: 4009 });
|
||||
return;
|
||||
}
|
||||
|
||||
this.debug(`[${tag}] Sending a heartbeat.`);
|
||||
this.lastHeartbeatAcked = false;
|
||||
this.lastPingTimestamp = Date.now();
|
||||
this.send({ op: GatewayOpcodes.Heartbeat, d: this.sequence }, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Acknowledges a heartbeat.
|
||||
* @private
|
||||
*/
|
||||
ackHeartbeat() {
|
||||
this.lastHeartbeatAcked = true;
|
||||
const latency = Date.now() - this.lastPingTimestamp;
|
||||
this.debug(`Heartbeat acknowledged, latency of ${latency}ms.`);
|
||||
this.ping = latency;
|
||||
}
|
||||
|
||||
/**
|
||||
* Identifies the client on the connection.
|
||||
* @private
|
||||
* @returns {void}
|
||||
*/
|
||||
identify() {
|
||||
return this.sessionId ? this.identifyResume() : this.identifyNew();
|
||||
}
|
||||
|
||||
/**
|
||||
* Identifies as a new connection on the gateway.
|
||||
* @private
|
||||
*/
|
||||
identifyNew() {
|
||||
const { client } = this.manager;
|
||||
if (!client.token) {
|
||||
this.debug('[IDENTIFY] No token available to identify a new session.');
|
||||
return;
|
||||
}
|
||||
|
||||
this.status = Status.Identifying;
|
||||
|
||||
// Clone the identify payload and assign the token and shard info
|
||||
const d = {
|
||||
...client.options.ws,
|
||||
intents: client.options.intents.bitfield,
|
||||
token: client.token,
|
||||
shard: [this.id, Number(client.options.shardCount)],
|
||||
};
|
||||
|
||||
this.debug(`[IDENTIFY] Shard ${this.id}/${client.options.shardCount} with intents: ${d.intents}`);
|
||||
this.send({ op: GatewayOpcodes.Identify, d }, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Resumes a session on the gateway.
|
||||
* @private
|
||||
*/
|
||||
identifyResume() {
|
||||
if (!this.sessionId) {
|
||||
this.debug('[RESUME] No session id was present; identifying as a new session.');
|
||||
this.identifyNew();
|
||||
return;
|
||||
}
|
||||
|
||||
this.status = Status.Resuming;
|
||||
|
||||
this.debug(`[RESUME] Session ${this.sessionId}, sequence ${this.closeSequence}`);
|
||||
|
||||
const d = {
|
||||
token: this.manager.client.token,
|
||||
session_id: this.sessionId,
|
||||
seq: this.closeSequence,
|
||||
};
|
||||
|
||||
this.send({ op: GatewayOpcodes.Resume, d }, true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds a packet to the queue to be sent to the gateway.
|
||||
* <warn>If you use this method, make sure you understand that you need to provide
|
||||
@@ -743,161 +214,17 @@ class WebSocketShard extends EventEmitter {
|
||||
* Do not use this method if you don't know what you're doing.</warn>
|
||||
* @param {Object} data The full packet to send
|
||||
* @param {boolean} [important=false] If this packet should be added first in queue
|
||||
* <warn>This parameter is **deprecated**. Important payloads are determined by their opcode instead.</warn>
|
||||
*/
|
||||
send(data, important = false) {
|
||||
this.ratelimit.queue[important ? 'unshift' : 'push'](data);
|
||||
this.processQueue();
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends data, bypassing the queue.
|
||||
* @param {Object} data Packet to send
|
||||
* @returns {void}
|
||||
* @private
|
||||
*/
|
||||
_send(data) {
|
||||
if (this.connection?.readyState !== WebSocket.OPEN) {
|
||||
this.debug(
|
||||
`Tried to send packet '${JSON.stringify(data).replaceAll(
|
||||
this.manager.client.token,
|
||||
this.manager.client._censoredToken,
|
||||
)}' but no WebSocket is available!`,
|
||||
if (important && !deprecationEmittedForImportant) {
|
||||
process.emitWarning(
|
||||
'Sending important payloads explicitly is deprecated. They are determined by their opcode implicitly now.',
|
||||
'DeprecationWarning',
|
||||
);
|
||||
this.destroy({ closeCode: 4_000 });
|
||||
return;
|
||||
deprecationEmittedForImportant = true;
|
||||
}
|
||||
|
||||
this.connection.send(WebSocket.pack(data), err => {
|
||||
if (err) this.manager.client.emit(Events.ShardError, err, this.id);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Processes the current WebSocket queue.
|
||||
* @returns {void}
|
||||
* @private
|
||||
*/
|
||||
processQueue() {
|
||||
if (this.ratelimit.remaining === 0) return;
|
||||
if (this.ratelimit.queue.length === 0) return;
|
||||
if (this.ratelimit.remaining === this.ratelimit.total) {
|
||||
this.ratelimit.timer = setTimeout(() => {
|
||||
this.ratelimit.remaining = this.ratelimit.total;
|
||||
this.processQueue();
|
||||
}, this.ratelimit.time).unref();
|
||||
}
|
||||
while (this.ratelimit.remaining > 0) {
|
||||
const item = this.ratelimit.queue.shift();
|
||||
if (!item) return;
|
||||
this._send(item);
|
||||
this.ratelimit.remaining--;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Destroys this shard and closes its WebSocket connection.
|
||||
* @param {Object} [options={ closeCode: 1000, reset: false, emit: true, log: true }] Options for destroying the shard
|
||||
* @private
|
||||
*/
|
||||
destroy({ closeCode = 1_000, reset = false, emit = true, log = true } = {}) {
|
||||
if (log) {
|
||||
this.debug(`[DESTROY]
|
||||
Close Code : ${closeCode}
|
||||
Reset : ${reset}
|
||||
Emit DESTROYED: ${emit}`);
|
||||
}
|
||||
|
||||
// Step 0: Remove all timers
|
||||
this.setHeartbeatTimer(-1);
|
||||
this.setHelloTimeout(-1);
|
||||
|
||||
this.debug(
|
||||
`[WebSocket] Destroy: Attempting to close the WebSocket. | WS State: ${
|
||||
CONNECTION_STATE[this.connection?.readyState ?? WebSocket.CLOSED]
|
||||
}`,
|
||||
);
|
||||
// Step 1: Close the WebSocket connection, if any, otherwise, emit DESTROYED
|
||||
if (this.connection) {
|
||||
// If the connection is currently opened, we will (hopefully) receive close
|
||||
if (this.connection.readyState === WebSocket.OPEN) {
|
||||
this.connection.close(closeCode);
|
||||
this.debug(`[WebSocket] Close: Tried closing. | WS State: ${CONNECTION_STATE[this.connection.readyState]}`);
|
||||
} else {
|
||||
// Connection is not OPEN
|
||||
this.debug(`WS State: ${CONNECTION_STATE[this.connection.readyState]}`);
|
||||
// Remove listeners from the connection
|
||||
this._cleanupConnection();
|
||||
// Attempt to close the connection just in case
|
||||
try {
|
||||
this.connection.close(closeCode);
|
||||
} catch (err) {
|
||||
this.debug(
|
||||
`[WebSocket] Close: Something went wrong while closing the WebSocket: ${
|
||||
err.message || err
|
||||
}. Forcefully terminating the connection | WS State: ${CONNECTION_STATE[this.connection.readyState]}`,
|
||||
);
|
||||
this.connection.terminate();
|
||||
}
|
||||
|
||||
// Emit the destroyed event if needed
|
||||
if (emit) this._emitDestroyed();
|
||||
}
|
||||
} else if (emit) {
|
||||
// We requested a destroy, but we had no connection. Emit destroyed
|
||||
this._emitDestroyed();
|
||||
}
|
||||
|
||||
this.debug(
|
||||
`[WebSocket] Adding a WebSocket close timeout to ensure a correct WS reconnect.
|
||||
Timeout: ${this.manager.client.options.closeTimeout}ms`,
|
||||
);
|
||||
this.setWsCloseTimeout(this.manager.client.options.closeTimeout);
|
||||
|
||||
// Step 2: Null the connection object
|
||||
this.connection = null;
|
||||
|
||||
// Step 3: Set the shard status to disconnected
|
||||
this.status = Status.Disconnected;
|
||||
|
||||
// Step 4: Cache the old sequence (use to attempt a resume)
|
||||
if (this.sequence !== -1) this.closeSequence = this.sequence;
|
||||
|
||||
// Step 5: Reset the sequence, resume url and session id if requested
|
||||
if (reset) {
|
||||
this.sequence = -1;
|
||||
this.sessionId = null;
|
||||
this.resumeURL = null;
|
||||
}
|
||||
|
||||
// Step 6: reset the rate limit data
|
||||
this.ratelimit.remaining = this.ratelimit.total;
|
||||
this.ratelimit.queue.length = 0;
|
||||
if (this.ratelimit.timer) {
|
||||
clearTimeout(this.ratelimit.timer);
|
||||
this.ratelimit.timer = null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Cleans up the WebSocket connection listeners.
|
||||
* @private
|
||||
*/
|
||||
_cleanupConnection() {
|
||||
this.connection.onopen = this.connection.onclose = this.connection.onmessage = null;
|
||||
this.connection.onerror = () => null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Emits the DESTROYED event on the shard
|
||||
* @private
|
||||
*/
|
||||
_emitDestroyed() {
|
||||
/**
|
||||
* Emitted when a shard is destroyed, but no WebSocket connection was present.
|
||||
* @private
|
||||
* @event WebSocketShard#destroyed
|
||||
*/
|
||||
this.emit(WebSocketShardEvents.Destroyed);
|
||||
this.manager._ws.send(this.id, data);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
const Events = require('../../../util/Events');
|
||||
|
||||
module.exports = (client, packet, shard) => {
|
||||
const replayed = shard.sequence - shard.closeSequence;
|
||||
const replayed = shard.sessionInfo.sequence - shard.closeSequence;
|
||||
/**
|
||||
* Emitted when a shard resumes successfully.
|
||||
* @event Client#shardResume
|
||||
|
||||
@@ -13,16 +13,23 @@
|
||||
* @property {'ApplicationCommandPermissionsTokenMissing'} ApplicationCommandPermissionsTokenMissing
|
||||
|
||||
* @property {'WSCloseRequested'} WSCloseRequested
|
||||
* <warn>This property is deprecated.</warn>
|
||||
* @property {'WSConnectionExists'} WSConnectionExists
|
||||
* <warn>This property is deprecated.</warn>
|
||||
* @property {'WSNotOpen'} WSNotOpen
|
||||
* <warn>This property is deprecated.</warn>
|
||||
* @property {'ManagerDestroyed'} ManagerDestroyed
|
||||
|
||||
* @property {'BitFieldInvalid'} BitFieldInvalid
|
||||
|
||||
* @property {'ShardingInvalid'} ShardingInvalid
|
||||
* <warn>This property is deprecated.</warn>
|
||||
* @property {'ShardingRequired'} ShardingRequired
|
||||
* <warn>This property is deprecated.</warn>
|
||||
* @property {'InvalidIntents'} InvalidIntents
|
||||
* <warn>This property is deprecated.</warn>
|
||||
* @property {'DisallowedIntents'} DisallowedIntents
|
||||
* <warn>This property is deprecated.</warn>
|
||||
* @property {'ShardingNoShards'} ShardingNoShards
|
||||
* @property {'ShardingInProcess'} ShardingInProcess
|
||||
* @property {'ShardingInvalidEvalBroadcast'} ShardingInvalidEvalBroadcast
|
||||
|
||||
@@ -204,11 +204,10 @@ exports.WidgetMember = require('./structures/WidgetMember');
|
||||
exports.WelcomeChannel = require('./structures/WelcomeChannel');
|
||||
exports.WelcomeScreen = require('./structures/WelcomeScreen');
|
||||
|
||||
exports.WebSocket = require('./WebSocket');
|
||||
|
||||
// External
|
||||
__exportStar(require('discord-api-types/v10'), exports);
|
||||
__exportStar(require('@discordjs/builders'), exports);
|
||||
__exportStar(require('@discordjs/formatters'), exports);
|
||||
__exportStar(require('@discordjs/rest'), exports);
|
||||
__exportStar(require('@discordjs/util'), exports);
|
||||
__exportStar(require('@discordjs/ws'), exports);
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
'use strict';
|
||||
|
||||
const process = require('node:process');
|
||||
const { DefaultRestOptions, DefaultUserAgentAppendix } = require('@discordjs/rest');
|
||||
const { toSnakeCase } = require('./Transformers');
|
||||
const { version } = require('../../package.json');
|
||||
@@ -58,6 +57,16 @@ const { version } = require('../../package.json');
|
||||
* <info>This property is optional when the key is `invites`, `messages`, or `threads` and `lifetime` is set</info>
|
||||
*/
|
||||
|
||||
/**
|
||||
* A function to determine what strategy to use for sharding internally.
|
||||
* ```js
|
||||
* (manager) => new WorkerShardingStrategy(manager, { shardsPerWorker: 2 })
|
||||
* ```
|
||||
* @typedef {Function} BuildStrategyFunction
|
||||
* @param {WSWebSocketManager} manager The WebSocketManager that is going to initiate the sharding
|
||||
* @returns {IShardingStrategy} The strategy to use for sharding
|
||||
*/
|
||||
|
||||
/**
|
||||
* WebSocket options (these are left as snake_case to match the API)
|
||||
* @typedef {Object} WebsocketOptions
|
||||
@@ -65,6 +74,7 @@ const { version } = require('../../package.json');
|
||||
* sent in the initial guild member list, must be between 50 and 250
|
||||
* @property {number} [version=10] The Discord gateway version to use <warn>Changing this can break the library;
|
||||
* only set this if you know what you are doing</warn>
|
||||
* @property {BuildStrategyFunction} [buildStrategy] Builds the strategy to use for sharding
|
||||
*/
|
||||
|
||||
/**
|
||||
@@ -95,12 +105,6 @@ class Options extends null {
|
||||
sweepers: this.DefaultSweeperSettings,
|
||||
ws: {
|
||||
large_threshold: 50,
|
||||
compress: false,
|
||||
properties: {
|
||||
os: process.platform,
|
||||
browser: 'discord.js',
|
||||
device: 'discord.js',
|
||||
},
|
||||
version: 10,
|
||||
},
|
||||
rest: {
|
||||
@@ -200,3 +204,13 @@ module.exports = Options;
|
||||
* @external RESTOptions
|
||||
* @see {@link https://discord.js.org/docs/packages/rest/stable/RESTOptions:Interface}
|
||||
*/
|
||||
|
||||
/**
|
||||
* @external WSWebSocketManager
|
||||
* @see {@link https://discord.js.org/docs/packages/ws/stable/WebSocketManager:Class}
|
||||
*/
|
||||
|
||||
/**
|
||||
* @external IShardingStrategy
|
||||
* @see {@link https://discord.js.org/docs/packages/ws/stable/IShardingStrategy:Interface}
|
||||
*/
|
||||
|
||||
66
packages/discord.js/typings/index.d.ts
vendored
66
packages/discord.js/typings/index.d.ts
vendored
@@ -40,6 +40,7 @@ import {
|
||||
import { Awaitable, JSONEncodable } from '@discordjs/util';
|
||||
import { Collection } from '@discordjs/collection';
|
||||
import { BaseImageURLOptions, ImageURLOptions, RawFile, REST, RESTOptions } from '@discordjs/rest';
|
||||
import { WebSocketManager as WSWebSocketManager, IShardingStrategy, SessionInfo } from '@discordjs/ws';
|
||||
import {
|
||||
APIActionRowComponent,
|
||||
APIApplicationCommandInteractionData,
|
||||
@@ -3302,11 +3303,8 @@ export class WebhookClient extends WebhookMixin(BaseClient) {
|
||||
|
||||
export class WebSocketManager extends EventEmitter {
|
||||
private constructor(client: Client);
|
||||
private totalShards: number | string;
|
||||
private shardQueue: Set<WebSocketShard>;
|
||||
private readonly packetQueue: unknown[];
|
||||
private destroyed: boolean;
|
||||
private reconnecting: boolean;
|
||||
|
||||
public readonly client: Client;
|
||||
public gateway: string | null;
|
||||
@@ -3317,10 +3315,8 @@ export class WebSocketManager extends EventEmitter {
|
||||
public on(event: GatewayDispatchEvents, listener: (data: any, shardId: number) => void): this;
|
||||
public once(event: GatewayDispatchEvents, listener: (data: any, shardId: number) => void): this;
|
||||
|
||||
private debug(message: string, shard?: WebSocketShard): void;
|
||||
private debug(message: string, shardId?: number): void;
|
||||
private connect(): Promise<void>;
|
||||
private createShards(): Promise<void>;
|
||||
private reconnect(): Promise<void>;
|
||||
private broadcast(packet: unknown): void;
|
||||
private destroy(): void;
|
||||
private handlePacket(packet?: unknown, shard?: WebSocketShard): boolean;
|
||||
@@ -3339,26 +3335,11 @@ export interface WebSocketShardEventTypes {
|
||||
|
||||
export class WebSocketShard extends EventEmitter {
|
||||
private constructor(manager: WebSocketManager, id: number);
|
||||
private sequence: number;
|
||||
private closeSequence: number;
|
||||
private sessionId: string | null;
|
||||
private resumeURL: string | null;
|
||||
private sessionInfo: SessionInfo | null;
|
||||
public lastPingTimestamp: number;
|
||||
private lastHeartbeatAcked: boolean;
|
||||
private readonly ratelimit: {
|
||||
queue: unknown[];
|
||||
total: number;
|
||||
remaining: number;
|
||||
time: 60e3;
|
||||
timer: NodeJS.Timeout | null;
|
||||
};
|
||||
private connection: WebSocket | null;
|
||||
private helloTimeout: NodeJS.Timeout | null;
|
||||
private eventsAttached: boolean;
|
||||
private expectedGuilds: Set<Snowflake> | null;
|
||||
private readyTimeout: NodeJS.Timeout | null;
|
||||
private closeEmitted: boolean;
|
||||
private wsCloseTimeout: NodeJS.Timeout | null;
|
||||
|
||||
public manager: WebSocketManager;
|
||||
public id: number;
|
||||
@@ -3366,27 +3347,10 @@ export class WebSocketShard extends EventEmitter {
|
||||
public ping: number;
|
||||
|
||||
private debug(message: string): void;
|
||||
private connect(): Promise<void>;
|
||||
private onOpen(): void;
|
||||
private onMessage(event: MessageEvent): void;
|
||||
private onError(error: ErrorEvent | unknown): void;
|
||||
private onClose(event: CloseEvent): void;
|
||||
private onPacket(packet: unknown): void;
|
||||
private onReadyPacket(packet: unknown): void;
|
||||
private gotGuild(guildId: Snowflake): void;
|
||||
private checkReady(): void;
|
||||
private setHelloTimeout(time?: number): void;
|
||||
private setWsCloseTimeout(time?: number): void;
|
||||
private setHeartbeatTimer(time: number): void;
|
||||
private sendHeartbeat(): void;
|
||||
private ackHeartbeat(): void;
|
||||
private identify(): void;
|
||||
private identifyNew(): void;
|
||||
private identifyResume(): void;
|
||||
private _send(data: unknown): void;
|
||||
private processQueue(): void;
|
||||
private destroy(destroyOptions?: { closeCode?: number; reset?: boolean; emit?: boolean; log?: boolean }): void;
|
||||
private emitClose(event?: CloseEvent): void;
|
||||
private _cleanupConnection(): void;
|
||||
private _emitDestroyed(): void;
|
||||
|
||||
public send(data: unknown, important?: boolean): void;
|
||||
|
||||
@@ -3509,16 +3473,23 @@ export enum DiscordjsErrorCodes {
|
||||
TokenMissing = 'TokenMissing',
|
||||
ApplicationCommandPermissionsTokenMissing = 'ApplicationCommandPermissionsTokenMissing',
|
||||
|
||||
/** @deprecated */
|
||||
WSCloseRequested = 'WSCloseRequested',
|
||||
/** @deprecated */
|
||||
WSConnectionExists = 'WSConnectionExists',
|
||||
/** @deprecated */
|
||||
WSNotOpen = 'WSNotOpen',
|
||||
ManagerDestroyed = 'ManagerDestroyed',
|
||||
|
||||
BitFieldInvalid = 'BitFieldInvalid',
|
||||
|
||||
/** @deprecated */
|
||||
ShardingInvalid = 'ShardingInvalid',
|
||||
/** @deprecated */
|
||||
ShardingRequired = 'ShardingRequired',
|
||||
/** @deprecated */
|
||||
InvalidIntents = 'InvalidIntents',
|
||||
/** @deprecated */
|
||||
DisallowedIntents = 'DisallowedIntents',
|
||||
ShardingNoShards = 'ShardingNoShards',
|
||||
ShardingInProcess = 'ShardingInProcess',
|
||||
@@ -4874,10 +4845,11 @@ export interface ClientUserEditOptions {
|
||||
}
|
||||
|
||||
export interface CloseEvent {
|
||||
/** @deprecated */
|
||||
wasClean: boolean;
|
||||
code: number;
|
||||
/** @deprecated */
|
||||
reason: string;
|
||||
target: WebSocket;
|
||||
}
|
||||
|
||||
export type CollectorFilter<T extends unknown[]> = (...args: T) => boolean | Promise<boolean>;
|
||||
@@ -6332,15 +6304,8 @@ export interface WebhookMessageCreateOptions extends Omit<MessageCreateOptions,
|
||||
|
||||
export interface WebSocketOptions {
|
||||
large_threshold?: number;
|
||||
compress?: boolean;
|
||||
properties?: WebSocketProperties;
|
||||
version?: number;
|
||||
}
|
||||
|
||||
export interface WebSocketProperties {
|
||||
os?: string;
|
||||
browser?: string;
|
||||
device?: string;
|
||||
buildStrategy?(manager: WSWebSocketManager): IShardingStrategy;
|
||||
}
|
||||
|
||||
export interface WidgetActivity {
|
||||
@@ -6418,3 +6383,4 @@ export * from '@discordjs/builders';
|
||||
export * from '@discordjs/formatters';
|
||||
export * from '@discordjs/rest';
|
||||
export * from '@discordjs/util';
|
||||
export * from '@discordjs/ws';
|
||||
|
||||
Reference in New Issue
Block a user