summaryrefslogtreecommitdiff
path: root/node_modules/discord.js/src/sharding/Shard.js
diff options
context:
space:
mode:
Diffstat (limited to 'node_modules/discord.js/src/sharding/Shard.js')
-rw-r--r--node_modules/discord.js/src/sharding/Shard.js475
1 files changed, 475 insertions, 0 deletions
diff --git a/node_modules/discord.js/src/sharding/Shard.js b/node_modules/discord.js/src/sharding/Shard.js
new file mode 100644
index 0000000..f833309
--- /dev/null
+++ b/node_modules/discord.js/src/sharding/Shard.js
@@ -0,0 +1,475 @@
+'use strict';
+
+const EventEmitter = require('node:events');
+const path = require('node:path');
+const process = require('node:process');
+const { setTimeout, clearTimeout } = require('node:timers');
+const { setTimeout: sleep } = require('node:timers/promises');
+const { DiscordjsError, ErrorCodes } = require('../errors');
+const ShardEvents = require('../util/ShardEvents');
+const { makeError, makePlainError } = require('../util/Util');
+let childProcess = null;
+let Worker = null;
+
+/**
+ * A self-contained shard created by the {@link ShardingManager}. Each one has a {@link ChildProcess} that contains
+ * an instance of the bot and its {@link Client}. When its child process/worker exits for any reason, the shard will
+ * spawn a new one to replace it as necessary.
+ * @extends {EventEmitter}
+ */
+class Shard extends EventEmitter {
+ constructor(manager, id) {
+ super();
+
+ switch (manager.mode) {
+ case 'process':
+ childProcess = require('node:child_process');
+ break;
+ case 'worker':
+ Worker = require('node:worker_threads').Worker;
+ break;
+ }
+
+ /**
+ * Manager that created the shard
+ * @type {ShardingManager}
+ */
+ this.manager = manager;
+
+ /**
+ * The shard's id in the manager
+ * @type {number}
+ */
+ this.id = id;
+
+ /**
+ * Whether to pass silent flag to the shard's process (only when {@link ShardingManager#mode} is `process`)
+ * @type {boolean}
+ */
+ this.silent = manager.silent;
+
+ /**
+ * Arguments for the shard's process (only when {@link ShardingManager#mode} is `process`)
+ * @type {string[]}
+ */
+ this.args = manager.shardArgs ?? [];
+
+ /**
+ * Arguments for the shard's process executable (only when {@link ShardingManager#mode} is `process`)
+ * @type {string[]}
+ */
+ this.execArgv = manager.execArgv;
+
+ /**
+ * Environment variables for the shard's process, or workerData for the shard's worker
+ * @type {Object}
+ */
+ this.env = Object.assign({}, process.env, {
+ SHARDING_MANAGER: true,
+ SHARDS: this.id,
+ SHARD_COUNT: this.manager.totalShards,
+ DISCORD_TOKEN: this.manager.token,
+ });
+
+ /**
+ * Whether the shard's {@link Client} is ready
+ * @type {boolean}
+ */
+ this.ready = false;
+
+ /**
+ * Process of the shard (if {@link ShardingManager#mode} is `process`)
+ * @type {?ChildProcess}
+ */
+ this.process = null;
+
+ /**
+ * Worker of the shard (if {@link ShardingManager#mode} is `worker`)
+ * @type {?Worker}
+ */
+ this.worker = null;
+
+ /**
+ * Ongoing promises for calls to {@link Shard#eval}, mapped by the `script` they were called with
+ * @type {Map<string, Promise>}
+ * @private
+ */
+ this._evals = new Map();
+
+ /**
+ * Ongoing promises for calls to {@link Shard#fetchClientValue}, mapped by the `prop` they were called with
+ * @type {Map<string, Promise>}
+ * @private
+ */
+ this._fetches = new Map();
+
+ /**
+ * Listener function for the {@link ChildProcess}' `exit` event
+ * @type {Function}
+ * @private
+ */
+ this._exitListener = null;
+ }
+
+ /**
+ * Forks a child process or creates a worker thread for the shard.
+ * <warn>You should not need to call this manually.</warn>
+ * @param {number} [timeout=30000] The amount in milliseconds to wait until the {@link Client} has become ready
+ * before resolving (`-1` or `Infinity` for no wait)
+ * @returns {Promise<ChildProcess>}
+ */
+ spawn(timeout = 30_000) {
+ if (this.process) throw new DiscordjsError(ErrorCodes.ShardingProcessExists, this.id);
+ if (this.worker) throw new DiscordjsError(ErrorCodes.ShardingWorkerExists, this.id);
+
+ this._exitListener = this._handleExit.bind(this, undefined, timeout);
+
+ switch (this.manager.mode) {
+ case 'process':
+ this.process = childProcess
+ .fork(path.resolve(this.manager.file), this.args, {
+ env: this.env,
+ execArgv: this.execArgv,
+ silent: this.silent,
+ })
+ .on('message', this._handleMessage.bind(this))
+ .on('exit', this._exitListener);
+ break;
+ case 'worker':
+ this.worker = new Worker(path.resolve(this.manager.file), { workerData: this.env })
+ .on('message', this._handleMessage.bind(this))
+ .on('exit', this._exitListener);
+ break;
+ }
+
+ this._evals.clear();
+ this._fetches.clear();
+
+ const child = this.process ?? this.worker;
+
+ /**
+ * Emitted upon the creation of the shard's child process/worker.
+ * @event Shard#spawn
+ * @param {ChildProcess|Worker} process Child process/worker that was created
+ */
+ this.emit(ShardEvents.Spawn, child);
+
+ if (timeout === -1 || timeout === Infinity) return Promise.resolve(child);
+ return new Promise((resolve, reject) => {
+ const cleanup = () => {
+ clearTimeout(spawnTimeoutTimer);
+ this.off('ready', onReady);
+ this.off('disconnect', onDisconnect);
+ this.off('death', onDeath);
+ };
+
+ const onReady = () => {
+ cleanup();
+ resolve(child);
+ };
+
+ const onDisconnect = () => {
+ cleanup();
+ reject(new DiscordjsError(ErrorCodes.ShardingReadyDisconnected, this.id));
+ };
+
+ const onDeath = () => {
+ cleanup();
+ reject(new DiscordjsError(ErrorCodes.ShardingReadyDied, this.id));
+ };
+
+ const onTimeout = () => {
+ cleanup();
+ reject(new DiscordjsError(ErrorCodes.ShardingReadyTimeout, this.id));
+ };
+
+ const spawnTimeoutTimer = setTimeout(onTimeout, timeout);
+ this.once('ready', onReady);
+ this.once('disconnect', onDisconnect);
+ this.once('death', onDeath);
+ });
+ }
+
+ /**
+ * Immediately kills the shard's process/worker and does not restart it.
+ */
+ kill() {
+ if (this.process) {
+ this.process.removeListener('exit', this._exitListener);
+ this.process.kill();
+ } else {
+ this.worker.removeListener('exit', this._exitListener);
+ this.worker.terminate();
+ }
+
+ this._handleExit(false);
+ }
+
+ /**
+ * Options used to respawn a shard.
+ * @typedef {Object} ShardRespawnOptions
+ * @property {number} [delay=500] How long to wait between killing the process/worker and
+ * restarting it (in milliseconds)
+ * @property {number} [timeout=30000] The amount in milliseconds to wait until the {@link Client}
+ * has become ready before resolving (`-1` or `Infinity` for no wait)
+ */
+
+ /**
+ * Kills and restarts the shard's process/worker.
+ * @param {ShardRespawnOptions} [options] Options for respawning the shard
+ * @returns {Promise<ChildProcess>}
+ */
+ async respawn({ delay = 500, timeout = 30_000 } = {}) {
+ this.kill();
+ if (delay > 0) await sleep(delay);
+ return this.spawn(timeout);
+ }
+
+ /**
+ * Sends a message to the shard's process/worker.
+ * @param {*} message Message to send to the shard
+ * @returns {Promise<Shard>}
+ */
+ send(message) {
+ return new Promise((resolve, reject) => {
+ if (this.process) {
+ this.process.send(message, err => {
+ if (err) reject(err);
+ else resolve(this);
+ });
+ } else {
+ this.worker.postMessage(message);
+ resolve(this);
+ }
+ });
+ }
+
+ /**
+ * Fetches a client property value of the shard.
+ * @param {string} prop Name of the client property to get, using periods for nesting
+ * @returns {Promise<*>}
+ * @example
+ * shard.fetchClientValue('guilds.cache.size')
+ * .then(count => console.log(`${count} guilds in shard ${shard.id}`))
+ * .catch(console.error);
+ */
+ fetchClientValue(prop) {
+ // Shard is dead (maybe respawning), don't cache anything and error immediately
+ if (!this.process && !this.worker) {
+ return Promise.reject(new DiscordjsError(ErrorCodes.ShardingNoChildExists, this.id));
+ }
+
+ // Cached promise from previous call
+ if (this._fetches.has(prop)) return this._fetches.get(prop);
+
+ const promise = new Promise((resolve, reject) => {
+ const child = this.process ?? this.worker;
+
+ const listener = message => {
+ if (message?._fetchProp !== prop) return;
+ child.removeListener('message', listener);
+ this.decrementMaxListeners(child);
+ this._fetches.delete(prop);
+ if (!message._error) resolve(message._result);
+ else reject(makeError(message._error));
+ };
+
+ this.incrementMaxListeners(child);
+ child.on('message', listener);
+
+ this.send({ _fetchProp: prop }).catch(err => {
+ child.removeListener('message', listener);
+ this.decrementMaxListeners(child);
+ this._fetches.delete(prop);
+ reject(err);
+ });
+ });
+
+ this._fetches.set(prop, promise);
+ return promise;
+ }
+
+ /**
+ * Evaluates a script or function on the shard, in the context of the {@link Client}.
+ * @param {string|Function} script JavaScript to run on the shard
+ * @param {*} [context] The context for the eval
+ * @returns {Promise<*>} Result of the script execution
+ */
+ eval(script, context) {
+ // Stringify the script if it's a Function
+ const _eval = typeof script === 'function' ? `(${script})(this, ${JSON.stringify(context)})` : script;
+
+ // Shard is dead (maybe respawning), don't cache anything and error immediately
+ if (!this.process && !this.worker) {
+ return Promise.reject(new DiscordjsError(ErrorCodes.ShardingNoChildExists, this.id));
+ }
+
+ // Cached promise from previous call
+ if (this._evals.has(_eval)) return this._evals.get(_eval);
+
+ const promise = new Promise((resolve, reject) => {
+ const child = this.process ?? this.worker;
+
+ const listener = message => {
+ if (message?._eval !== _eval) return;
+ child.removeListener('message', listener);
+ this.decrementMaxListeners(child);
+ this._evals.delete(_eval);
+ if (!message._error) resolve(message._result);
+ else reject(makeError(message._error));
+ };
+
+ this.incrementMaxListeners(child);
+ child.on('message', listener);
+
+ this.send({ _eval }).catch(err => {
+ child.removeListener('message', listener);
+ this.decrementMaxListeners(child);
+ this._evals.delete(_eval);
+ reject(err);
+ });
+ });
+
+ this._evals.set(_eval, promise);
+ return promise;
+ }
+
+ /**
+ * Handles a message received from the child process/worker.
+ * @param {*} message Message received
+ * @private
+ */
+ _handleMessage(message) {
+ if (message) {
+ // Shard is ready
+ if (message._ready) {
+ this.ready = true;
+ /**
+ * Emitted upon the shard's {@link Client#event:shardReady} event.
+ * @event Shard#ready
+ */
+ this.emit(ShardEvents.Ready);
+ return;
+ }
+
+ // Shard has disconnected
+ if (message._disconnect) {
+ this.ready = false;
+ /**
+ * Emitted upon the shard's {@link Client#event:shardDisconnect} event.
+ * @event Shard#disconnect
+ */
+ this.emit(ShardEvents.Disconnect);
+ return;
+ }
+
+ // Shard is attempting to reconnect
+ if (message._reconnecting) {
+ this.ready = false;
+ /**
+ * Emitted upon the shard's {@link Client#event:shardReconnecting} event.
+ * @event Shard#reconnecting
+ */
+ this.emit(ShardEvents.Reconnecting);
+ return;
+ }
+
+ // Shard has resumed
+ if (message._resume) {
+ this.ready = true;
+ /**
+ * Emitted upon the shard's {@link Client#event:shardResume} event.
+ * @event Shard#resume
+ */
+ this.emit(ShardEvents.Resume);
+ return;
+ }
+
+ // Shard is requesting a property fetch
+ if (message._sFetchProp) {
+ const resp = { _sFetchProp: message._sFetchProp, _sFetchPropShard: message._sFetchPropShard };
+ this.manager.fetchClientValues(message._sFetchProp, message._sFetchPropShard).then(
+ results => this.send({ ...resp, _result: results }),
+ err => this.send({ ...resp, _error: makePlainError(err) }),
+ );
+ return;
+ }
+
+ // Shard is requesting an eval broadcast
+ if (message._sEval) {
+ const resp = { _sEval: message._sEval, _sEvalShard: message._sEvalShard };
+ this.manager._performOnShards('eval', [message._sEval], message._sEvalShard).then(
+ results => this.send({ ...resp, _result: results }),
+ err => this.send({ ...resp, _error: makePlainError(err) }),
+ );
+ return;
+ }
+
+ // Shard is requesting a respawn of all shards
+ if (message._sRespawnAll) {
+ const { shardDelay, respawnDelay, timeout } = message._sRespawnAll;
+ this.manager.respawnAll({ shardDelay, respawnDelay, timeout }).catch(() => {
+ // Do nothing
+ });
+ return;
+ }
+ }
+
+ /**
+ * Emitted upon receiving a message from the child process/worker.
+ * @event Shard#message
+ * @param {*} message Message that was received
+ */
+ this.emit(ShardEvents.Message, message);
+ }
+
+ /**
+ * Handles the shard's process/worker exiting.
+ * @param {boolean} [respawn=this.manager.respawn] Whether to spawn the shard again
+ * @param {number} [timeout] The amount in milliseconds to wait until the {@link Client}
+ * has become ready (`-1` or `Infinity` for no wait)
+ * @private
+ */
+ _handleExit(respawn = this.manager.respawn, timeout) {
+ /**
+ * Emitted upon the shard's child process/worker exiting.
+ * @event Shard#death
+ * @param {ChildProcess|Worker} process Child process/worker that exited
+ */
+ this.emit(ShardEvents.Death, this.process ?? this.worker);
+
+ this.ready = false;
+ this.process = null;
+ this.worker = null;
+ this._evals.clear();
+ this._fetches.clear();
+
+ if (respawn) this.spawn(timeout).catch(err => this.emit(ShardEvents.Error, err));
+ }
+
+ /**
+ * Increments max listeners by one for a given emitter, if they are not zero.
+ * @param {EventEmitter|process} emitter The emitter that emits the events.
+ * @private
+ */
+ incrementMaxListeners(emitter) {
+ const maxListeners = emitter.getMaxListeners();
+ if (maxListeners !== 0) {
+ emitter.setMaxListeners(maxListeners + 1);
+ }
+ }
+
+ /**
+ * Decrements max listeners by one for a given emitter, if they are not zero.
+ * @param {EventEmitter|process} emitter The emitter that emits the events.
+ * @private
+ */
+ decrementMaxListeners(emitter) {
+ const maxListeners = emitter.getMaxListeners();
+ if (maxListeners !== 0) {
+ emitter.setMaxListeners(maxListeners - 1);
+ }
+ }
+}
+
+module.exports = Shard;