123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270 |
- /*!
- * socket.io-node
- * Copyright(c) 2011 LearnBoost <dev@learnboost.com>
- * MIT Licensed
- */
- /**
- * Module dependencies.
- */
- var crypto = require('crypto')
- , Store = require('../store')
- , assert = require('assert');
- /**
- * Exports the constructor.
- */
- exports = module.exports = Redis;
- Redis.Client = Client;
- /**
- * Redis store.
- * Options:
- * - nodeId (fn) gets an id that uniquely identifies this node
- * - redis (fn) redis constructor, defaults to redis
- * - redisPub (object) options to pass to the pub redis client
- * - redisSub (object) options to pass to the sub redis client
- * - redisClient (object) options to pass to the general redis client
- * - pack (fn) custom packing, defaults to JSON or msgpack if installed
- * - unpack (fn) custom packing, defaults to JSON or msgpack if installed
- *
- * @api public
- */
- function Redis (opts) {
- opts = opts || {};
- // node id to uniquely identify this node
- var nodeId = opts.nodeId || function () {
- // by default, we generate a random id
- return Math.abs(Math.random() * Math.random() * Date.now() | 0);
- };
- this.nodeId = nodeId();
- // packing / unpacking mechanism
- if (opts.pack) {
- this.pack = opts.pack;
- this.unpack = opts.unpack;
- } else {
- try {
- var msgpack = require('msgpack');
- this.pack = msgpack.pack;
- this.unpack = msgpack.unpack;
- } catch (e) {
- this.pack = JSON.stringify;
- this.unpack = JSON.parse;
- }
- }
- var redis = opts.redis || require('redis')
- , RedisClient = redis.RedisClient;
- // initialize a pubsub client and a regular client
- if (opts.redisPub instanceof RedisClient) {
- this.pub = opts.redisPub;
- } else {
- opts.redisPub || (opts.redisPub = {});
- this.pub = redis.createClient(opts.redisPub.port, opts.redisPub.host, opts.redisPub);
- }
- if (opts.redisSub instanceof RedisClient) {
- this.sub = opts.redisSub;
- } else {
- opts.redisSub || (opts.redisSub = {});
- this.sub = redis.createClient(opts.redisSub.port, opts.redisSub.host, opts.redisSub);
- }
- if (opts.redisClient instanceof RedisClient) {
- this.cmd = opts.redisClient;
- } else {
- opts.redisClient || (opts.redisClient = {});
- this.cmd = redis.createClient(opts.redisClient.port, opts.redisClient.host, opts.redisClient);
- }
- Store.call(this, opts);
- this.sub.setMaxListeners(0);
- this.setMaxListeners(0);
- };
- /**
- * Inherits from Store.
- */
- Redis.prototype.__proto__ = Store.prototype;
- /**
- * Publishes a message.
- *
- * @api private
- */
- Redis.prototype.publish = function (name) {
- var args = Array.prototype.slice.call(arguments, 1);
- this.pub.publish(name, this.pack({ nodeId: this.nodeId, args: args }));
- this.emit.apply(this, ['publish', name].concat(args));
- };
- /**
- * Subscribes to a channel
- *
- * @api private
- */
- Redis.prototype.subscribe = function (name, consumer, fn) {
- this.sub.subscribe(name);
- if (consumer || fn) {
- var self = this;
- self.sub.on('subscribe', function subscribe (ch) {
- if (name == ch) {
- function message (ch, msg) {
- if (name == ch) {
- msg = self.unpack(msg);
- // we check that the message consumed wasnt emitted by this node
- if (self.nodeId != msg.nodeId) {
- consumer.apply(null, msg.args);
- }
- }
- };
- self.sub.on('message', message);
- self.on('unsubscribe', function unsubscribe (ch) {
- if (name == ch) {
- self.sub.removeListener('message', message);
- self.removeListener('unsubscribe', unsubscribe);
- }
- });
- self.sub.removeListener('subscribe', subscribe);
- fn && fn();
- }
- });
- }
- this.emit('subscribe', name, consumer, fn);
- };
- /**
- * Unsubscribes
- *
- * @api private
- */
- Redis.prototype.unsubscribe = function (name, fn) {
- this.sub.unsubscribe(name);
- if (fn) {
- var client = this.sub;
- client.on('unsubscribe', function unsubscribe (ch) {
- if (name == ch) {
- fn();
- client.removeListener('unsubscribe', unsubscribe);
- }
- });
- }
- this.emit('unsubscribe', name, fn);
- };
- /**
- * Destroys the store
- *
- * @api public
- */
- Redis.prototype.destroy = function () {
- Store.prototype.destroy.call(this);
- this.pub.end();
- this.sub.end();
- this.cmd.end();
- };
- /**
- * Client constructor
- *
- * @api private
- */
- function Client (store, id) {
- Store.Client.call(this, store, id);
- };
- /**
- * Inherits from Store.Client
- */
- Client.prototype.__proto__ = Store.Client;
- /**
- * Redis hash get
- *
- * @api private
- */
- Client.prototype.get = function (key, fn) {
- this.store.cmd.hget(this.id, key, fn);
- return this;
- };
- /**
- * Redis hash set
- *
- * @api private
- */
- Client.prototype.set = function (key, value, fn) {
- this.store.cmd.hset(this.id, key, value, fn);
- return this;
- };
- /**
- * Redis hash del
- *
- * @api private
- */
- Client.prototype.del = function (key, fn) {
- this.store.cmd.hdel(this.id, key, fn);
- return this;
- };
- /**
- * Redis hash has
- *
- * @api private
- */
- Client.prototype.has = function (key, fn) {
- this.store.cmd.hexists(this.id, key, function (err, has) {
- if (err) return fn(err);
- fn(null, !!has);
- });
- return this;
- };
- /**
- * Destroys client
- *
- * @param {Number} number of seconds to expire data
- * @api private
- */
- Client.prototype.destroy = function (expiration) {
- if ('number' != typeof expiration) {
- this.store.cmd.del(this.id);
- } else {
- this.store.cmd.expire(this.id, expiration);
- }
- return this;
- };
|