Parse log lines individually

lib/Game.js

@@ -3,7 +3,8 @@ var events   = require("events"),
     spawn    = require("child_process").spawn,
     util     = require("util"),
     _        = require("lodash"),
-    patterns = require("./patterns");
+    patterns = require("./patterns"),
+    byline = require("./byline");
 
 /**
  * Create an instance of a minecraft server. Each instance can spawn a process
@@ -16,11 +17,29 @@ var events   = require("events"),
  * @param {Object} options
  */
 function Game(dir, jar, options) {
+    var game = this;
     _.extend(this, options);
     this.dir = dir;
     this.jar = jar;
     this.players = [];
 
+    this.mcTxtStream = new byline.LineStream({encoding: 'utf8'});
+    this.mcTxtStream.on("data", function (data) {
+        var java = "";
+        game.log += data;
+        if (game.debug) {
+            console.log(data);
+        }
+
+        if (/^Error/.test(data) || /^Exception/.test(data)) {
+            game.emit("error", new Error(data));
+        } else if (/^java/.test(data) || /^\s+/.test(data)) {
+            game.emit("java", java);
+        } else {
+            Game.emitLog(game, data);
+        }
+    });
+
     events.EventEmitter.call(this);
     //this.setMaxListeners(20);
 }
@@ -66,28 +85,9 @@ Game.prototype.start = function (callback) {
     this.process = spawn(this.java, this.args, { cwd: this.dir });
 
     this.process.stderr.setEncoding("utf8");
-    this.process.stderr.on("data", function (data) {
-        var java = "";
-        game.log += data;
-        if (game.debug) {
-            console.log(data.trim());
-        }
-
-        if (/^Error/.test(data) || /^Exception/.test(data)) {
-            game.emit("error", new Error(data));
-        } else if (/^java/.test(data) || /^\s+/.test(data)) {
-            java += data;
-
-            _.debounce(function () {
-                java = "";
-                //console.warn(java);
-                game.emit("java", java);
-            }, 500);
-        } else {
-            Game.emitLog(game, data);
-        }
-    });
 
+    this.process.stderr.pipe(this.mcTxtStream);
+
     this.process.once("exit", function (code) {
         game.status = "Stopped";
         game.process = null;
@@ -170,6 +170,7 @@ Game.parseLog = function (line) {
 };
 
 Game.emitLog = function (game, log) {
+
     var meta = Game.parseLog(log);
 
     game.emit("log", meta);

lib/byline.js

@@ -0,0 +1,129 @@
+// Copyright (C) 2011-2013 John Hewson
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to
+// deal in the Software without restriction, including without limitation the
+// rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+// sell copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+// 
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+// 
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+// IN THE SOFTWARE.
+
+var stream = require('stream'),
+    util = require('util');
+
+// convinience API
+module.exports = function(readStream, options) {
+  return module.exports.createStream(readStream, options);
+};
+
+// basic API
+module.exports.createStream = function(readStream, options) {
+  if (readStream) {
+    return createLineStream(readStream, options);
+  } else {
+    return new LineStream(options);
+  }
+};
+
+// deprecated API
+module.exports.createLineStream = function(readStream) {
+  console.log('WARNING: byline#createLineStream is deprecated and will be removed soon');
+  return createLineStream(readStream);
+};
+
+function createLineStream(readStream, options) {
+  if (!readStream) {
+    throw new Error('expected readStream');
+  }
+  if (!readStream.readable) {
+    throw new Error('readStream must be readable');
+  }
+  var ls = new LineStream(options);
+  readStream.pipe(ls);
+  return ls;
+}
+
+//
+// using the new node v0.10 "streams2" API
+//
+
+module.exports.LineStream = LineStream;
+
+function LineStream(options) {
+  stream.Transform.call(this, options);
+  // use objectMode to stop the output from being buffered
+  // which re-concatanates the lines, just without newlines.
+  this._readableState.objectMode = true;
+  this._lineBuffer = [];
+}
+util.inherits(LineStream, stream.Transform);
+
+LineStream.prototype._transform = function(chunk, encoding, done) {
+  // decode binary chunks as UTF-8
+  encoding = encoding || 'utf8';
+
+  if (Buffer.isBuffer(chunk)) {
+    if (encoding == 'buffer') {
+      chunk = chunk.toString(); // utf8
+      encoding = 'utf8';
+    }
+    else {
+     chunk = chunk.toString(encoding);
+    }
+  }
+  this._encoding = encoding;
+
+  var lines = chunk.split(/\r\n|\r|\n/g);
+
+  if (this._lineBuffer.length > 0) {
+    this._lineBuffer[this._lineBuffer.length - 1] += lines[0];
+    lines.shift();
+  }
+
+  this._lineBuffer = this._lineBuffer.concat(lines);
+
+  // always buffer the last (possibly partial) line
+  while (this._lineBuffer.length > 1) {
+    var line = this._lineBuffer.shift();
+    // skip empty lines
+    if (line.length > 0 ) {
+      if (!this.push(this._reencode(line))) {
+        break;
+      }
+    }
+  }
+
+  done();
+};
+
+LineStream.prototype._flush = function(done) {
+  // flush all buffered lines
+  while (this._lineBuffer.length > 0) {
+    var line = this._lineBuffer.shift();
+    // skip empty lines
+    if (line.length > 0 ) {
+      if (!this.push(this._reencode(line))) {
+        break;
+      }
+    }
+  }
+  done();
+};
+
+// see Readable::push
+LineStream.prototype._reencode = function(chunk, encoding) {
+  if (encoding !== this._readableState.encoding) {
+    chunk = new Buffer(chunk, encoding);
+  }
+  return chunk;
+};

test/Game.js

@@ -145,6 +145,47 @@ describe("Game", function () {
         });
     });
 
+    describe("mc text stream", function () {
+        afterEach(function () {
+            game.removeAllListeners();
+        });
+
+        it("parses log events by line", function(done){
+            var count = 0;
+            function checkDone() {
+                count++;
+                if(count == 2) done();
+                if(count > 2) done('Too many calls');
+            };
+            game.on("left", checkDone);
+            game.on("joined", checkDone);
+            game.mcTxtStream.write("2013-09-04 20:33:12 [INFO] heneryville[/192.168.1.101:1158] logged in with entity id 33 at (559.5, 36.0, 85.5)\r\n" +
+                "2013-09-04 20:33:12 [INFO] heneryville joined the game\r\n" +
+                "2013-09-04 20:33:20 [INFO] heneryville lost connection: disconnect.quitting\r\n" +
+                "2013-09-04 20:33:20 [INFO] heneryville left the game\r\n"
+            );
+        });
+
+        it("parses mixed events by line", function(done){
+            var count = 0;
+            function checkDone() {
+                count++;
+                if(count == 6) done();
+                if(count > 6) done('Too many calls');
+            };
+            game.on("java", checkDone);
+            game.mcTxtStream.write("2013-09-07 18:32:55 [INFO] Saving playersi\r\n"
+            + "java.net.SocketException: Socket closed\r\n"
+            + "\tat java.net.PlainSocketImpl.socketAccept(Native Method)\r\n"
+            + "\tat java.net.AbstractPlainSocketImpl.accept(AbstractPlainSocketImpl.java:375)\r\n"
+            + "\tat java.net.ServerSocket.implAccept(ServerSocket.java:478)\r\n"
+            + "\tat java.net.ServerSocket.accept(ServerSocket.java:446)\r\n"
+            + "\tat ix.run(SourceFile:61)\r\n"
+            + "2013-09-07 18:32:55 [INFO] Closing listening thread);\r\n"
+            );
+        });
+    });
+
     describe("#start()", function () {
         this.timeout("15s");