diff options
| author | 8cy <[email protected]> | 2020-04-30 15:46:16 -0700 |
|---|---|---|
| committer | 8cy <[email protected]> | 2020-04-30 15:46:16 -0700 |
| commit | 3a4deac89054021b56ad5bd8005b2044cc085c98 (patch) | |
| tree | 3dd6af8503e497e46180b6b5231674f36bdce9f2 /node_modules/discord.js/src/sharding | |
| download | uppity-3a4deac89054021b56ad5bd8005b2044cc085c98.tar.xz uppity-3a4deac89054021b56ad5bd8005b2044cc085c98.zip | |
Up, up, uppity.
Diffstat (limited to 'node_modules/discord.js/src/sharding')
| -rw-r--r-- | node_modules/discord.js/src/sharding/Shard.js | 383 | ||||
| -rw-r--r-- | node_modules/discord.js/src/sharding/ShardClientUtil.js | 229 | ||||
| -rw-r--r-- | node_modules/discord.js/src/sharding/ShardingManager.js | 274 |
3 files changed, 886 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..3ef8b9b --- /dev/null +++ b/node_modules/discord.js/src/sharding/Shard.js @@ -0,0 +1,383 @@ +'use strict'; + +const EventEmitter = require('events'); +const path = require('path'); +const { Error } = require('../errors'); +const Util = 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 { + /** + * @param {ShardingManager} manager Manager that is creating this shard + * @param {number} id ID of this shard + */ + constructor(manager, id) { + super(); + + if (manager.mode === 'process') childProcess = require('child_process'); + else if (manager.mode === 'worker') Worker = require('worker_threads').Worker; + + /** + * Manager that created the shard + * @type {ShardingManager} + */ + this.manager = manager; + + /** + * ID of the shard in the manager + * @type {number} + */ + this.id = id; + + /** + * 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 = this._handleExit.bind(this, undefined); + } + + /** + * Forks a child process or creates a worker thread for the shard. + * <warn>You should not need to call this manually.</warn> + * @param {number} [spawnTimeout=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>} + */ + async spawn(spawnTimeout = 30000) { + if (this.process) throw new Error('SHARDING_PROCESS_EXISTS', this.id); + if (this.worker) throw new Error('SHARDING_WORKER_EXISTS', this.id); + + if (this.manager.mode === 'process') { + this.process = childProcess + .fork(path.resolve(this.manager.file), this.args, { + env: this.env, + execArgv: this.execArgv, + }) + .on('message', this._handleMessage.bind(this)) + .on('exit', this._exitListener); + } else if (this.manager.mode === 'worker') { + this.worker = new Worker(path.resolve(this.manager.file), { workerData: this.env }) + .on('message', this._handleMessage.bind(this)) + .on('exit', this._exitListener); + } + + /** + * 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('spawn', this.process || this.worker); + + if (spawnTimeout === -1 || spawnTimeout === Infinity) return this.process || this.worker; + await new Promise((resolve, reject) => { + const cleanup = () => { + clearTimeout(spawnTimeoutTimer); + this.off('ready', onReady); + this.off('disconnect', onDisconnect); + this.off('death', onDeath); + }; + + const onReady = () => { + cleanup(); + resolve(); + }; + + const onDisconnect = () => { + cleanup(); + reject(new Error('SHARDING_READY_DISCONNECTED', this.id)); + }; + + const onDeath = () => { + cleanup(); + reject(new Error('SHARDING_READY_DIED', this.id)); + }; + + const onTimeout = () => { + cleanup(); + reject(new Error('SHARDING_READY_TIMEOUT', this.id)); + }; + + const spawnTimeoutTimer = setTimeout(onTimeout, spawnTimeout); + this.once('ready', onReady); + this.once('disconnect', onDisconnect); + this.once('death', onDeath); + }); + return this.process || this.worker; + } + + /** + * 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); + } + + /** + * Kills and restarts the shard's process/worker. + * @param {number} [delay=500] How long to wait between killing the process/worker and restarting it (in milliseconds) + * @param {number} [spawnTimeout=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>} + */ + async respawn(delay = 500, spawnTimeout) { + this.kill(); + if (delay > 0) await Util.delayFor(delay); + return this.spawn(spawnTimeout); + } + + /** + * 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) { + 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 || message._fetchProp !== prop) return; + child.removeListener('message', listener); + this._fetches.delete(prop); + resolve(message._result); + }; + child.on('message', listener); + + this.send({ _fetchProp: prop }).catch(err => { + child.removeListener('message', listener); + 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 + * @returns {Promise<*>} Result of the script execution + */ + eval(script) { + if (this._evals.has(script)) return this._evals.get(script); + + const promise = new Promise((resolve, reject) => { + const child = this.process || this.worker; + + const listener = message => { + if (!message || message._eval !== script) return; + child.removeListener('message', listener); + this._evals.delete(script); + if (!message._error) resolve(message._result); + else reject(Util.makeError(message._error)); + }; + child.on('message', listener); + + const _eval = typeof script === 'function' ? `(${script})(this)` : script; + this.send({ _eval }).catch(err => { + child.removeListener('message', listener); + this._evals.delete(script); + reject(err); + }); + }); + + this._evals.set(script, 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#ready} event. + * @event Shard#ready + */ + this.emit('ready'); + return; + } + + // Shard has disconnected + if (message._disconnect) { + this.ready = false; + /** + * Emitted upon the shard's {@link Client#disconnect} event. + * @event Shard#disconnect + */ + this.emit('disconnect'); + return; + } + + // Shard is attempting to reconnect + if (message._reconnecting) { + this.ready = false; + /** + * Emitted upon the shard's {@link Client#reconnecting} event. + * @event Shard#reconnecting + */ + this.emit('reconnecting'); + return; + } + + // Shard is requesting a property fetch + if (message._sFetchProp) { + this.manager.fetchClientValues(message._sFetchProp).then( + results => this.send({ _sFetchProp: message._sFetchProp, _result: results }), + err => this.send({ _sFetchProp: message._sFetchProp, _error: Util.makePlainError(err) }), + ); + return; + } + + // Shard is requesting an eval broadcast + if (message._sEval) { + this.manager.broadcastEval(message._sEval).then( + results => this.send({ _sEval: message._sEval, _result: results }), + err => this.send({ _sEval: message._sEval, _error: Util.makePlainError(err) }), + ); + return; + } + + // Shard is requesting a respawn of all shards + if (message._sRespawnAll) { + const { shardDelay, respawnDelay, spawnTimeout } = message._sRespawnAll; + this.manager.respawnAll(shardDelay, respawnDelay, spawnTimeout).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('message', message); + } + + /** + * Handles the shard's process/worker exiting. + * @param {boolean} [respawn=this.manager.respawn] Whether to spawn the shard again + * @private + */ + _handleExit(respawn = this.manager.respawn) { + /** + * Emitted upon the shard's child process/worker exiting. + * @event Shard#death + * @param {ChildProcess|Worker} process Child process/worker that exited + */ + this.emit('death', this.process || this.worker); + + this.ready = false; + this.process = null; + this.worker = null; + this._evals.clear(); + this._fetches.clear(); + + if (respawn) this.spawn().catch(err => this.emit('error', err)); + } +} + +module.exports = Shard; diff --git a/node_modules/discord.js/src/sharding/ShardClientUtil.js b/node_modules/discord.js/src/sharding/ShardClientUtil.js new file mode 100644 index 0000000..79345d8 --- /dev/null +++ b/node_modules/discord.js/src/sharding/ShardClientUtil.js @@ -0,0 +1,229 @@ +'use strict'; + +const { Events } = require('../util/Constants'); +const Util = require('../util/Util'); + +/** + * Helper class for sharded clients spawned as a child process/worker, such as from a {@link ShardingManager}. + * Utilises IPC to send and receive data to/from the master process and other shards. + */ +class ShardClientUtil { + /** + * @param {Client} client Client of the current shard + * @param {ShardingManagerMode} mode Mode the shard was spawned with + */ + constructor(client, mode) { + /** + * Client for the shard + * @type {Client} + */ + this.client = client; + + /** + * Mode the shard was spawned with + * @type {ShardingManagerMode} + */ + this.mode = mode; + + /** + * Message port for the master process (only when {@link ShardClientUtil#mode} is `worker`) + * @type {?MessagePort} + */ + this.parentPort = null; + + if (mode === 'process') { + process.on('message', this._handleMessage.bind(this)); + client.on('ready', () => { + process.send({ _ready: true }); + }); + client.on('disconnect', () => { + process.send({ _disconnect: true }); + }); + client.on('reconnecting', () => { + process.send({ _reconnecting: true }); + }); + } else if (mode === 'worker') { + this.parentPort = require('worker_threads').parentPort; + this.parentPort.on('message', this._handleMessage.bind(this)); + client.on('ready', () => { + this.parentPort.postMessage({ _ready: true }); + }); + client.on('disconnect', () => { + this.parentPort.postMessage({ _disconnect: true }); + }); + client.on('reconnecting', () => { + this.parentPort.postMessage({ _reconnecting: true }); + }); + } + } + + /** + * Array of shard IDs of this client + * @type {number[]} + * @readonly + */ + get ids() { + return this.client.options.shards; + } + + /** + * Total number of shards + * @type {number} + * @readonly + */ + get count() { + return this.client.options.shardCount; + } + + /** + * Sends a message to the master process. + * @param {*} message Message to send + * @returns {Promise<void>} + * @emits Shard#message + */ + send(message) { + return new Promise((resolve, reject) => { + if (this.mode === 'process') { + process.send(message, err => { + if (err) reject(err); + else resolve(); + }); + } else if (this.mode === 'worker') { + this.parentPort.postMessage(message); + resolve(); + } + }); + } + + /** + * Fetches a client property value of each shard. + * @param {string} prop Name of the client property to get, using periods for nesting + * @returns {Promise<Array<*>>} + * @example + * client.shard.fetchClientValues('guilds.cache.size') + * .then(results => console.log(`${results.reduce((prev, val) => prev + val, 0)} total guilds`)) + * .catch(console.error); + * @see {@link ShardingManager#fetchClientValues} + */ + fetchClientValues(prop) { + return new Promise((resolve, reject) => { + const parent = this.parentPort || process; + + const listener = message => { + if (!message || message._sFetchProp !== prop) return; + parent.removeListener('message', listener); + if (!message._error) resolve(message._result); + else reject(Util.makeError(message._error)); + }; + parent.on('message', listener); + + this.send({ _sFetchProp: prop }).catch(err => { + parent.removeListener('message', listener); + reject(err); + }); + }); + } + + /** + * Evaluates a script or function on all shards, in the context of the {@link Clients}. + * @param {string|Function} script JavaScript to run on each shard + * @returns {Promise<Array<*>>} Results of the script execution + * @example + * client.shard.broadcastEval('this.guilds.cache.size') + * .then(results => console.log(`${results.reduce((prev, val) => prev + val, 0)} total guilds`)) + * .catch(console.error); + * @see {@link ShardingManager#broadcastEval} + */ + broadcastEval(script) { + return new Promise((resolve, reject) => { + const parent = this.parentPort || process; + script = typeof script === 'function' ? `(${script})(this)` : script; + + const listener = message => { + if (!message || message._sEval !== script) return; + parent.removeListener('message', listener); + if (!message._error) resolve(message._result); + else reject(Util.makeError(message._error)); + }; + parent.on('message', listener); + + this.send({ _sEval: script }).catch(err => { + parent.removeListener('message', listener); + reject(err); + }); + }); + } + + /** + * Requests a respawn of all shards. + * @param {number} [shardDelay=5000] How long to wait between shards (in milliseconds) + * @param {number} [respawnDelay=500] How long to wait between killing a shard's process/worker and restarting it + * (in milliseconds) + * @param {number} [spawnTimeout=30000] The amount in milliseconds to wait for a shard to become ready before + * continuing to another. (-1 or Infinity for no wait) + * @returns {Promise<void>} Resolves upon the message being sent + * @see {@link ShardingManager#respawnAll} + */ + respawnAll(shardDelay = 5000, respawnDelay = 500, spawnTimeout = 30000) { + return this.send({ _sRespawnAll: { shardDelay, respawnDelay, spawnTimeout } }); + } + + /** + * Handles an IPC message. + * @param {*} message Message received + * @private + */ + async _handleMessage(message) { + if (!message) return; + if (message._fetchProp) { + const props = message._fetchProp.split('.'); + let value = this.client; + for (const prop of props) value = value[prop]; + this._respond('fetchProp', { _fetchProp: message._fetchProp, _result: value }); + } else if (message._eval) { + try { + this._respond('eval', { _eval: message._eval, _result: await this.client._eval(message._eval) }); + } catch (err) { + this._respond('eval', { _eval: message._eval, _error: Util.makePlainError(err) }); + } + } + } + + /** + * Sends a message to the master process, emitting an error from the client upon failure. + * @param {string} type Type of response to send + * @param {*} message Message to send + * @private + */ + _respond(type, message) { + this.send(message).catch(err => { + err.message = `Error when sending ${type} response to master process: ${err.message}`; + /** + * Emitted when the client encounters an error. + * @event Client#error + * @param {Error} error The error encountered + */ + this.client.emit(Events.ERROR, err); + }); + } + + /** + * Creates/gets the singleton of this class. + * @param {Client} client The client to use + * @param {ShardingManagerMode} mode Mode the shard was spawned with + * @returns {ShardClientUtil} + */ + static singleton(client, mode) { + if (!this._singleton) { + this._singleton = new this(client, mode); + } else { + client.emit( + Events.WARN, + 'Multiple clients created in child process/worker; only the first will handle sharding helpers.', + ); + } + return this._singleton; + } +} + +module.exports = ShardClientUtil; diff --git a/node_modules/discord.js/src/sharding/ShardingManager.js b/node_modules/discord.js/src/sharding/ShardingManager.js new file mode 100644 index 0000000..0b9fdd9 --- /dev/null +++ b/node_modules/discord.js/src/sharding/ShardingManager.js @@ -0,0 +1,274 @@ +'use strict'; + +const EventEmitter = require('events'); +const fs = require('fs'); +const path = require('path'); +const Shard = require('./Shard'); +const { Error, TypeError, RangeError } = require('../errors'); +const Collection = require('../util/Collection'); +const Util = require('../util/Util'); + +/** + * This is a utility class that makes multi-process sharding of a bot an easy and painless experience. + * It works by spawning a self-contained {@link ChildProcess} or {@link Worker} for each individual shard, each + * containing its own instance of your bot's {@link Client}. They all have a line of communication with the master + * process, and there are several useful methods that utilise it in order to simplify tasks that are normally difficult + * with sharding. It can spawn a specific number of shards or the amount that Discord suggests for the bot, and takes a + * path to your main bot script to launch for each one. + * @extends {EventEmitter} + */ +class ShardingManager extends EventEmitter { + /** + * The mode to spawn shards with for a {@link ShardingManager}: either "process" to use child processes, or + * "worker" to use workers. The "worker" mode relies on the experimental + * [Worker threads](https://nodejs.org/api/worker_threads.html) functionality that is present in Node v10.5.0 or + * newer. Node must be started with the `--experimental-worker` flag to expose it. + * @typedef {Object} ShardingManagerMode + */ + + /** + * @param {string} file Path to your shard script file + * @param {Object} [options] Options for the sharding manager + * @param {string|number} [options.totalShards='auto'] Number of total shards of all shard managers or "auto" + * @param {string|number[]} [options.shardList='auto'] List of shards to spawn or "auto" + * @param {ShardingManagerMode} [options.mode='process'] Which mode to use for shards + * @param {boolean} [options.respawn=true] Whether shards should automatically respawn upon exiting + * @param {string[]} [options.shardArgs=[]] Arguments to pass to the shard script when spawning + * (only available when using the `process` mode) + * @param {string[]} [options.execArgv=[]] Arguments to pass to the shard script executable when spawning + * (only available when using the `process` mode) + * @param {string} [options.token] Token to use for automatic shard count and passing to shards + */ + constructor(file, options = {}) { + super(); + options = Util.mergeDefault( + { + totalShards: 'auto', + mode: 'process', + respawn: true, + shardArgs: [], + execArgv: [], + token: process.env.DISCORD_TOKEN, + }, + options, + ); + + /** + * Path to the shard script file + * @type {string} + */ + this.file = file; + if (!file) throw new Error('CLIENT_INVALID_OPTION', 'File', 'specified.'); + if (!path.isAbsolute(file)) this.file = path.resolve(process.cwd(), file); + const stats = fs.statSync(this.file); + if (!stats.isFile()) throw new Error('CLIENT_INVALID_OPTION', 'File', 'a file'); + + /** + * List of shards this sharding manager spawns + * @type {string|number[]} + */ + this.shardList = options.shardList || 'auto'; + if (this.shardList !== 'auto') { + if (!Array.isArray(this.shardList)) { + throw new TypeError('CLIENT_INVALID_OPTION', 'shardList', 'an array.'); + } + this.shardList = [...new Set(this.shardList)]; + if (this.shardList.length < 1) throw new RangeError('CLIENT_INVALID_OPTION', 'shardList', 'at least 1 ID.'); + if ( + this.shardList.some( + shardID => typeof shardID !== 'number' || isNaN(shardID) || !Number.isInteger(shardID) || shardID < 0, + ) + ) { + throw new TypeError('CLIENT_INVALID_OPTION', 'shardList', 'an array of positive integers.'); + } + } + + /** + * Amount of shards that all sharding managers spawn in total + * @type {number} + */ + this.totalShards = options.totalShards || 'auto'; + if (this.totalShards !== 'auto') { + if (typeof this.totalShards !== 'number' || isNaN(this.totalShards)) { + throw new TypeError('CLIENT_INVALID_OPTION', 'Amount of shards', 'a number.'); + } + if (this.totalShards < 1) throw new RangeError('CLIENT_INVALID_OPTION', 'Amount of shards', 'at least 1.'); + if (!Number.isInteger(this.totalShards)) { + throw new RangeError('CLIENT_INVALID_OPTION', 'Amount of shards', 'an integer.'); + } + } + + /** + * Mode for shards to spawn with + * @type {ShardingManagerMode} + */ + this.mode = options.mode; + if (this.mode !== 'process' && this.mode !== 'worker') { + throw new RangeError('CLIENT_INVALID_OPTION', 'Sharding mode', '"process" or "worker"'); + } + + /** + * Whether shards should automatically respawn upon exiting + * @type {boolean} + */ + this.respawn = options.respawn; + + /** + * An array of arguments to pass to shards (only when {@link ShardingManager#mode} is `process`) + * @type {string[]} + */ + this.shardArgs = options.shardArgs; + + /** + * An array of arguments to pass to the executable (only when {@link ShardingManager#mode} is `process`) + * @type {string[]} + */ + this.execArgv = options.execArgv; + + /** + * Token to use for obtaining the automatic shard count, and passing to shards + * @type {?string} + */ + this.token = options.token ? options.token.replace(/^Bot\s*/i, '') : null; + + /** + * A collection of shards that this manager has spawned + * @type {Collection<number, Shard>} + */ + this.shards = new Collection(); + + process.env.SHARDING_MANAGER = true; + process.env.SHARDING_MANAGER_MODE = this.mode; + process.env.DISCORD_TOKEN = this.token; + } + + /** + * Creates a single shard. + * <warn>Using this method is usually not necessary if you use the spawn method.</warn> + * @param {number} [id=this.shards.size] ID of the shard to create + * <info>This is usually not necessary to manually specify.</info> + * @returns {Shard} Note that the created shard needs to be explicitly spawned using its spawn method. + */ + createShard(id = this.shards.size) { + const shard = new Shard(this, id); + this.shards.set(id, shard); + /** + * Emitted upon creating a shard. + * @event ShardingManager#shardCreate + * @param {Shard} shard Shard that was created + */ + this.emit('shardCreate', shard); + return shard; + } + + /** + * Spawns multiple shards. + * @param {number|string} [amount=this.totalShards] Number of shards to spawn + * @param {number} [delay=5500] How long to wait in between spawning each shard (in milliseconds) + * @param {number} [spawnTimeout=30000] The amount in milliseconds to wait until the {@link Client} has become ready + * before resolving. (-1 or Infinity for no wait) + * @returns {Promise<Collection<number, Shard>>} + */ + async spawn(amount = this.totalShards, delay = 5500, spawnTimeout) { + // Obtain/verify the number of shards to spawn + if (amount === 'auto') { + amount = await Util.fetchRecommendedShards(this.token); + } else { + if (typeof amount !== 'number' || isNaN(amount)) { + throw new TypeError('CLIENT_INVALID_OPTION', 'Amount of shards', 'a number.'); + } + if (amount < 1) throw new RangeError('CLIENT_INVALID_OPTION', 'Amount of shards', 'at least 1.'); + if (!Number.isInteger(amount)) { + throw new TypeError('CLIENT_INVALID_OPTION', 'Amount of shards', 'an integer.'); + } + } + + // Make sure this many shards haven't already been spawned + if (this.shards.size >= amount) throw new Error('SHARDING_ALREADY_SPAWNED', this.shards.size); + if (this.shardList === 'auto' || this.totalShards === 'auto' || this.totalShards !== amount) { + this.shardList = [...Array(amount).keys()]; + } + if (this.totalShards === 'auto' || this.totalShards !== amount) { + this.totalShards = amount; + } + + if (this.shardList.some(shardID => shardID >= amount)) { + throw new RangeError( + 'CLIENT_INVALID_OPTION', + 'Amount of shards', + 'bigger than the highest shardID in the shardList option.', + ); + } + + // Spawn the shards + for (const shardID of this.shardList) { + const promises = []; + const shard = this.createShard(shardID); + promises.push(shard.spawn(spawnTimeout)); + if (delay > 0 && this.shards.size !== this.shardList.length) promises.push(Util.delayFor(delay)); + await Promise.all(promises); // eslint-disable-line no-await-in-loop + } + + return this.shards; + } + + /** + * Sends a message to all shards. + * @param {*} message Message to be sent to the shards + * @returns {Promise<Shard[]>} + */ + broadcast(message) { + const promises = []; + for (const shard of this.shards.values()) promises.push(shard.send(message)); + return Promise.all(promises); + } + + /** + * Evaluates a script on all shards, in the context of the {@link Client}s. + * @param {string} script JavaScript to run on each shard + * @returns {Promise<Array<*>>} Results of the script execution + */ + broadcastEval(script) { + const promises = []; + for (const shard of this.shards.values()) promises.push(shard.eval(script)); + return Promise.all(promises); + } + + /** + * Fetches a client property value of each shard. + * @param {string} prop Name of the client property to get, using periods for nesting + * @returns {Promise<Array<*>>} + * @example + * manager.fetchClientValues('guilds.cache.size') + * .then(results => console.log(`${results.reduce((prev, val) => prev + val, 0)} total guilds`)) + * .catch(console.error); + */ + fetchClientValues(prop) { + if (this.shards.size === 0) return Promise.reject(new Error('SHARDING_NO_SHARDS')); + if (this.shards.size !== this.shardList.length) return Promise.reject(new Error('SHARDING_IN_PROCESS')); + const promises = []; + for (const shard of this.shards.values()) promises.push(shard.fetchClientValue(prop)); + return Promise.all(promises); + } + + /** + * Kills all running shards and respawns them. + * @param {number} [shardDelay=5000] How long to wait between shards (in milliseconds) + * @param {number} [respawnDelay=500] How long to wait between killing a shard's process and restarting it + * (in milliseconds) + * @param {number} [spawnTimeout=30000] The amount in milliseconds to wait for a shard to become ready before + * continuing to another. (-1 or Infinity for no wait) + * @returns {Promise<Collection<string, Shard>>} + */ + async respawnAll(shardDelay = 5000, respawnDelay = 500, spawnTimeout) { + let s = 0; + for (const shard of this.shards.values()) { + const promises = [shard.respawn(respawnDelay, spawnTimeout)]; + if (++s < this.shards.size && shardDelay > 0) promises.push(Util.delayFor(shardDelay)); + await Promise.all(promises); // eslint-disable-line no-await-in-loop + } + return this.shards; + } +} + +module.exports = ShardingManager; |