UTF8 fixes and ns.Stream emitter

lib/ns.js

@@ -2,6 +2,8 @@
 
 var assert = require('assert');
 var Buffer = require('buffer').Buffer;
+var events = require('events');
+var sys    = require('sys');
 
 // Get the length of the netstring payload (i.e. excluding header and footer)
 // pointed to by Buffer or String 'buf'. Returns -1 if the buffer is
@@ -63,6 +65,10 @@ exports.nsLength = nsLength;
 var nsPayload = function(buf, off) {
     off = off || 0;
 
+    if (typeof buf === 'string') {
+        buf = new Buffer(buf.substring(0, buf.length));
+    }
+
     var len = nsPayloadLength(buf, off);
     if (len < 0) {
         return len;
@@ -77,11 +83,7 @@ var nsPayload = function(buf, off) {
 
     var start = off + (nsLen - len - 1);
 
-    if (typeof buf === 'string') {
-        return buf.substring(start, start + len);
-    } else if (typeof buf === 'object') {
-        return buf.slice(start, start + len);
-    }
+    return buf.slice(start, start + len);
 };
 exports.nsPayload = nsPayload;
 
@@ -126,7 +128,7 @@ var nsWrite = function(pay, payStart, payEnd, buf, bufOff) {
     payEnd = (payEnd === undefined) ? pay.length : payEnd;
     bufOff = bufOff || 0;
 
-    if (payStart < 0 || payStart >= pay.length) {
+    if (payStart < 0 || payStart > pay.length) {
         throw new Error('payStart is out of bounds');
     }
 
@@ -172,3 +174,44 @@ var nsWrite = function(pay, payStart, payEnd, buf, bufOff) {
     }
 };
 exports.nsWrite = nsWrite;
+
+var Stream = function(s) {
+    var self = this;
+
+    events.EventEmitter.call(self);
+
+    self.buf = null;
+
+    s.addListener('data', function(d) {
+        if (self.buf) {
+            var b = new Buffer(self.buf.length + d.length);
+            self.buf.copy(b, 0, 0, self.buf.length);
+            d.copy(b, self.buf.length, 0, d.length);
+
+            self.buf = b;
+        } else {
+            self.buf = d;
+        }
+
+        while (self.buf && self.buf.length > 0) {
+            try {
+                pay = nsPayload(self.buf);
+
+                if (pay == -1) {
+                    break;
+                }
+
+                var nsLen = nsWriteLength(pay.length);
+                self.buf = self.buf.slice(nsLen, self.buf.length);
+
+                self.emit('data', pay);
+            } catch (exception) {
+                self.emit('error', exception);
+                break;
+            }
+        }
+    });
+}
+
+sys.inherits(Stream, events.EventEmitter);
+exports.Stream = Stream;

package.json

@@ -0,0 +1,22 @@
+{
+    "name" : "netstring",
+    "version" : "0.1.0",
+    "description" : "A netstring implementation",
+    "author" : "Peter Griess <pg@std.in>",
+    "engines" : {
+        "node" : ">=0.1.98"
+    },
+    "repositories" : [
+        {
+            "type" : "git",
+            "url" : "http://github.com/pgriess/node-netstring"
+        }
+    ],
+    "licenses" : [
+        {
+            "type" : "BSD",
+            "url" : "http://github.com/pgriess/node-netstring/blob/master/LICENSE"
+        }
+    ],
+    "main" : "./lib/ns.js"
+}

test/test.js

@@ -2,6 +2,7 @@
 
 var at = require('./async_testing');
 var Buffer = require('buffer').Buffer;
+var events = require('events');
 var ns = require('../lib/ns');
 
 (function() {
@@ -20,13 +21,18 @@ var ns = require('../lib/ns');
             as.equal(ns.nsPayloadLength('30:'), 30);
         },
         'incomplete' : function(as) {
+            as.equal(ns.nsPayloadLength(''), -1);
             as.equal(ns.nsPayloadLength('30'), -1);
         },
         'leading zero' : function(as) {
             as.equal(ns.nsPayloadLength('0:,'), 0);
             ex(as, '03:', 'Invalid netstring with leading 0');
             ex(as, '00:', 'Invalid netstring with leading 0');
         },
+        'leading colon' : function(as) {
+            ex(as, ':', 'Invalid netstring with leading \':\'');
+            ex(as, ':a', 'Invalid netstring with leading \':\'');
+        },
         'invalid char' : function(as) {
             ex(as, '3;', 'Unexpected character \';\' found at offset 1');
             ex(as, ';', 'Unexpected character \';\' found at offset 0');
@@ -50,6 +56,7 @@ var ns = require('../lib/ns');
             as.equal(ns.nsLength('30:'), 34);
         },
         'incomplete' : function(as) {
+            as.equal(ns.nsLength(''), -1);
             as.equal(ns.nsLength('30'), -1);
         },
     });
@@ -77,14 +84,20 @@ var ns = require('../lib/ns');
 
     ts.addTests({
         'simple' : function(as) {
+            as.equal(ns.nsPayload('0:,'), '');
             as.equal(ns.nsPayload('3:abc,'), 'abc');
             as.equal(ns.nsPayload(new Buffer('3:abc,')), 'abc');
         },
         'incomplete' : function(as) {
+            as.equal(ns.nsPayload(''), -1);
             as.equal(ns.nsPayload('3'), -1);
             as.equal(ns.nsPayload('3:ab'), -1);
             as.equal(ns.nsPayload('3:abc'), -1);
         },
+        'utf8' : function(as) {
+          as.equal(ns.nsPayload('3:☃,'), '☃');
+          as.equal(ns.nsPayload(new Buffer('3:☃,')), '☃');
+        }
     });
 
     ts.runTests();
@@ -119,9 +132,14 @@ var ns = require('../lib/ns');
 
     ts.addTests({
         'simple' : function(as) {
+            as.equal(ns.nsWrite(''), '0:,');
             as.equal(ns.nsWrite('abc'), '3:abc,');
             as.equal(ns.nsWrite('a'), '1:a,');
         },
+        'utf8' : function(as) {
+          as.equal(ns.nsWrite('☃'), '3:☃,');
+          as.equal(ns.nsWrite(new Buffer('☃')), '3:☃,');
+        },
         'start' : function(as) {
             as.equal(ns.nsWrite('abc', 1), '2:bc,');
             as.equal(ns.nsWrite('abc', 0), '3:abc,');
@@ -149,3 +167,37 @@ var ns = require('../lib/ns');
 
     ts.runTests();
 })();
+
+(function() {
+    var ts = new at.TestSuite('Stream');
+
+    ts.addTests({
+        'simple' : function(as) {
+            var is  = new events.EventEmitter();
+            var ins = new ns.Stream(is);
+
+            var MSGS = [
+                "abc",
+                "hello world!",
+                "café",
+                "a",
+                "b",
+                "c"
+            ];
+
+            var msgsReceived = 0;
+            ins.addListener('data', function(d) {
+                as.equal(d.toString(), MSGS[msgsReceived]);
+                msgsReceived++;
+            });
+
+            is.emit('data', new Buffer("3:abc,"));
+            is.emit('data', new Buffer("12:hello"));
+            is.emit('data', new Buffer(" world!,"));
+            is.emit('data', new Buffer("5:café,"));
+            is.emit('data', new Buffer("1:a,1:b,1:c,"));
+        }
+    });
+
+    ts.runTests();
+})();