diff --git a/index.js b/index.js index 77c3aa7d27..f91bc6d13e 100644 --- a/index.js +++ b/index.js @@ -10,7 +10,7 @@ const Buffer = require('buffer').Buffer const net = require('net') const util = require('util') const utils = require('./lib/utils') -const Command = require('./lib/command') +const reconnect = require('./lib/reconnect') const Queue = require('denque') const errorClasses = require('./lib/customErrors') const EventEmitter = require('events') @@ -31,19 +31,6 @@ const SUBSCRIBE_COMMANDS = { function noop () {} -function handleDetectBuffersReply (reply, command, bufferArgs) { - if (bufferArgs === false || this.messageBuffers) { - // If detectBuffers option was specified, then the reply from the parser will be a buffer. - // If this command did not use Buffer arguments, then convert the reply to Strings here. - reply = utils.replyToStrings(reply) - } - - if (command === 'hgetall') { - reply = utils.replyToObject(reply) - } - return reply -} - // Attention: The second parameter might be removed at will and is not officially supported. // Do not rely on this function RedisClient (options, stream) { @@ -93,10 +80,6 @@ function RedisClient (options, stream) { ) options.detectBuffers = false } - if (options.detectBuffers) { - // We only need to look at the arguments if we do not know what we have to return - this.handleReply = handleDetectBuffersReply - } this.shouldBuffer = false this.commandQueue = new Queue() // Holds sent commands to de-pipeline them this.offlineQueue = new Queue() // Holds commands issued but not able to be sent @@ -137,7 +120,6 @@ function RedisClient (options, stream) { this.on('newListener', function (event) { if ((event === 'messageBuffer' || event === 'pmessageBuffer') && !this.buffers && !this.messageBuffers) { this.messageBuffers = true - this.handleReply = handleDetectBuffersReply this._replyParser.setReturnBuffers(true) } }) @@ -154,13 +136,6 @@ RedisClient.connectionId = 0 ******************************************************************************/ -RedisClient.prototype.handleReply = function (reply, command) { - if (command === 'hgetall') { - reply = utils.replyToObject(reply) - } - return reply -} - RedisClient.prototype.cork = noop RedisClient.prototype.uncork = noop @@ -216,143 +191,7 @@ RedisClient.prototype.onError = function (err) { } // 'error' events get turned into exceptions if they aren't listened for. If the user handled this error // then we should try to reconnect. - this.connectionGone('error', err) -} - -RedisClient.prototype.onConnect = function () { - debug('Stream connected %s id %s', this.address, this.connectionId) - - this.connected = true - this.ready = false - this.emittedEnd = false - this._stream.setKeepAlive(this.options.socketKeepalive) - this._stream.setTimeout(0) - - this.emit('connect') - this.initializeRetryVars() - - if (this.options.noReadyCheck) { - this.onReady() - } else { - this.readyCheck() - } -} - -RedisClient.prototype.onReady = function () { - debug('onReady called %s id %s', this.address, this.connectionId) - this.ready = true - - this.cork = () => { - this.pipeline = true - this._stream.cork() - } - this.uncork = () => { - if (this.fireStrings) { - this.writeStrings() - } else { - this.writeBuffers() - } - this.pipeline = false - this.fireStrings = true - // TODO: Consider using next tick here. See https://github.com/NodeRedis/nodeRedis/issues/1033 - this._stream.uncork() - } - - // Restore modal commands from previous connection. The order of the commands is important - if (this.selectedDb !== undefined) { - this.internalSendCommand(new Command('select', [this.selectedDb])).catch((err) => { - if (!this.closing) { - // TODO: These internal things should be wrapped in a - // special error that describes what is happening - process.nextTick(() => this.emit('error', err)) - } - }) - } - if (this.monitoring) { // Monitor has to be fired before pub sub commands - this.internalSendCommand(new Command('monitor', [])).catch((err) => { - if (!this.closing) { - process.nextTick(() => this.emit('error', err)) - } - }) - } - const callbackCount = Object.keys(this.subscriptionSet).length - // TODO: Replace the disableResubscribing by a individual function that may be called - // Add HOOKS!!! - // Replace the disableResubscribing by: - // resubmit: { - // select: true, - // monitor: true, - // subscriptions: true, - // // individual: function noop () {} - // } - if (!this.options.disableResubscribing && callbackCount) { - debug('Sending pub/sub onReady commands') - for (const key in this.subscriptionSet) { - const command = key.slice(0, key.indexOf('_')) - const args = this.subscriptionSet[key] - this[command]([args]).catch((err) => { - if (!this.closing) { - process.nextTick(() => this.emit('error', err)) - } - }) - } - } - this.sendOfflineQueue() - this.emit('ready') -} - -RedisClient.prototype.onInfoFail = function (err) { - if (this.closing) { - return - } - - if (err.message === "ERR unknown command 'info'") { - this.onReady() - return - } - err.message = `Ready check failed: ${err.message}` - this.emit('error', err) - return -} - -RedisClient.prototype.onInfoCmd = function (res) { - /* istanbul ignore if: some servers might not respond with any info data. This is just a safety check that is difficult to test */ - if (!res) { - debug('The info command returned without any data.') - this.onReady() - return - } - - if (!this.serverInfo.loading || this.serverInfo.loading === '0') { - // If the master_link_status exists but the link is not up, try again after 50 ms - if (this.serverInfo.master_link_status && this.serverInfo.master_link_status !== 'up') { - this.serverInfo.loading_eta_seconds = 0.05 - } else { - // Eta loading should change - debug('Redis server ready.') - this.onReady() - return - } - } - - var retryTime = +this.serverInfo.loading_eta_seconds * 1000 - if (retryTime > 1000) { - retryTime = 1000 - } - debug('Redis server still loading, trying again in %s', retryTime) - return new Promise((resolve) => { - setTimeout((self) => resolve(self.readyCheck()), retryTime, this) - }) -} - -RedisClient.prototype.readyCheck = function () { - debug('Checking server ready state...') - // Always fire this info command as first command even if other commands are already queued up - this.ready = true - this.info() - .then((res) => this.onInfoCmd(res)) - .catch((err) => this.onInfoFail(err)) - this.ready = false + reconnect(this, 'error', err) } RedisClient.prototype.sendOfflineQueue = function () { @@ -362,116 +201,6 @@ RedisClient.prototype.sendOfflineQueue = function () { } } -const retryConnection = function (self, error) { - debug('Retrying connection...') - - const reconnectParams = { - delay: self.retryDelay, - attempt: self.attempts, - error, - totalRetryTime: self.retryTotaltime, - timesConnected: self.timesConnected - } - self.emit('reconnecting', reconnectParams) - - self.retryTotaltime += self.retryDelay - self.attempts += 1 - connect(self) - self.retryTimer = null -} - -RedisClient.prototype.connectionGone = function (why, error) { - // If a retry is already in progress, just let that happen - if (this.retryTimer) { - return - } - error = error || null - - debug('Redis connection is gone from %s event.', why) - this.connected = false - this.ready = false - // Deactivate cork to work with the offline queue - this.cork = noop - this.uncork = noop - this.pipeline = false - this.pubSubMode = 0 - - // since we are collapsing end and close, users don't expect to be called twice - if (!this.emittedEnd) { - this.emit('end') - this.emittedEnd = true - } - - if (why === 'timeout') { - var message = 'Redis connection in broken state: connection timeout exceeded.' - const err = new Error(message) - // TODO: Find better error codes... - err.code = 'CONNECTION_BROKEN' - this.flushAndError({ - message: message, - code: 'CONNECTION_BROKEN' - }) - this.emit('error', err) - this.end(false) - return - } - - // If this is a requested shutdown, then don't retry - if (this.closing) { - debug('Connection ended by quit / end command, not retrying.') - this.flushAndError({ - message: 'Stream connection ended and command aborted.', - code: 'NR_CLOSED' - }, { - error - }) - return - } - - this.retryDelay = this.retryStrategy({ - attempt: this.attempts, - error, - totalRetryTime: this.retryTotaltime, - timesConnected: this.timesConnected - }) - if (typeof this.retryDelay !== 'number') { - // Pass individual error through - if (this.retryDelay instanceof Error) { - error = this.retryDelay - } - this.flushAndError({ - message: 'Stream connection ended and command aborted.', - code: 'NR_CLOSED' - }, { - error - }) - // TODO: Check if this is so smart - if (error) { - this.emit('error', error) - } - this.end(false) - return - } - - // Retry commands after a reconnect instead of throwing an error. Use this with caution - if (this.options.retryUnfulfilledCommands) { - this.offlineQueue.unshift.apply(this.offlineQueue, this.commandQueue.toArray()) - this.commandQueue.clear() - } else if (this.commandQueue.length !== 0) { - this.flushAndError({ - message: 'Redis connection lost and command aborted.', - code: 'UNCERTAIN_STATE' - }, { - error, - queues: ['commandQueue'] - }) - } - - debug('Retry connection in %s ms', this.retryDelay) - - this.retryTimer = setTimeout(retryConnection, this.retryDelay, this, error) -} - RedisClient.prototype.returnError = function (err) { const commandObj = this.commandQueue.shift() if (commandObj.error) { @@ -496,12 +225,12 @@ RedisClient.prototype.returnError = function (err) { commandObj.callback(err) } -function normalReply (self, reply) { - const commandObj = self.commandQueue.shift() - if (self._multi === false) { - reply = self.handleReply(reply, commandObj.command, commandObj.bufferArgs) +function normalReply (client, reply) { + const command = client.commandQueue.shift() + if (client._multi === false) { + reply = utils.handleReply(client, reply, command) } - commandObj.callback(null, reply) + command.callback(null, reply) } function subscribeUnsubscribe (self, reply, type) { diff --git a/lib/connect.js b/lib/connect.js index b54d60fd2d..71d5209163 100644 --- a/lib/connect.js +++ b/lib/connect.js @@ -3,6 +3,8 @@ const tls = require('tls') const Parser = require('redis-parser') const net = require('net') +const reconnect = require('./reconnect') +const onConnect = require('./readyHandler') const debug = require('./debug') /** @@ -77,7 +79,7 @@ function connect (client) { // TODO: Investigate why this is not properly triggered client._stream.setTimeout(client.connectTimeout, () => { // Note: This is only tested if a internet connection is established - client.connectionGone('timeout') + reconnect(client, 'timeout') }) } @@ -86,7 +88,7 @@ function connect (client) { client._stream.once(connectEvent, () => { client._stream.removeAllListeners('timeout') client.timesConnected++ - client.onConnect() + onConnect(client) }) client._stream.on('data', (bufferFromSocket) => { @@ -105,11 +107,11 @@ function connect (client) { }) client._stream.once('close', (hadError) => { - client.connectionGone('close') + reconnect(client, 'close') }) client._stream.once('end', () => { - client.connectionGone('end') + reconnect(client, 'end') }) client._stream.setNoDelay() diff --git a/lib/debug.js b/lib/debug.js index 66535700d0..a03a127ffc 100644 --- a/lib/debug.js +++ b/lib/debug.js @@ -1,12 +1,15 @@ 'use strict' -const index = require('../') +var index +function lazyIndex () { + return index || require('../') +} /** * @description Print a debug statement if in debug mode */ function debug () { - if (index.debugMode) { + if (lazyIndex().debugMode) { console.error.apply(null, arguments) } } diff --git a/lib/multi.js b/lib/multi.js index 9d894a5253..2160037549 100644 --- a/lib/multi.js +++ b/lib/multi.js @@ -3,6 +3,7 @@ const Queue = require('denque') const utils = require('./utils') const Command = require('./command') +const handleReply = utils.handleReply /** * @description Queues all transaction commands and checks if a queuing error @@ -23,9 +24,6 @@ function pipelineTransactionCommand (multi, command, index) { multi.errors.push(err) return } - // Keep track of who wants buffer responses: - // By the time the callback is called the command got the bufferArgs attribute attached - multi.wantsBuffers[index] = command.bufferArgs tmp(null, reply) } return multi._client.internalSendCommand(command) @@ -41,8 +39,10 @@ function pipelineTransactionCommand (multi, command, index) { function multiCallback (multi, replies) { if (replies) { var i = 0 - while (multi._queue.length !== 0) { - const command = multi._queue.shift() + const queue = multi._queue + const client = multi._client + while (queue.length !== 0) { + const command = queue.shift() if (replies[i] instanceof Error) { const match = replies[i].message.match(utils.errCode) // LUA script could return user errors that don't behave like all other errors! @@ -53,7 +53,7 @@ function multiCallback (multi, replies) { command.callback(replies[i]) } else { // If we asked for strings, even in detectBuffers mode, then return strings: - replies[i] = multi._client.handleReply(replies[i], command.command, multi.wantsBuffers[i]) + replies[i] = handleReply(client, replies[i], command) command.callback(null, replies[i]) } i++ diff --git a/lib/readyHandler.js b/lib/readyHandler.js new file mode 100644 index 0000000000..a27d5c5628 --- /dev/null +++ b/lib/readyHandler.js @@ -0,0 +1,140 @@ +'use strict' + +const debug = require('./debug') +const Command = require('./command') + +function onConnect (client) { + debug('Stream connected %s id %s', client.address, client.connectionId) + + client.connected = true + client.ready = false + client.emittedEnd = false + client._stream.setKeepAlive(client.options.socketKeepalive) + client._stream.setTimeout(0) + + client.emit('connect') + client.initializeRetryVars() + + if (client.options.noReadyCheck) { + onReady(client) + } else { + readyCheck(client) + } +} + +function onReady (client) { + debug('onReady called %s id %s', client.address, client.connectionId) + client.ready = true + + client.cork = () => { + client.pipeline = true + client._stream.cork() + } + client.uncork = () => { + if (client.fireStrings) { + client.writeStrings() + } else { + client.writeBuffers() + } + client.pipeline = false + client.fireStrings = true + // TODO: Consider using next tick here. See https://github.com/NodeRedis/nodeRedis/issues/1033 + client._stream.uncork() + } + + // Restore modal commands from previous connection. The order of the commands is important + if (client.selectedDb !== undefined) { + client.internalSendCommand(new Command('select', [client.selectedDb])).catch((err) => { + if (!client.closing) { + // TODO: These internal things should be wrapped in a + // special error that describes what is happening + process.nextTick(() => client.emit('error', err)) + } + }) + } + if (client.monitoring) { // Monitor has to be fired before pub sub commands + client.internalSendCommand(new Command('monitor', [])).catch((err) => { + if (!client.closing) { + process.nextTick(() => client.emit('error', err)) + } + }) + } + const callbackCount = Object.keys(client.subscriptionSet).length + // TODO: Replace the disableResubscribing by a individual function that may be called + // Add HOOKS!!! + // Replace the disableResubscribing by: + // resubmit: { + // select: true, + // monitor: true, + // subscriptions: true, + // // individual: function noop () {} + // } + if (!client.options.disableResubscribing && callbackCount) { + debug('Sending pub/sub onReady commands') + for (const key in client.subscriptionSet) { + const command = key.slice(0, key.indexOf('_')) + const args = client.subscriptionSet[key] + client[command]([args]).catch((err) => { + if (!client.closing) { + process.nextTick(() => client.emit('error', err)) + } + }) + } + } + client.sendOfflineQueue() + client.emit('ready') +} + +function onInfoFail (client, err) { + if (client.closing) { + return + } + + if (err.message === "ERR unknown command 'info'") { + onReady(client) + return + } + err.message = `Ready check failed: ${err.message}` + client.emit('error', err) + return +} + +function onInfoCmd (client, res) { + /* istanbul ignore if: some servers might not respond with any info data. client is just a safety check that is difficult to test */ + if (!res) { + debug('The info command returned without any data.') + onReady(client) + return + } + + if (!client.serverInfo.loading || client.serverInfo.loading === '0') { + // If the master_link_status exists but the link is not up, try again after 50 ms + if (client.serverInfo.master_link_status && client.serverInfo.master_link_status !== 'up') { + client.serverInfo.loading_eta_seconds = 0.05 + } else { + // Eta loading should change + debug('Redis server ready.') + onReady(client) + return + } + } + + var retryTime = +client.serverInfo.loading_eta_seconds * 1000 + if (retryTime > 1000) { + retryTime = 1000 + } + debug('Redis server still loading, trying again in %s', retryTime) + setTimeout((client) => readyCheck(client), retryTime, client) +} + +function readyCheck (client) { + debug('Checking server ready state...') + // Always fire client info command as first command even if other commands are already queued up + client.ready = true + client.info() + .then((res) => onInfoCmd(client, res)) + .catch((err) => onInfoFail(client, err)) + client.ready = false +} + +module.exports = onConnect diff --git a/lib/reconnect.js b/lib/reconnect.js new file mode 100644 index 0000000000..2dc123e9db --- /dev/null +++ b/lib/reconnect.js @@ -0,0 +1,134 @@ +'use strict' + +const debug = require('./debug') +var lazyConnect = function (client) { + lazyConnect = require('./connect') + lazyConnect(client) +} +const noop = () => {} + +/** + * @description Try connecting to a server again + * + * @param {RedisClient} client + * @param {Error} [error] + */ +function retryConnection (client, error) { + debug('Retrying connection...') + + const reconnectParams = { + delay: client.retryDelay, + attempt: client.attempts, + error, + totalRetryTime: client.retryTotaltime, + timesConnected: client.timesConnected + } + client.emit('reconnecting', reconnectParams) + + client.retryTotaltime += client.retryDelay + client.attempts += 1 + lazyConnect(client) + client.retryTimer = null +} + +/** + * @description The connection is lost. Retry if requested + * + * @param {RedisClient} client + * @param {string} why + * @param {Error} [error] + */ +function reconnect (client, why, error) { + // If a retry is already in progress, just let that happen + if (client.retryTimer) { + return + } + // TODO: Always return an error? + error = error || null + + debug('Redis connection is gone from %s event.', why) + client.connected = false + client.ready = false + // Deactivate cork to work with the offline queue + client.cork = noop + client.uncork = noop + client.pipeline = false + client.pubSubMode = 0 + + // since we are collapsing end and close, users don't expect to be called twice + if (!client.emittedEnd) { + client.emit('end') + client.emittedEnd = true + } + + if (why === 'timeout') { + var message = 'Redis connection in broken state: connection timeout exceeded.' + const err = new Error(message) + // TODO: Find better error codes... + err.code = 'CONNECTION_BROKEN' + client.flushAndError({ + message: message, + code: 'CONNECTION_BROKEN' + }) + client.emit('error', err) + client.end(false) + return + } + + // If client is a requested shutdown, then don't retry + if (client.closing) { + debug('Connection ended by quit / end command, not retrying.') + client.flushAndError({ + message: 'Stream connection ended and command aborted.', + code: 'NR_CLOSED' + }, { + error + }) + return + } + + client.retryDelay = client.retryStrategy({ + attempt: client.attempts, + error, + totalRetryTime: client.retryTotaltime, + timesConnected: client.timesConnected + }) + if (typeof client.retryDelay !== 'number') { + // Pass individual error through + if (client.retryDelay instanceof Error) { + error = client.retryDelay + } + client.flushAndError({ + message: 'Stream connection ended and command aborted.', + code: 'NR_CLOSED' + }, { + error + }) + // TODO: Check if client is so smart + if (error) { + client.emit('error', error) + } + client.end(false) + return + } + + // Retry commands after a reconnect instead of throwing an error. Use this with caution + if (client.options.retryUnfulfilledCommands) { + client.offlineQueue.unshift.apply(client.offlineQueue, client.commandQueue.toArray()) + client.commandQueue.clear() + } else if (client.commandQueue.length !== 0) { + client.flushAndError({ + message: 'Redis connection lost and command aborted.', + code: 'UNCERTAIN_STATE' + }, { + error, + queues: ['commandQueue'] + }) + } + + debug('Retry connection in %s ms', client.retryDelay) + + client.retryTimer = setTimeout((client, error) => retryConnection(client, error), client.retryDelay, client, error) +} + +module.exports = reconnect diff --git a/lib/utils.js b/lib/utils.js index 85742f6c52..af5b6285b4 100644 --- a/lib/utils.js +++ b/lib/utils.js @@ -129,6 +129,28 @@ function warn (client, msg) { } } +/** + * @description Transform the reply as needed. + * + * If detectBuffers option was specified, then the reply from the parser will be a buffer. + * If this command did not use Buffer arguments, then convert the reply to Strings here. + * + * @param {RedisClient} client + * @param {string|number|null|Buffer|any[]} reply + * @param {Command} command + * @returns {string|number|null|Buffer|any[]|object} + */ +function handleReply (client, reply, command) { + if (client.options.detectBuffers === true && command.bufferArgs === false) { // client.messageBuffers + reply = replyToStrings(reply) + } + + if (command.command === 'hgetall') { + reply = replyToObject(reply) + } + return reply +} + module.exports = { replyToStrings, replyToObject, @@ -136,5 +158,6 @@ module.exports = { monitorRegex: /^[0-9]{10,11}\.[0-9]+ \[[0-9]+ .+]( ".+?")+$/, clone: convenienceClone, replyInOrder, - warn + warn, + handleReply }