You've already forked node-redis
mirror of
https://github.com/redis/node-redis.git
synced 2025-08-04 15:02:09 +03:00
Move parsers into seperate module and improve js parser performance
This commit is contained in:
Ruben Bridgewater
70
README.md
70
README.md
@@ -653,43 +653,43 @@ hiredis parser (Lenovo T450s i7-5600U):
|
||||
|
||||
```
|
||||
Client count: 1, node version: 4.2.2, server version: 3.0.3, parser: hiredis
|
||||
PING, 1/1 min/max/avg/p95: 0/ 2/ 0.02/ 0.00 2501ms total, 47503.80 ops/sec
|
||||
PING, batch 50/1 min/max/avg/p95: 0/ 2/ 0.09/ 1.00 2501ms total, 529668.13 ops/sec
|
||||
SET 4B str, 1/1 min/max/avg/p95: 0/ 2/ 0.02/ 0.00 2501ms total, 41900.04 ops/sec
|
||||
SET 4B str, batch 50/1 min/max/avg/p95: 0/ 2/ 0.14/ 1.00 2501ms total, 354658.14 ops/sec
|
||||
SET 4B buf, 1/1 min/max/avg/p95: 0/ 4/ 0.04/ 0.00 2501ms total, 23499.00 ops/sec
|
||||
SET 4B buf, batch 50/1 min/max/avg/p95: 0/ 2/ 0.31/ 1.00 2501ms total, 159836.07 ops/sec
|
||||
GET 4B str, 1/1 min/max/avg/p95: 0/ 4/ 0.02/ 0.00 2501ms total, 43489.80 ops/sec
|
||||
GET 4B str, batch 50/1 min/max/avg/p95: 0/ 2/ 0.11/ 1.00 2501ms total, 444202.32 ops/sec
|
||||
GET 4B buf, 1/1 min/max/avg/p95: 0/ 3/ 0.02/ 0.00 2501ms total, 38561.38 ops/sec
|
||||
GET 4B buf, batch 50/1 min/max/avg/p95: 0/ 2/ 0.11/ 1.00 2501ms total, 452139.14 ops/sec
|
||||
SET 4KiB str, 1/1 min/max/avg/p95: 0/ 2/ 0.03/ 0.00 2501ms total, 32990.80 ops/sec
|
||||
SET 4KiB str, batch 50/1 min/max/avg/p95: 0/ 2/ 0.34/ 1.00 2501ms total, 146161.54 ops/sec
|
||||
SET 4KiB buf, 1/1 min/max/avg/p95: 0/ 1/ 0.04/ 0.00 2501ms total, 23294.28 ops/sec
|
||||
SET 4KiB buf, batch 50/1 min/max/avg/p95: 0/ 2/ 0.36/ 1.00 2501ms total, 137584.97 ops/sec
|
||||
GET 4KiB str, 1/1 min/max/avg/p95: 0/ 2/ 0.03/ 0.00 2501ms total, 36350.66 ops/sec
|
||||
GET 4KiB str, batch 50/1 min/max/avg/p95: 0/ 2/ 0.32/ 1.00 2501ms total, 155157.94 ops/sec
|
||||
GET 4KiB buf, 1/1 min/max/avg/p95: 0/ 4/ 0.02/ 0.00 2501ms total, 39776.49 ops/sec
|
||||
GET 4KiB buf, batch 50/1 min/max/avg/p95: 0/ 2/ 0.32/ 1.00 2501ms total, 155457.82 ops/sec
|
||||
INCR, 1/1 min/max/avg/p95: 0/ 3/ 0.02/ 0.00 2501ms total, 43972.41 ops/sec
|
||||
INCR, batch 50/1 min/max/avg/p95: 0/ 1/ 0.12/ 1.00 2501ms total, 425809.68 ops/sec
|
||||
LPUSH, 1/1 min/max/avg/p95: 0/ 2/ 0.02/ 0.00 2501ms total, 38998.40 ops/sec
|
||||
LPUSH, batch 50/1 min/max/avg/p95: 0/ 4/ 0.14/ 1.00 2501ms total, 365013.99 ops/sec
|
||||
LRANGE 10, 1/1 min/max/avg/p95: 0/ 2/ 0.03/ 0.00 2501ms total, 31879.25 ops/sec
|
||||
LRANGE 10, batch 50/1 min/max/avg/p95: 0/ 1/ 0.32/ 1.00 2501ms total, 153698.52 ops/sec
|
||||
LRANGE 100, 1/1 min/max/avg/p95: 0/ 4/ 0.06/ 0.00 2501ms total, 16676.13 ops/sec
|
||||
LRANGE 100, batch 50/1 min/max/avg/p95: 1/ 6/ 2.03/ 2.00 2502ms total, 24520.38 ops/sec
|
||||
SET 4MiB str, 1/1 min/max/avg/p95: 1/ 6/ 2.11/ 3.00 2502ms total, 472.82 ops/sec
|
||||
SET 4MiB str, batch 20/1 min/max/avg/p95: 85/ 112/ 94.93/ 109.60 2563ms total, 210.69 ops/sec
|
||||
SET 4MiB buf, 1/1 min/max/avg/p95: 1/ 8/ 2.02/ 3.00 2502ms total, 490.01 ops/sec
|
||||
SET 4MiB buf, batch 20/1 min/max/avg/p95: 37/ 52/ 39.48/ 46.75 2528ms total, 506.33 ops/sec
|
||||
GET 4MiB str, 1/1 min/max/avg/p95: 3/ 13/ 5.26/ 9.00 2504ms total, 190.10 ops/sec
|
||||
GET 4MiB str, batch 20/1 min/max/avg/p95: 70/ 106/ 89.36/ 103.75 2503ms total, 223.73 ops/sec
|
||||
GET 4MiB buf, 1/1 min/max/avg/p95: 3/ 11/ 5.04/ 8.15 2502ms total, 198.24 ops/sec
|
||||
GET 4MiB buf, batch 20/1 min/max/avg/p95: 70/ 105/ 88.07/ 103.00 2554ms total, 227.09 ops/sec
|
||||
PING, 1/1 min/max/avg: 0/ 2/ 0.02/ 2501ms total, 47503.80 ops/sec
|
||||
PING, batch 50/1 min/max/avg: 0/ 2/ 0.09/ 2501ms total, 529668.13 ops/sec
|
||||
SET 4B str, 1/1 min/max/avg: 0/ 2/ 0.02/ 2501ms total, 41900.04 ops/sec
|
||||
SET 4B str, batch 50/1 min/max/avg: 0/ 2/ 0.14/ 2501ms total, 354658.14 ops/sec
|
||||
SET 4B buf, 1/1 min/max/avg: 0/ 4/ 0.04/ 2501ms total, 23499.00 ops/sec
|
||||
SET 4B buf, batch 50/1 min/max/avg: 0/ 2/ 0.31/ 2501ms total, 159836.07 ops/sec
|
||||
GET 4B str, 1/1 min/max/avg: 0/ 4/ 0.02/ 2501ms total, 43489.80 ops/sec
|
||||
GET 4B str, batch 50/1 min/max/avg: 0/ 2/ 0.11/ 2501ms total, 444202.32 ops/sec
|
||||
GET 4B buf, 1/1 min/max/avg: 0/ 3/ 0.02/ 2501ms total, 38561.38 ops/sec
|
||||
GET 4B buf, batch 50/1 min/max/avg: 0/ 2/ 0.11/ 2501ms total, 452139.14 ops/sec
|
||||
SET 4KiB str, 1/1 min/max/avg: 0/ 2/ 0.03/ 2501ms total, 32990.80 ops/sec
|
||||
SET 4KiB str, batch 50/1 min/max/avg: 0/ 2/ 0.34/ 2501ms total, 146161.54 ops/sec
|
||||
SET 4KiB buf, 1/1 min/max/avg: 0/ 1/ 0.04/ 2501ms total, 23294.28 ops/sec
|
||||
SET 4KiB buf, batch 50/1 min/max/avg: 0/ 2/ 0.36/ 2501ms total, 137584.97 ops/sec
|
||||
GET 4KiB str, 1/1 min/max/avg: 0/ 2/ 0.03/ 2501ms total, 36350.66 ops/sec
|
||||
GET 4KiB str, batch 50/1 min/max/avg: 0/ 2/ 0.32/ 2501ms total, 155157.94 ops/sec
|
||||
GET 4KiB buf, 1/1 min/max/avg: 0/ 4/ 0.02/ 2501ms total, 39776.49 ops/sec
|
||||
GET 4KiB buf, batch 50/1 min/max/avg: 0/ 2/ 0.32/ 2501ms total, 155457.82 ops/sec
|
||||
INCR, 1/1 min/max/avg: 0/ 3/ 0.02/ 2501ms total, 43972.41 ops/sec
|
||||
INCR, batch 50/1 min/max/avg: 0/ 1/ 0.12/ 2501ms total, 425809.68 ops/sec
|
||||
LPUSH, 1/1 min/max/avg: 0/ 2/ 0.02/ 2501ms total, 38998.40 ops/sec
|
||||
LPUSH, batch 50/1 min/max/avg: 0/ 4/ 0.14/ 2501ms total, 365013.99 ops/sec
|
||||
LRANGE 10, 1/1 min/max/avg: 0/ 2/ 0.03/ 2501ms total, 31879.25 ops/sec
|
||||
LRANGE 10, batch 50/1 min/max/avg: 0/ 1/ 0.32/ 2501ms total, 153698.52 ops/sec
|
||||
LRANGE 100, 1/1 min/max/avg: 0/ 4/ 0.06/ 2501ms total, 16676.13 ops/sec
|
||||
LRANGE 100, batch 50/1 min/max/avg: 1/ 6/ 2.03/ 2502ms total, 24520.38 ops/sec
|
||||
SET 4MiB str, 1/1 min/max/avg: 1/ 6/ 2.11/ 2502ms total, 472.82 ops/sec
|
||||
SET 4MiB str, batch 20/1 min/max/avg: 85/ 112/ 94.93/ 2563ms total, 210.69 ops/sec
|
||||
SET 4MiB buf, 1/1 min/max/avg: 1/ 8/ 2.02/ 2502ms total, 490.01 ops/sec
|
||||
SET 4MiB buf, batch 20/1 min/max/avg: 37/ 52/ 39.48/ 2528ms total, 506.33 ops/sec
|
||||
GET 4MiB str, 1/1 min/max/avg: 3/ 13/ 5.26/ 2504ms total, 190.10 ops/sec
|
||||
GET 4MiB str, batch 20/1 min/max/avg: 70/ 106/ 89.36/ 2503ms total, 223.73 ops/sec
|
||||
GET 4MiB buf, 1/1 min/max/avg: 3/ 11/ 5.04/ 2502ms total, 198.24 ops/sec
|
||||
GET 4MiB buf, batch 20/1 min/max/avg: 70/ 105/ 88.07/ 2554ms total, 227.09 ops/sec
|
||||
```
|
||||
|
||||
The hiredis and js parser should most of the time be on the same level. But if you use Redis for big SUNION/SINTER/LRANGE/ZRANGE hiredis is significantly faster.
|
||||
The hiredis and js parser should most of the time be on the same level. But if you use Redis for big SUNION/SINTER/LRANGE/ZRANGE hiredis is faster.
|
||||
Therefor the hiredis parser is the default used in node_redis. To use `hiredis`, do:
|
||||
|
||||
npm install hiredis redis
|
||||
|
||||
@@ -42,7 +42,7 @@ function lpad(input, len, chr) {
|
||||
|
||||
metrics.Histogram.prototype.print_line = function () {
|
||||
var obj = this.printObj();
|
||||
return lpad(obj.min, 4) + '/' + lpad(obj.max, 4) + '/' + lpad(obj.mean.toFixed(2), 7) + '/' + lpad(obj.p95.toFixed(2), 7);
|
||||
return lpad(obj.min, 4) + '/' + lpad(obj.max, 4) + '/' + lpad(obj.mean.toFixed(2), 7);
|
||||
};
|
||||
|
||||
function Test(args) {
|
||||
@@ -205,7 +205,7 @@ Test.prototype.print_stats = function () {
|
||||
var duration = Date.now() - this.test_start;
|
||||
totalTime += duration;
|
||||
|
||||
console.log('min/max/avg/p95: ' + this.command_latency.print_line() + ' ' + lpad(duration, 6) + 'ms total, ' +
|
||||
console.log('min/max/avg: ' + this.command_latency.print_line() + ' ' + lpad(duration, 6) + 'ms total, ' +
|
||||
lpad((this.commands_completed / (duration / 1000)).toFixed(2), 9) + ' ops/sec');
|
||||
};
|
||||
|
||||
|
||||
@@ -1,6 +1,13 @@
|
||||
Changelog
|
||||
=========
|
||||
|
||||
## v.2.5.0 - xx Dez, 2015
|
||||
|
||||
Features
|
||||
|
||||
- The parsers moved into the [redis-parser](https://github.com/NodeRedis/node-redis-parser) module and will be maintained in there from now on ([@BridgeAR](https://github.com/BridgeAR))
|
||||
- Improve js parser speed significantly for big SUNION/SINTER/LRANGE/ZRANGE ([@BridgeAR](https://github.com/BridgeAR))
|
||||
|
||||
## v.2.4.2 - 27 Nov, 2015
|
||||
|
||||
Bugfixes
|
||||
|
||||
117
index.js
117
index.js
@@ -8,7 +8,7 @@ var utils = require('./lib/utils');
|
||||
var Queue = require('double-ended-queue');
|
||||
var Command = require('./lib/command');
|
||||
var events = require('events');
|
||||
var parsers = [];
|
||||
var Parser = require('redis-parser');
|
||||
var commands = require('redis-commands');
|
||||
var connection_id = 0;
|
||||
var default_port = 6379;
|
||||
@@ -18,17 +18,20 @@ function noop () {}
|
||||
function clone (obj) { return JSON.parse(JSON.stringify(obj || {})); }
|
||||
function debug (msg) { if (exports.debug_mode) { console.error(msg); } }
|
||||
|
||||
exports.debug_mode = /\bredis\b/i.test(process.env.NODE_DEBUG);
|
||||
function handle_detect_buffers_reply (reply, command, buffer_args) {
|
||||
if (buffer_args === false) {
|
||||
// If detect_buffers 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.reply_to_strings(reply);
|
||||
}
|
||||
|
||||
// Hiredis might not be installed
|
||||
try {
|
||||
parsers.push(require('./lib/parsers/hiredis'));
|
||||
} catch (err) {
|
||||
/* istanbul ignore next: won't be reached with tests */
|
||||
debug('Hiredis parser not installed.');
|
||||
if (command === 'hgetall') {
|
||||
reply = utils.reply_to_object(reply);
|
||||
}
|
||||
return reply;
|
||||
}
|
||||
|
||||
parsers.push(require('./lib/parsers/javascript'));
|
||||
exports.debug_mode = /\bredis\b/i.test(process.env.NODE_DEBUG);
|
||||
|
||||
function RedisClient (options) {
|
||||
// Copy the options so they are not mutated
|
||||
@@ -69,6 +72,10 @@ function RedisClient (options) {
|
||||
console.warn('>> WARNING: You activated return_buffers and detect_buffers at the same time. The return value is always going to be a buffer.');
|
||||
options.detect_buffers = false;
|
||||
}
|
||||
if (options.detect_buffers) {
|
||||
// We only need to look at the arguments if we do not know what we have to return
|
||||
this.handle_reply = handle_detect_buffers_reply;
|
||||
}
|
||||
this.should_buffer = false;
|
||||
this.max_attempts = options.max_attempts | 0;
|
||||
this.command_queue = new Queue(); // Holds sent commands to de-pipeline them
|
||||
@@ -83,13 +90,22 @@ function RedisClient (options) {
|
||||
this.closing = false;
|
||||
this.server_info = {};
|
||||
this.auth_pass = options.auth_pass;
|
||||
this.parser_module = null;
|
||||
this.selected_db = null; // Save the selected db here, used when reconnecting
|
||||
this.old_state = null;
|
||||
this.pipeline = 0;
|
||||
this.options = options;
|
||||
// Init parser once per instance
|
||||
this.init_parser();
|
||||
// Init parser
|
||||
var self = this;
|
||||
this.reply_parser = new Parser({
|
||||
returnReply: function (data) {
|
||||
self.return_reply(data);
|
||||
},
|
||||
returnError: function (data) {
|
||||
self.return_error(data);
|
||||
},
|
||||
returnBuffers: options.return_buffers || options.detect_buffers,
|
||||
name: options.parser
|
||||
});
|
||||
this.create_stream();
|
||||
}
|
||||
util.inherits(RedisClient, events.EventEmitter);
|
||||
@@ -153,6 +169,13 @@ RedisClient.prototype.create_stream = function () {
|
||||
});
|
||||
};
|
||||
|
||||
RedisClient.prototype.handle_reply = function (reply, command) {
|
||||
if (command === 'hgetall') {
|
||||
reply = utils.reply_to_object(reply);
|
||||
}
|
||||
return reply;
|
||||
};
|
||||
|
||||
RedisClient.prototype.cork = noop;
|
||||
RedisClient.prototype.uncork = noop;
|
||||
|
||||
@@ -300,39 +323,6 @@ RedisClient.prototype.on_connect = function () {
|
||||
}
|
||||
};
|
||||
|
||||
RedisClient.prototype.init_parser = function () {
|
||||
var self = this;
|
||||
|
||||
if (this.options.parser) {
|
||||
if (!parsers.some(function (parser) {
|
||||
if (parser.name === self.options.parser) {
|
||||
self.parser_module = parser;
|
||||
debug('Using parser module: ' + self.parser_module.name);
|
||||
return true;
|
||||
}
|
||||
})) {
|
||||
// Do not emit this error
|
||||
// This should take down the app if anyone made such a huge mistake or should somehow be handled in user code
|
||||
throw new Error("Couldn't find named parser " + self.options.parser + " on this system");
|
||||
}
|
||||
} else {
|
||||
debug('Using default parser module: ' + parsers[0].name);
|
||||
this.parser_module = parsers[0];
|
||||
}
|
||||
|
||||
// return_buffers sends back Buffers from parser to callback. detect_buffers sends back Buffers from parser, but
|
||||
// converts to Strings if the input arguments are not Buffers.
|
||||
this.reply_parser = new this.parser_module.Parser(self.options.return_buffers || self.options.detect_buffers);
|
||||
// Important: Only send results / errors async.
|
||||
// That way the result / error won't stay in a try catch block and catch user things
|
||||
this.reply_parser.send_error = function (data) {
|
||||
self.return_error(data);
|
||||
};
|
||||
this.reply_parser.send_reply = function (data) {
|
||||
self.return_reply(data);
|
||||
};
|
||||
};
|
||||
|
||||
RedisClient.prototype.on_ready = function () {
|
||||
var self = this;
|
||||
|
||||
@@ -599,7 +589,7 @@ RedisClient.prototype.return_error = function (err) {
|
||||
err.command = command_obj.command;
|
||||
}
|
||||
|
||||
var match = err.message.match(utils.errCode);
|
||||
var match = err.message.match(utils.err_code);
|
||||
// LUA script could return user errors that don't behave like all other errors!
|
||||
if (match) {
|
||||
err.code = match[1];
|
||||
@@ -650,16 +640,7 @@ RedisClient.prototype.return_reply = function (reply) {
|
||||
if (command_obj && !command_obj.sub_command) {
|
||||
if (typeof command_obj.callback === 'function') {
|
||||
if ('exec' !== command_obj.command) {
|
||||
if (command_obj.buffer_args === false) {
|
||||
// If detect_buffers option was specified, then the reply from the parser will be Buffers.
|
||||
// If this command did not use Buffer arguments, then convert the reply to Strings here.
|
||||
reply = utils.reply_to_strings(reply);
|
||||
}
|
||||
|
||||
// TODO - confusing and error-prone that hgetall is special cased in two places
|
||||
if ('hgetall' === command_obj.command) {
|
||||
reply = utils.reply_to_object(reply);
|
||||
}
|
||||
reply = this.handle_reply(reply, command_obj.command, command_obj.buffer_args);
|
||||
}
|
||||
command_obj.callback(null, reply);
|
||||
} else {
|
||||
@@ -722,8 +703,7 @@ RedisClient.prototype.send_command = function (command, args, callback) {
|
||||
command_str = '',
|
||||
buffer_args = false,
|
||||
big_data = false,
|
||||
prefix_keys,
|
||||
buffer = this.options.return_buffers;
|
||||
prefix_keys;
|
||||
|
||||
if (args === undefined) {
|
||||
args = [];
|
||||
@@ -770,11 +750,8 @@ RedisClient.prototype.send_command = function (command, args, callback) {
|
||||
}
|
||||
}
|
||||
}
|
||||
if (this.options.detect_buffers) {
|
||||
buffer = buffer_args;
|
||||
}
|
||||
|
||||
command_obj = new Command(command, args, false, buffer, callback);
|
||||
command_obj = new Command(command, args, false, buffer_args, callback);
|
||||
|
||||
if (!this.ready && !this.send_anyway || !stream.writable) {
|
||||
if (this.closing || !this.enable_offline_queue) {
|
||||
@@ -1149,11 +1126,7 @@ Multi.prototype.exec_transaction = function (callback) {
|
||||
cb = undefined;
|
||||
}
|
||||
// Keep track of who wants buffer responses:
|
||||
if (this._client.options.return_buffers) {
|
||||
this.wants_buffers[index] = true;
|
||||
} else if (!this._client.options.detect_buffers) {
|
||||
this.wants_buffers[index] = false;
|
||||
} else {
|
||||
if (this._client.options.detect_buffers) {
|
||||
this.wants_buffers[index] = false;
|
||||
for (var i = 0; i < args.length; i += 1) {
|
||||
if (Buffer.isBuffer(args[i])) {
|
||||
@@ -1193,20 +1166,14 @@ Multi.prototype.execute_callback = function (err, replies) {
|
||||
while (args = this.queue.shift()) {
|
||||
// If we asked for strings, even in detect_buffers mode, then return strings:
|
||||
if (replies[i] instanceof Error) {
|
||||
var match = replies[i].message.match(utils.errCode);
|
||||
var match = replies[i].message.match(utils.err_code);
|
||||
// LUA script could return user errors that don't behave like all other errors!
|
||||
if (match) {
|
||||
replies[i].code = match[1];
|
||||
}
|
||||
replies[i].command = args[0].toUpperCase();
|
||||
} else if (replies[i]) {
|
||||
if (this.wants_buffers[i] === false) {
|
||||
replies[i] = utils.reply_to_strings(replies[i]);
|
||||
}
|
||||
if (args[0] === 'hgetall') {
|
||||
// TODO - confusing and error-prone that hgetall is special cased in two places
|
||||
replies[i] = utils.reply_to_object(replies[i]);
|
||||
}
|
||||
replies[i] = this._client.handle_reply(replies[i], args[0], this.wants_buffers[i]);
|
||||
}
|
||||
|
||||
if (typeof args[args.length - 1] === 'function') {
|
||||
|
||||
@@ -1,37 +0,0 @@
|
||||
'use strict';
|
||||
|
||||
var hiredis = require('hiredis');
|
||||
|
||||
function HiredisReplyParser(return_buffers) {
|
||||
this.name = exports.name;
|
||||
this.reader = new hiredis.Reader({
|
||||
return_buffers: return_buffers
|
||||
});
|
||||
}
|
||||
|
||||
HiredisReplyParser.prototype.return_data = function () {
|
||||
try {
|
||||
return this.reader.get();
|
||||
} catch (err) {
|
||||
// Protocol errors land here
|
||||
this.send_error(err);
|
||||
return void 0;
|
||||
}
|
||||
};
|
||||
|
||||
HiredisReplyParser.prototype.execute = function (data) {
|
||||
this.reader.feed(data);
|
||||
var reply = this.return_data();
|
||||
|
||||
while (reply !== undefined) {
|
||||
if (reply && reply.name === 'Error') {
|
||||
this.send_error(reply);
|
||||
} else {
|
||||
this.send_reply(reply);
|
||||
}
|
||||
reply = this.return_data();
|
||||
}
|
||||
};
|
||||
|
||||
exports.Parser = HiredisReplyParser;
|
||||
exports.name = 'hiredis';
|
||||
@@ -1,167 +0,0 @@
|
||||
'use strict';
|
||||
|
||||
var util = require('util');
|
||||
|
||||
function JavascriptReplyParser() {
|
||||
this.name = exports.name;
|
||||
this._buffer = new Buffer(0);
|
||||
this._offset = 0;
|
||||
this._big_str_size = 0;
|
||||
this._chunks_size = 0;
|
||||
this._buffers = [];
|
||||
this._type = 0;
|
||||
this._protocol_error = false;
|
||||
}
|
||||
|
||||
function IncompleteReadBuffer(message) {
|
||||
this.name = 'IncompleteReadBuffer';
|
||||
this.message = message;
|
||||
}
|
||||
util.inherits(IncompleteReadBuffer, Error);
|
||||
|
||||
JavascriptReplyParser.prototype._parseResult = function (type) {
|
||||
var start = 0,
|
||||
end = 0,
|
||||
offset = 0,
|
||||
packetHeader = 0,
|
||||
res,
|
||||
reply;
|
||||
|
||||
if (type === 43 || type === 58 || type === 45) { // + or : or -
|
||||
// Up to the delimiter
|
||||
end = this._packetEndOffset();
|
||||
start = this._offset;
|
||||
// Include the delimiter
|
||||
this._offset = end + 2;
|
||||
|
||||
if (type === 43) {
|
||||
return this._buffer.slice(start, end);
|
||||
} else if (type === 58) {
|
||||
// Return the coerced numeric value
|
||||
return +this._buffer.toString('ascii', start, end);
|
||||
}
|
||||
return new Error(this._buffer.toString('utf-8', start, end));
|
||||
} else if (type === 36) { // $
|
||||
packetHeader = this.parseHeader();
|
||||
|
||||
// Packets with a size of -1 are considered null
|
||||
if (packetHeader === -1) {
|
||||
return null;
|
||||
}
|
||||
end = this._offset + packetHeader;
|
||||
start = this._offset;
|
||||
|
||||
if (end + 2 > this._buffer.length) {
|
||||
this._chunks_size = this._buffer.length - this._offset - 2;
|
||||
this._big_str_size = packetHeader;
|
||||
throw new IncompleteReadBuffer('Wait for more data.');
|
||||
}
|
||||
// Set the offset to after the delimiter
|
||||
this._offset = end + 2;
|
||||
|
||||
return this._buffer.slice(start, end);
|
||||
} else if (type === 42) { // *
|
||||
// Set a rewind point, as the packet is larger than the buffer in memory
|
||||
offset = this._offset;
|
||||
packetHeader = this.parseHeader();
|
||||
|
||||
if (packetHeader === -1) {
|
||||
return null;
|
||||
}
|
||||
reply = [];
|
||||
offset = this._offset - 1;
|
||||
|
||||
for (var i = 0; i < packetHeader; i++) {
|
||||
if (this._offset >= this._buffer.length) {
|
||||
throw new IncompleteReadBuffer('Wait for more data.');
|
||||
}
|
||||
res = this._parseResult(this._buffer[this._offset++]);
|
||||
reply.push(res);
|
||||
}
|
||||
return reply;
|
||||
} else {
|
||||
return void 0;
|
||||
}
|
||||
};
|
||||
|
||||
JavascriptReplyParser.prototype.execute = function (buffer) {
|
||||
if (this._chunks_size !== 0) {
|
||||
if (this._big_str_size > this._chunks_size + buffer.length) {
|
||||
this._buffers.push(buffer);
|
||||
this._chunks_size += buffer.length;
|
||||
return;
|
||||
}
|
||||
this._buffers.unshift(this._offset === 0 ? this._buffer : this._buffer.slice(this._offset));
|
||||
this._buffers.push(buffer);
|
||||
this._buffer = Buffer.concat(this._buffers);
|
||||
this._buffers = [];
|
||||
this._big_str_size = 0;
|
||||
this._chunks_size = 0;
|
||||
} else if (this._offset >= this._buffer.length) {
|
||||
this._buffer = buffer;
|
||||
} else {
|
||||
this._buffer = Buffer.concat([this._buffer.slice(this._offset), buffer]);
|
||||
}
|
||||
this._offset = 0;
|
||||
this._protocol_error = true;
|
||||
this.run();
|
||||
};
|
||||
|
||||
JavascriptReplyParser.prototype.try_parsing = function () {
|
||||
// Set a rewind point. If a failure occurs, wait for the next execute()/append() and try again
|
||||
var offset = this._offset - 1;
|
||||
try {
|
||||
return this._parseResult(this._type);
|
||||
} catch (err) {
|
||||
// Catch the error (not enough data), rewind if it's an array,
|
||||
// and wait for the next packet to appear
|
||||
this._offset = offset;
|
||||
this._protocol_error = false;
|
||||
return void 0;
|
||||
}
|
||||
};
|
||||
|
||||
JavascriptReplyParser.prototype.run = function (buffer) {
|
||||
this._type = this._buffer[this._offset++];
|
||||
var reply = this.try_parsing();
|
||||
|
||||
while (reply !== undefined) {
|
||||
if (this._type === 45) { // Errors -
|
||||
this.send_error(reply);
|
||||
} else {
|
||||
this.send_reply(reply); // Strings + // Integers : // Bulk strings $ // Arrays *
|
||||
}
|
||||
this._type = this._buffer[this._offset++];
|
||||
reply = this.try_parsing();
|
||||
}
|
||||
if (this._type !== undefined && this._protocol_error === true) {
|
||||
// Reset the buffer so the parser can handle following commands properly
|
||||
this._buffer = new Buffer(0);
|
||||
this.send_error(new Error('Protocol error, got ' + JSON.stringify(String.fromCharCode(this._type)) + ' as reply type byte'));
|
||||
}
|
||||
};
|
||||
|
||||
JavascriptReplyParser.prototype.parseHeader = function () {
|
||||
var end = this._packetEndOffset(),
|
||||
value = this._buffer.toString('ascii', this._offset, end) | 0;
|
||||
|
||||
this._offset = end + 2;
|
||||
return value;
|
||||
};
|
||||
|
||||
JavascriptReplyParser.prototype._packetEndOffset = function () {
|
||||
var offset = this._offset,
|
||||
len = this._buffer.length - 1;
|
||||
|
||||
while (this._buffer[offset] !== 0x0d && this._buffer[offset + 1] !== 0x0a) {
|
||||
offset++;
|
||||
|
||||
if (offset >= len) {
|
||||
throw new IncompleteReadBuffer('Did not see LF after NL reading multi bulk count (' + offset + ' => ' + this._buffer.length + ', ' + this._offset + ')');
|
||||
}
|
||||
}
|
||||
return offset;
|
||||
};
|
||||
|
||||
exports.Parser = JavascriptReplyParser;
|
||||
exports.name = 'javascript';
|
||||
@@ -62,5 +62,5 @@ module.exports = {
|
||||
reply_to_object: replyToObject,
|
||||
to_array: toArray,
|
||||
print: print,
|
||||
errCode: redisErrCode
|
||||
err_code: redisErrCode
|
||||
};
|
||||
|
||||
@@ -8,7 +8,8 @@
|
||||
"transaction",
|
||||
"pipelining",
|
||||
"performance",
|
||||
"queue"
|
||||
"queue",
|
||||
"nodejs"
|
||||
],
|
||||
"author": "Matt Ranney <mjr@ranney.com>",
|
||||
"license": "MIT",
|
||||
@@ -23,7 +24,8 @@
|
||||
},
|
||||
"dependencies": {
|
||||
"double-ended-queue": "^2.1.0-0",
|
||||
"redis-commands": "^1.0.1"
|
||||
"redis-commands": "^1.0.1",
|
||||
"redis-parser": "^1.0.0"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=0.10.0"
|
||||
|
||||
@@ -8,20 +8,6 @@ var redis = config.redis;
|
||||
|
||||
describe("The node_redis client", function () {
|
||||
|
||||
describe("testing parser existence", function () {
|
||||
it('throws on non-existence', function (done) {
|
||||
try {
|
||||
redis.createClient({
|
||||
parser: 'nonExistingParser'
|
||||
});
|
||||
done(new Error('test failed'));
|
||||
} catch (err) {
|
||||
assert.equal(err.message, 'Couldn\'t find named parser nonExistingParser on this system');
|
||||
done();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
helper.allTests({
|
||||
allConnections: true
|
||||
}, function(parser, ip, args) {
|
||||
|
||||
@@ -1,317 +0,0 @@
|
||||
'use strict';
|
||||
|
||||
var assert = require('assert');
|
||||
var utils = require("../lib/utils");
|
||||
var parsers = [
|
||||
require("../lib/parsers/javascript").Parser
|
||||
];
|
||||
try {
|
||||
// Test the hiredis parser if available
|
||||
parsers.push(require("../lib/parsers/hiredis").Parser);
|
||||
} catch (e) {}
|
||||
|
||||
describe('parsers', function () {
|
||||
|
||||
parsers.forEach(function (Parser) {
|
||||
|
||||
describe(Parser.name, function () {
|
||||
|
||||
it('handles multi-bulk reply', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
function check_reply(reply) {
|
||||
reply = utils.reply_to_strings(reply);
|
||||
assert.deepEqual(reply, [['a']], "Expecting multi-bulk reply of [['a']]");
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer('*1\r\n*1\r\n$1\r\na\r\n'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
|
||||
parser.execute(new Buffer('*1\r\n*1\r'));
|
||||
parser.execute(new Buffer('\n$1\r\na\r\n'));
|
||||
assert.strictEqual(reply_count, 2);
|
||||
|
||||
parser.execute(new Buffer('*1\r\n*1\r\n'));
|
||||
parser.execute(new Buffer('$1\r\na\r\n'));
|
||||
|
||||
assert.equal(reply_count, 3, "check reply should have been called three times");
|
||||
});
|
||||
|
||||
it('parser error', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
function check_reply (reply) {
|
||||
assert.strictEqual(reply.message, 'Protocol error, got "a" as reply type byte');
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_error = check_reply;
|
||||
|
||||
parser.execute(new Buffer('a*1\r*1\r$1`zasd\r\na'));
|
||||
assert.equal(reply_count, 1);
|
||||
});
|
||||
|
||||
it('parser error v2', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
var err_count = 0;
|
||||
function check_reply (reply) {
|
||||
reply = utils.reply_to_strings(reply);
|
||||
assert.strictEqual(reply[0], 'OK');
|
||||
reply_count++;
|
||||
}
|
||||
function check_error (err) {
|
||||
assert.strictEqual(err.message, 'Protocol error, got "b" as reply type byte');
|
||||
err_count++;
|
||||
}
|
||||
parser.send_error = check_error;
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer('*1\r\n+OK\r\nb$1`zasd\r\na'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
assert.strictEqual(err_count, 1);
|
||||
});
|
||||
|
||||
it('parser error v3', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
var err_count = 0;
|
||||
function check_reply (reply) {
|
||||
reply = utils.reply_to_strings(reply);
|
||||
assert.strictEqual(reply[0], 'OK');
|
||||
reply_count++;
|
||||
}
|
||||
function check_error (err) {
|
||||
assert.strictEqual(err.message, 'Protocol error, got "\\n" as reply type byte');
|
||||
err_count++;
|
||||
}
|
||||
parser.send_error = check_error;
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer('*1\r\n+OK\r\n\n+zasd\r\n'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
assert.strictEqual(err_count, 1);
|
||||
});
|
||||
|
||||
it('should handle \\r and \\n characters properly', function () {
|
||||
// If a string contains \r or \n characters it will always be send as a bulk string
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
var entries = ['foo\r', 'foo\r\nbar', '\r\nfoo', 'foo\r\n'];
|
||||
function check_reply (reply) {
|
||||
reply = utils.reply_to_strings(reply);
|
||||
assert.strictEqual(reply, entries[reply_count]);
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer('$4\r\nfoo\r\r\n$8\r\nfoo\r\nbar\r\n$5\r\n\r\n'));
|
||||
assert.strictEqual(reply_count, 2);
|
||||
parser.execute(new Buffer('foo\r\n$5\r\nfoo\r\n\r\n'));
|
||||
assert.strictEqual(reply_count, 4);
|
||||
});
|
||||
|
||||
it('line breaks in the beginning of the last chunk', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
function check_reply(reply) {
|
||||
reply = utils.reply_to_strings(reply);
|
||||
assert.deepEqual(reply, [['a']], "Expecting multi-bulk reply of [['a']]");
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer('*1\r\n*1\r\n$1\r\na'));
|
||||
assert.equal(reply_count, 0);
|
||||
|
||||
parser.execute(new Buffer('\r\n*1\r\n*1\r'));
|
||||
assert.equal(reply_count, 1);
|
||||
parser.execute(new Buffer('\n$1\r\na\r\n*1\r\n*1\r\n$1\r\na\r\n'));
|
||||
|
||||
assert.equal(reply_count, 3, "check reply should have been called three times");
|
||||
});
|
||||
|
||||
it('multiple chunks in a bulk string', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
function check_reply(reply) {
|
||||
reply = utils.reply_to_strings(reply);
|
||||
assert.strictEqual(reply, "abcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghij");
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer('$100\r\nabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij'));
|
||||
assert.strictEqual(reply_count, 0);
|
||||
parser.execute(new Buffer('\r\n'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
|
||||
parser.execute(new Buffer('$100\r'));
|
||||
parser.execute(new Buffer('\nabcdefghijabcdefghijabcdefghijabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghij'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
parser.execute(new Buffer(
|
||||
'abcdefghij\r\n' +
|
||||
'$100\r\nabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghij\r\n' +
|
||||
'$100\r\nabcdefghijabcdefghijabcdefghijabcdefghij'
|
||||
));
|
||||
assert.strictEqual(reply_count, 3);
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij\r'));
|
||||
assert.strictEqual(reply_count, 3);
|
||||
parser.execute(new Buffer('\n'));
|
||||
|
||||
assert.equal(reply_count, 4, "check reply should have been called three times");
|
||||
});
|
||||
|
||||
it('multiple chunks with arrays different types', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
var predefined_data = [
|
||||
'abcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghijabcdefghij',
|
||||
'test',
|
||||
100,
|
||||
new Error('Error message'),
|
||||
['The force awakens']
|
||||
];
|
||||
function check_reply(reply) {
|
||||
reply = utils.reply_to_strings(reply);
|
||||
for (var i = 0; i < reply.length; i++) {
|
||||
if (i < 3) {
|
||||
assert.strictEqual(reply[i], predefined_data[i]);
|
||||
} else {
|
||||
assert.deepEqual(reply[i], predefined_data[i]);
|
||||
}
|
||||
}
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer('*5\r\n$100\r\nabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij'));
|
||||
parser.execute(new Buffer('abcdefghijabcdefghijabcdefghij\r\n'));
|
||||
parser.execute(new Buffer('+test\r'));
|
||||
parser.execute(new Buffer('\n:100'));
|
||||
parser.execute(new Buffer('\r\n-Error message'));
|
||||
parser.execute(new Buffer('\r\n*1\r\n$17\r\nThe force'));
|
||||
assert.strictEqual(reply_count, 0);
|
||||
parser.execute(new Buffer(' awakens\r\n$5'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
});
|
||||
|
||||
it('return normal errors', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
function check_reply(reply) {
|
||||
assert.equal(reply.message, 'Error message');
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_error = check_reply;
|
||||
|
||||
parser.execute(new Buffer('-Error '));
|
||||
parser.execute(new Buffer('message\r\n*3\r\n$17\r\nThe force'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
parser.execute(new Buffer(' awakens\r\n$5'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
});
|
||||
|
||||
it('return null for empty arrays and empty bulk strings', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
function check_reply(reply) {
|
||||
assert.equal(reply, null);
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer('$-1\r\n*-'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
parser.execute(new Buffer('1'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
parser.execute(new Buffer('\r\n$-'));
|
||||
assert.strictEqual(reply_count, 2);
|
||||
});
|
||||
|
||||
it('return value even if all chunks are only 1 character long', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
function check_reply(reply) {
|
||||
assert.equal(reply, 1);
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer(':'));
|
||||
assert.strictEqual(reply_count, 0);
|
||||
parser.execute(new Buffer('1'));
|
||||
assert.strictEqual(reply_count, 0);
|
||||
parser.execute(new Buffer('\r'));
|
||||
assert.strictEqual(reply_count, 0);
|
||||
parser.execute(new Buffer('\n'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
});
|
||||
|
||||
it('do not return before \\r\\n', function () {
|
||||
var parser = new Parser();
|
||||
var reply_count = 0;
|
||||
function check_reply(reply) {
|
||||
assert.equal(reply, 1);
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
|
||||
parser.execute(new Buffer(':1\r\n:'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
parser.execute(new Buffer('1'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
parser.execute(new Buffer('\r'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
parser.execute(new Buffer('\n'));
|
||||
assert.strictEqual(reply_count, 2);
|
||||
});
|
||||
|
||||
it('return data as buffer if requested', function () {
|
||||
var parser = new Parser(true);
|
||||
var reply_count = 0;
|
||||
function check_reply(reply) {
|
||||
if (Array.isArray(reply)) {
|
||||
reply = reply[0];
|
||||
}
|
||||
assert(Buffer.isBuffer(reply));
|
||||
assert.strictEqual(reply.inspect(), new Buffer('test').inspect());
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
parser.execute(new Buffer('+test\r\n'));
|
||||
assert.strictEqual(reply_count, 1);
|
||||
parser.execute(new Buffer('$4\r\ntest\r\n'));
|
||||
assert.strictEqual(reply_count, 2);
|
||||
parser.execute(new Buffer('*1\r\n$4\r\ntest\r\n'));
|
||||
assert.strictEqual(reply_count, 3);
|
||||
});
|
||||
|
||||
it('regression test v.2.4.1', function () {
|
||||
var parser = new Parser(true);
|
||||
var reply_count = 0;
|
||||
var entries = ['test test ', 'test test test test ', '1234'];
|
||||
function check_reply(reply) {
|
||||
assert.strictEqual(reply.toString(), entries[reply_count]);
|
||||
reply_count++;
|
||||
}
|
||||
parser.send_reply = check_reply;
|
||||
parser.execute(new Buffer('$10\r\ntest '));
|
||||
assert.strictEqual(reply_count, 0);
|
||||
parser.execute(new Buffer('test \r\n$20\r\ntest test test test \r\n:1234\r'));
|
||||
assert.strictEqual(reply_count, 2);
|
||||
parser.execute(new Buffer('\n'));
|
||||
assert.strictEqual(reply_count, 3);
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user