another pub/sub strategy

tests/stress/codec.js

@@ -3,12 +3,14 @@ var json = {
 	decode: JSON.parse
 };
 
-/*var msgpack = require('node-msgpack');
+var MsgPack = require('node-msgpack');
 msgpack = {
-	encode: msgpack.pack,
-	decode: msgpack.unpack
-};*/
+	encode: MsgPack.pack,
+	decode: function(str) { return MsgPack.unpack(new Buffer(str)); }
+};
 
 bison = require('bison');
 
 module.exports = json;
+//module.exports = msgpack;
+module.exports = bison;

tests/stress/pubsub/pub.js

@@ -6,8 +6,8 @@ var codec = require('../codec');
 var sent = 0;
 
 var pub = require('redis').createClient(null, null, {
-	command_queue_high_water: 5,
-	command_queue_low_water: 1
+	//command_queue_high_water: 5,
+	//command_queue_low_water: 1
 })
 .on('ready', function() {
 	this.emit('drain');

tests/stress/rpushblpop/pub.js

@@ -1,13 +1,39 @@
 'use strict';
 
+var freemem = require('os').freemem;
 var codec = require('../codec');
 
-var pub = require('redis').createClient()
-	.on('ready', function() {
-		while (true) {
-			pub.rpush('timeline', codec.encode({
-				cmd: Math.random(),
-				data: Math.random()
-			}));
-		}
+var sent = 0;
+
+var pub = require('redis').createClient(null, null, {
+	//command_queue_high_water: 5,
+	//command_queue_low_water: 1
+})
+.on('ready', function() {
+	this.del('timeline');
+	this.emit('drain');
+})
+.on('drain', function() {
+	process.nextTick(exec);
+});
+
+var payload = '1'; for (var i = 0; i < 10; ++i) payload += payload;
+
+function exec() {
+	pub.rpush('timeline', codec.encode({ foo: payload }));
+	++sent;
+	if (!pub.should_buffer) {
+		process.nextTick(exec);
+	}
+}
+
+exec();
+
+setInterval(function() {
+	pub.llen('timeline', function(err, result) {
+		console.log('sent', sent, 'free', freemem(),
+			'cmdqlen', pub.command_queue.length, 'offqlen', pub.offline_queue.length,
+			'llen', result
+		);
 	});
+}, 1000);