You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
502 lines
17 KiB
502 lines
17 KiB
var net = require("net"),
|
|
sys = require("sys"),
|
|
events = require("events"),
|
|
default_port = 6379,
|
|
default_host = "127.0.0.1",
|
|
sym = {},
|
|
inspector = require("eyes").inspector();
|
|
|
|
exports.debug_mode = false;
|
|
|
|
function RedisReplyParser() {
|
|
this.state = "type";
|
|
this.return_buffer = new Buffer(16384);
|
|
this.tmp_buffer = new Buffer(512);
|
|
|
|
events.EventEmitter.call(this);
|
|
}
|
|
sys.inherits(RedisReplyParser, events.EventEmitter);
|
|
|
|
RedisReplyParser.prototype.execute = function (incoming_buf) {
|
|
var pos = 0;
|
|
|
|
while (pos < incoming_buf.length) {
|
|
// console.log("execute " + this.state + " " + pos + " " + String.fromCharCode(incoming_buf[pos]));
|
|
switch (this.state) {
|
|
case "type":
|
|
this.type = incoming_buf[pos];
|
|
pos += 1;
|
|
|
|
switch (this.type) {
|
|
case 43: // +
|
|
this.state = "single line";
|
|
this.return_buffer.end = 0;
|
|
break;
|
|
case 42: // *
|
|
this.state = "multi bulk count";
|
|
this.tmp_buffer.end = 0;
|
|
break;
|
|
case 58: // :
|
|
this.state = "integer line";
|
|
this.return_buffer.end = 0;
|
|
break;
|
|
case 36: // $
|
|
this.state = "bulk length";
|
|
this.tmp_buffer.end = 0;
|
|
break;
|
|
case 45: // -
|
|
this.state = "error line";
|
|
this.return_buffer.end = 0;
|
|
break;
|
|
default:
|
|
this.state = "unknown type";
|
|
}
|
|
break;
|
|
case "integer line":
|
|
if (incoming_buf[pos] === 13) {
|
|
this.send_reply(parseInt(this.return_buffer.slice(0, this.return_buffer.end)),10);
|
|
this.state = "final lf";
|
|
} else {
|
|
this.return_buffer[this.return_buffer.end] = incoming_buf[pos];
|
|
this.return_buffer.end += 1;
|
|
// TODO - check for return_buffer overflow and then grow, copy, continue, and drink.
|
|
}
|
|
pos += 1;
|
|
break;
|
|
case "error line":
|
|
if (incoming_buf[pos] === 13) {
|
|
this.send_error(this.return_buffer.toString("utf8", 0, this.return_buffer.end));
|
|
this.state = "final lf";
|
|
} else {
|
|
this.return_buffer[this.return_buffer.end] = incoming_buf[pos];
|
|
this.return_buffer.end += 1;
|
|
}
|
|
pos += 1;
|
|
break;
|
|
case "single line":
|
|
if (incoming_buf[pos] === 13) {
|
|
this.send_reply(this.return_buffer.toString("utf8", 0, this.return_buffer.end));
|
|
this.state = "final lf";
|
|
} else {
|
|
this.return_buffer[this.return_buffer.end] = incoming_buf[pos];
|
|
this.return_buffer.end += 1;
|
|
// TODO - check for return_buffer overflow and then grow, copy, continue, and drink.
|
|
}
|
|
pos += 1;
|
|
break;
|
|
case "multi bulk count":
|
|
if (incoming_buf[pos] === 13) { // \r
|
|
this.state = "multi bulk count lf";
|
|
} else {
|
|
this.tmp_buffer[this.tmp_buffer.end] = incoming_buf[pos];
|
|
this.tmp_buffer.end += 1;
|
|
}
|
|
pos += 1;
|
|
break;
|
|
case "multi bulk count lf":
|
|
if (incoming_buf[pos] === 10) { // \n
|
|
this.multi_bulk_length = parseInt(this.tmp_buffer.toString("utf8", 0, this.tmp_buffer.end), 10);
|
|
this.multi_bulk_replies = [];
|
|
this.state = "type";
|
|
} else {
|
|
this.emit("error", new Error("didn't see LF after NL reading multi bulk count"));
|
|
this.state = "type"; // try to start over with next data chunk
|
|
return;
|
|
}
|
|
pos += 1;
|
|
break;
|
|
case "bulk length":
|
|
if (incoming_buf[pos] === 13) { // \r
|
|
this.state = "bulk lf";
|
|
} else {
|
|
this.tmp_buffer[this.tmp_buffer.end] = incoming_buf[pos];
|
|
this.tmp_buffer.end += 1;
|
|
}
|
|
pos += 1;
|
|
break;
|
|
case "bulk lf":
|
|
if (incoming_buf[pos] === 10) { // \n
|
|
this.bulk_length = parseInt(this.tmp_buffer.toString("utf8", 0, this.tmp_buffer.end), 10);
|
|
if (this.bulk_length === -1) {
|
|
this.send_reply(null);
|
|
this.state = "type";
|
|
} else {
|
|
this.state = "bulk data";
|
|
if (this.bulk_length > this.return_buffer.length) {
|
|
console.log("Ran out of receive buffer space. Need to fix this.");
|
|
// TODO - fix this
|
|
}
|
|
this.return_buffer.end = 0;
|
|
}
|
|
} else {
|
|
this.emit("error", new Error("didn't see LF after NL while reading bulk length"));
|
|
this.state = "type"; // try to start over with next chunk
|
|
return;
|
|
}
|
|
pos += 1;
|
|
break;
|
|
case "bulk data":
|
|
this.return_buffer[this.return_buffer.end] = incoming_buf[pos];
|
|
this.return_buffer.end += 1;
|
|
pos += 1;
|
|
if (this.return_buffer.end === this.bulk_length) {
|
|
// this ugilness could go away if we gave the user a volatile buffer, but that seems dangerous
|
|
var bd_tmp = new Buffer(this.bulk_length);
|
|
this.return_buffer.copy(bd_tmp, 0, 0, this.bulk_length);
|
|
this.send_reply(bd_tmp);
|
|
this.state = "final cr";
|
|
}
|
|
break;
|
|
case "final cr":
|
|
if (incoming_buf[pos] === 13) { // \r
|
|
this.state = "final lf";
|
|
pos += 1;
|
|
} else {
|
|
this.emit("error", new Error("saw " + incoming_buf[pos] + " when expecting final CR"));
|
|
this.state = "type"; // try to start over with next data chunk
|
|
return;
|
|
}
|
|
break;
|
|
case "final lf":
|
|
if (incoming_buf[pos] === 10) { // \n
|
|
this.state = "type";
|
|
pos += 1;
|
|
} else {
|
|
this.emit("error", new Error("saw " + incoming_buf[pos] + " when expecting final LF"));
|
|
this.state = "type"; // try to start over with next data chunk
|
|
return;
|
|
}
|
|
break;
|
|
default:
|
|
throw new Error("invalid state " + this.state);
|
|
}
|
|
}
|
|
};
|
|
|
|
RedisReplyParser.prototype.send_error = function (reply) {
|
|
if (this.multi_bulk_length > 0) {
|
|
// TODO - can this happen? Seems like maybe not.
|
|
this.add_multi_bulk_reply(reply);
|
|
} else {
|
|
this.emit("reply error", reply);
|
|
}
|
|
}
|
|
|
|
RedisReplyParser.prototype.send_reply = function (reply) {
|
|
if (this.multi_bulk_length > 0) {
|
|
this.add_multi_bulk_reply(reply);
|
|
} else {
|
|
this.emit("reply", reply);
|
|
}
|
|
};
|
|
|
|
RedisReplyParser.prototype.add_multi_bulk_reply = function (reply) {
|
|
this.multi_bulk_replies.push(reply);
|
|
if (this.multi_bulk_replies.length === this.multi_bulk_length) {
|
|
this.emit("reply", this.multi_bulk_replies);
|
|
this.multi_bulk_length = 0;
|
|
this.multi_bulk_replies = null;
|
|
}
|
|
};
|
|
|
|
function RedisClient(stream) {
|
|
events.EventEmitter.call(this);
|
|
|
|
this.stream = stream;
|
|
this.connected = false;
|
|
this.connections = 0;
|
|
this.attempts = 1;
|
|
this.command_queue = [];
|
|
this.commands_sent = 0;
|
|
this.retry_delay = 250;
|
|
this.retry_backoff = 1.7;
|
|
|
|
var self = this;
|
|
|
|
this.stream.on("connect", function () {
|
|
self.connected = true;
|
|
self.connections += 1;
|
|
self.command_queue = [];
|
|
|
|
self.reply_parser = new RedisReplyParser();
|
|
self.reply_parser.on("reply error", function (reply) {
|
|
self.return_error(reply);
|
|
});
|
|
self.reply_parser.on("reply", function (reply) {
|
|
self.return_reply(reply);
|
|
});
|
|
self.reply_parser.on("error", function (err) {
|
|
console.log("Redis reply parser error: " + err.stack);
|
|
});
|
|
|
|
self.retry_delay = 250;
|
|
self.stream.setNoDelay();
|
|
self.stream.setTimeout(0);
|
|
|
|
self.emit("connect");
|
|
});
|
|
|
|
this.stream.on("data", function (buffer_from_socket) {
|
|
self.on_data(buffer_from_socket);
|
|
});
|
|
|
|
this.stream.on("error", function (msg) {
|
|
if (exports.debug_mode) {
|
|
console.warn("Connecting to redis server: " + msg);
|
|
}
|
|
self.connected = false;
|
|
self.emit("error", msg);
|
|
});
|
|
|
|
this.stream.on("close", function () {
|
|
self.connection_gone();
|
|
});
|
|
|
|
this.stream.on("end", function () {
|
|
self.connection_gone();
|
|
});
|
|
|
|
events.EventEmitter.call(this);
|
|
}
|
|
sys.inherits(RedisClient, events.EventEmitter);
|
|
|
|
RedisClient.prototype.connection_gone = function () {
|
|
var self = this;
|
|
|
|
if (self.retry_timer) {
|
|
return;
|
|
}
|
|
|
|
if (exports.debug_mode) {
|
|
console.warn("Redis connection is gone.");
|
|
}
|
|
self.connected = false;
|
|
self.emit("close");
|
|
self.command_queue.forEach(function (args) {
|
|
if (typeof args[2] === "function") {
|
|
args[2]("Server connection closed");
|
|
}
|
|
});
|
|
if (exports.debug_mode) {
|
|
console.log("Retry conneciton in " + self.retry_delay + " ms");
|
|
}
|
|
self.attempts += 1;
|
|
self.emit("reconnecting", "delay " + self.retry_delay + ", attempt " + self.attempts);
|
|
self.retry_timer = setTimeout(function () {
|
|
if (exports.debug_mode) {
|
|
console.log("Retrying conneciton...");
|
|
}
|
|
self.retry_timer = null;
|
|
self.retry_delay = self.retry_delay * self.retry_backoff;
|
|
self.stream.destroy();
|
|
self.stream.connect(self.port, self.host);
|
|
}, self.retry_delay);
|
|
}
|
|
|
|
RedisClient.prototype.on_data = function (data) {
|
|
if (exports.debug_mode) {
|
|
console.log("on_data: " + data.toString());
|
|
}
|
|
|
|
try {
|
|
this.reply_parser.execute(data);
|
|
} catch (err) {
|
|
console.log("Exception in RedisReplyParser: " + err.stack);
|
|
}
|
|
};
|
|
|
|
RedisClient.prototype.return_error = function (err) {
|
|
var command_obj = this.command_queue.shift();
|
|
|
|
if (command_obj && typeof command_obj.callback === "function") {
|
|
command_obj.callback(err);
|
|
} else {
|
|
console.log("no callback to send error: " + err.stack);
|
|
// this will probably not make it anywhere useful, but we might as well try
|
|
throw err;
|
|
}
|
|
}
|
|
|
|
RedisClient.prototype.return_reply = function (reply_buffer) {
|
|
var command_obj = this.command_queue.shift();
|
|
|
|
if (command_obj && typeof command_obj.callback === "function") {
|
|
command_obj.callback(null, reply_buffer);
|
|
} else {
|
|
if (this.debug_mode) {
|
|
console.log("no callback for reply: " + reply_buffer.toString());
|
|
}
|
|
}
|
|
};
|
|
|
|
RedisClient.prototype.send_command = function (command, args, callback) {
|
|
if (! command) {
|
|
throw new Error("First argument of send_command must be the command name");
|
|
return;
|
|
}
|
|
|
|
if (! Array.isArray(args)) {
|
|
throw new Error("Second argument of send_command must an array of arguments");
|
|
return;
|
|
}
|
|
|
|
if (callback !== undefined && typeof callback !== "function") {
|
|
throw new Error("Third argument of send_command must a results callback function, or omitted");
|
|
return;
|
|
}
|
|
|
|
if (! this.connected) {
|
|
callback(new Error("Redis client is not connected"));
|
|
return;
|
|
}
|
|
|
|
this.command_queue.push({
|
|
command: command,
|
|
args: args,
|
|
callback: callback
|
|
});
|
|
this.commands_sent += 1;
|
|
|
|
var elem_count = 1, stream = this.stream, buffer_args = false, command_str = "";
|
|
|
|
elem_count += args.length;
|
|
buffer_args = args.some(function (arg) {
|
|
return arg instanceof Buffer;
|
|
});
|
|
|
|
// Always use "Multi bulk commands", but if passed any Buffer args, then do multiple writes, one for each arg
|
|
|
|
command_str = "*" + elem_count + "\r\n$" + command.length + "\r\n" + command + "\r\n";
|
|
|
|
if (! buffer_args) { // Build up a string and send entire command in one write
|
|
args.forEach(function (arg) {
|
|
if (typeof arg !== "string") {
|
|
arg = String(arg);
|
|
}
|
|
command_str += "$" + arg.length + "\r\n" + arg + "\r\n";
|
|
});
|
|
if (exports.debug_mode) {
|
|
console.log("send command: " + command_str);
|
|
}
|
|
stream.write(command_str);
|
|
} else {
|
|
if (exports.debug_mode) {
|
|
console.log("send command: " + command_str);
|
|
console.log("send command has Buffer arguments");
|
|
}
|
|
stream.write(command_str);
|
|
|
|
args.forEach(function (arg) {
|
|
if (arg.length === undefined) {
|
|
arg = String(number);
|
|
}
|
|
|
|
if (arg instanceof Buffer) {
|
|
stream.write("$" + arg.length + "\r\n")
|
|
stream.write(arg);
|
|
stream.write("\r\n");
|
|
} else {
|
|
stream.write("$" + arg.length + "\r\n" + arg + "\r\n");
|
|
}
|
|
});
|
|
};
|
|
};
|
|
|
|
// http://code.google.com/p/redis/wiki/CommandReference
|
|
exports.commands = [
|
|
// Commands operating on all value types
|
|
"EXISTS", "DEL", "TYPE", "KEYS", "RANDOMKEY", "RENAME", "RENAMENX", "DBSIZE", "EXPIRE", "TTL", "SELECT",
|
|
"MOVE", "FLUSHDB", "FLUSHALL",
|
|
// Commands operating on string values
|
|
"SET", "GET", "GETSET", "MGET", "SETNX", "SETEX", "MSET", "MSETNX", "INCR", "INCRBY", "DECR", "DECRBY", "APPEND", "SUBSTR",
|
|
// Commands operating on lists
|
|
"RPUSH", "LPUSH", "LLEN", "LRANGE", "LTRIM", "LINDEX", "LSET", "LREM", "LPOP", "RPOP", "BLPOP", "BRPOP", "RPOPLPUSH",
|
|
// Commands operating on sets
|
|
"SADD", "SREM", "SPOP", "SMOVE", "SCARD", "SISMEMBER", "SINTER", "SINTERSTORE", "SUNION", "SUNIONSTORE", "SDIFF", "SDIFFSTORE",
|
|
"SMEMBERS", "SRANDMEMBER",
|
|
// Commands operating on sorted zsets (sorted sets)
|
|
"ZADD", "ZREM", "ZINCRBY", "ZRANK", "ZREVRANK", "ZRANGE", "ZREVRANGE", "ZRANGEBYSCORE", "ZCOUNT", "ZCARD", "ZSCORE",
|
|
"ZREMRANGEBYRANK", "ZREMRANGEBYSCORE", "ZUNIONSTORE", "ZINTERSTORE",
|
|
// Commands operating on hashes
|
|
"HSET", "HGET", "HMGET", "HMSET", "HINCRBY", "HEXISTS", "HDEL", "HLEN", "HKEYS", "HVALS", "HGETALL",
|
|
// Sorting
|
|
"SORT",
|
|
// Persistence control commands
|
|
"SAVE", "BGSAVE", "LASTSAVE", "SHUTDOWN", "BGREWRITEAOF",
|
|
// Remote server control commands
|
|
"INFO", "MONITOR", "SLAVEOF", "CONFIG"
|
|
];
|
|
|
|
exports.commands.forEach(function (command) {
|
|
RedisClient.prototype[command] = function (args, callback) {
|
|
this.send_command(command, args, callback)
|
|
};
|
|
RedisClient.prototype[command.toLowerCase()] = function (args, callback) {
|
|
this.send_command(command, args, callback)
|
|
};
|
|
});
|
|
|
|
// Transactions
|
|
// "MULTI", "EXEC", "DISCARD", "WATCH", "UNWATCH",
|
|
// Publish/Subscribe
|
|
// "SUBSCRIBE", "UNSUBSCRIBE", "PUBLISH",
|
|
|
|
RedisClient.prototype.multi = function (commands) {
|
|
var self = this;
|
|
|
|
try {
|
|
this.send_command("MULTI", [], function (err, reply) {
|
|
if (err) {
|
|
console.warn("Error starting MULTI request: " + err.stack);
|
|
}
|
|
});
|
|
commands.forEach(function (args, command_num) {
|
|
self.send_command(args[0], args[1], function (err, reply) {
|
|
if (err) {
|
|
args[2](err);
|
|
commands.splice(command_num, 1); // what if this runs before all commands are sent?
|
|
} else {
|
|
if (reply !== "QUEUED") {
|
|
console.warn("Unexpected MULTI reply: " + reply + " instead of 'QUEUED'");
|
|
}
|
|
}
|
|
});
|
|
});
|
|
this.send_command("EXEC", [], function (err, replies) {
|
|
replies.forEach(function (reply, reply_num) {
|
|
if (typeof commands[reply_num][2] === "function") {
|
|
commands[reply_num][2](null, reply)
|
|
} else {
|
|
if (exports.debug_mode) {
|
|
console.log("no callback for multi response " + reply_num + ", skipping.");
|
|
}
|
|
}
|
|
});
|
|
});
|
|
} catch (err) {
|
|
console.log("Caught exception in MULTI: " + err.stack);
|
|
}
|
|
};
|
|
|
|
exports.createClient = function (port_arg, host_arg, options) {
|
|
var port = port_arg || default_port,
|
|
host = host || default_host,
|
|
red_client, net_client;
|
|
|
|
net_client = net.createConnection(port, host)
|
|
|
|
red_client = new RedisClient(net_client);
|
|
|
|
red_client.port = port;
|
|
red_client.host = host;
|
|
|
|
return red_client;
|
|
};
|
|
|
|
exports.print = function (err, reply) {
|
|
if (err) {
|
|
console.log("Error: " + err);
|
|
} else {
|
|
console.log("Reply: " + reply);
|
|
}
|
|
};
|
|
|