File size: 1,850 Bytes
5fae594 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
var fs = require('fs')
, Stream = require('stream').Stream
, clarinet = require('../clarinet')
, parse_stream = clarinet.createStream()
, previous = ''
, buffer = {}
, stack = []
, new_thing = false
;
function debug_log() {
if(process.env.DEBUG) {
console.log.apply(null, arguments);
}
}
parse_stream.on('openobject', function(name) {
if(new_thing) {
console.log(JSON.stringify(buffer, null, 2));
buffer = {};
new_thing = false;
}
previous = name;
stack.push(name);
debug_log('=== {', name, buffer);
});
parse_stream.on('closeobject', function() {
stack.pop();
debug_log('=== }', null, buffer);
});
parse_stream.on('key', function(name) {
previous = name;
stack.pop();
stack.push(name);
debug_log('=== ,', name, buffer);
});
parse_stream.on('value', function(value) {
if(previous === 'event') {
value = JSON.parse(value);
}
var expected = stack.length-1;
stack.reduce(function (ac, x, i) {
if(i === expected) {
ac[x] = value;
}
ac[x] = ac[x] || {};
return ac[x];
}, buffer);
debug_log('=== v', value, buffer);
});
parse_stream.on('error', function (e) {
new_thing = true;
});
function fixLogglyStream() {
var log_stream = new Stream();
log_stream.readable = true;
log_stream.writable = true;
log_stream.write = function (buf) {
var as_string = buf.toString('utf-8').replace(/\\\\/g, '\\');
this.emit('data', as_string);
};
log_stream.end = function (buf) {
if (arguments.length) {
log_stream.write(buf);
}
log_stream.writable = false;
};
log_stream.destroy = function () {
log_stream.writable = false;
};
return log_stream;
}
fs.createReadStream(__dirname + '/loggly.txt')
.pipe(fixLogglyStream())
.pipe(parse_stream)
; |