From 62d309764a1e2769b815b220b2ed64a8364866ab Mon Sep 17 00:00:00 2001 From: Rafael Sobral Date: Mon, 25 Apr 2016 16:38:56 -0300 Subject: [PATCH 1/2] Added support for Interleave Implementation works like decoder.on('packet',function(packet){ console.log('packet channel:',packet.channel); console.log('packet size:',packet.size); var bufs = []; packet.on('data',function(data){ bufs.push(data); }); packet.on('end',function(){ var payload = Buffer.concat(bufs); console.log('packet payload:',payload); }); }); --- decoder.js | 95 ++++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 93 insertions(+), 2 deletions(-) diff --git a/decoder.js b/decoder.js index 248988a..e53480f 100644 --- a/decoder.js +++ b/decoder.js @@ -6,19 +6,29 @@ var IncomingMessage = require('./lib/incoming-message') var debug = require('./lib/debug') var HEADER_MAX_BYTES = 1024 // TODO: Figure out the best max size of the header +var INTERLEAVED_HEADER_BYTES = 4 var CR = 0x0d var NL = 0x0a var NO_BODY_STATUS_CODES = [100, 304] +var INTERLEAVED_SIGN = 0x24 + var Decoder = module.exports = function (opts) { if (!(this instanceof Decoder)) return new Decoder(opts) stream.Writable.call(this, opts) + this._inHead = false this._inBody = false this._headerOffset = 0 this._header = new Buffer(HEADER_MAX_BYTES) this._bodyBytesWritten = 0 + + this._inInterleavedHead = false + this._inInterleavedPacket = false + this._interleavedHeaderOffset = 0 + this._interleavedHeader = new Buffer(INTERLEAVED_HEADER_BYTES) + this._interleavedPacketBytesWritten = 0 } util.inherits(Decoder, stream.Writable) @@ -33,15 +43,95 @@ Decoder.prototype._writeOffset = function (chunk, offset, cb) { if (this._inBody) { offset = this._writeBody(chunk, offset, cb) if (offset === 0) return // chunk not consumed - _writeOffset will be called again when ready - } else { + } else if (this._inHead) { offset = this._writeHead(chunk, offset) + } else if (this._inInterleavedHead) { + offset = this._writeInterleavedHead(chunk, offset) + } else if (this._inInterleavedPacket) { + offset = this._writeInterleavedPacket(chunk, offset,cb) + } else { + if (chunk[offset]===INTERLEAVED_SIGN) + offset = this._writeInterleavedHead(chunk, offset); + else + offset = this._writeHead(chunk, offset) } } cb() } +Decoder.prototype._writeInterleavedHead = function (chunk, offset) { + if (this._interleavedHeaderOffset === 0) { + debug('start of interleaved header'); + this._inInterleavedHead = true; + } + if (chunk.length-offset<4) { + chunk.copy(this._interleavedHeader, this._interleavedHeaderOffset, offset) + this._interleavedHeaderOffset += chunk.length-offset; + return chunk.length; + } + + chunk.copy(this._interleavedHeader,this._interleavedHeaderOffset, offset, offset+INTERLEAVED_HEADER_BYTES); + + + debug('end of interleaved header'); + this._inInterleavedHead = false; + + this._packet = new stream.PassThrough() + this._packet.channel = this._interleavedHeader.readInt8(1); + this._packet.size = this._interleavedHeader.readInt16BE(2); + + this._interleavedHeaderOffset = 0 + this._interleavedHeader = new Buffer(INTERLEAVED_HEADER_BYTES) + + this._inInterleavedPacket = true; + + // _writeInterleavedPacket logic to handle back-pressure + var self = this + this._packet.on('drain', function () { + if (!this._chunk) return + self._writeOffset(this._chunk, this._end, this._cb) + this._chunk = null + this._end = null + this._cb = null + }) + this.emit('packet', this._packet); + + return offset + (INTERLEAVED_HEADER_BYTES - this._interleavedHeaderOffset) +} + +Decoder.prototype._writeInterleavedPacket = function (chunk, offset, cb) { + if (this._interleavedPacketBytesWritten === 0) debug('start of packet') + + var bytesLeft = this._packet.size - this._interleavedPacketBytesWritten + var end = bytesLeft < chunk.length - offset ? offset + bytesLeft : chunk.length + + + var drained = this._packet.write(chunk.slice(offset, end)) + + this._interleavedPacketBytesWritten += end-offset; + + if (this._interleavedPacketBytesWritten >= this._packet.size) { + debug('end of packet') + this._inInterleavedPacket = false + this._interleavedPacketBytesWritten = 0 + this._packet.end() + } + + if (!drained) { + debug('back pressure detected in IncomingPacket') + this._packet._chunk = chunk + this._packet._end = end + this._packet._cb = cb + return 0 // indicate we didn't consume the chunk (yet) + } + + return end +} Decoder.prototype._writeHead = function (chunk, offset) { - if (this._headerOffset === 0) debug('start of header') + if (this._headerOffset === 0) { + debug('start of header'); + this._inHead = true; + } chunk.copy(this._header, this._headerOffset, offset) var origHeaderOffset = this._headerOffset @@ -50,6 +140,7 @@ Decoder.prototype._writeHead = function (chunk, offset) { if (!~bodyStart) return chunk.length // still reading the header debug('end of header') + this._inHead = false; this._msg = new IncomingMessage(this._header) From 3ff06b4cedffc204fce8f88b88631bec32d95551 Mon Sep 17 00:00:00 2001 From: Rafael Sobral Date: Mon, 25 Apr 2016 16:57:49 -0300 Subject: [PATCH 2/2] Meeting Javascript Standard Style --- decoder.js | 43 +++++++++++++++++++++---------------------- 1 file changed, 21 insertions(+), 22 deletions(-) diff --git a/decoder.js b/decoder.js index e53480f..3017908 100644 --- a/decoder.js +++ b/decoder.js @@ -48,41 +48,41 @@ Decoder.prototype._writeOffset = function (chunk, offset, cb) { } else if (this._inInterleavedHead) { offset = this._writeInterleavedHead(chunk, offset) } else if (this._inInterleavedPacket) { - offset = this._writeInterleavedPacket(chunk, offset,cb) + offset = this._writeInterleavedPacket(chunk, offset, cb) } else { - if (chunk[offset]===INTERLEAVED_SIGN) - offset = this._writeInterleavedHead(chunk, offset); - else + if (chunk[offset] === INTERLEAVED_SIGN) { + offset = this._writeInterleavedHead(chunk, offset) + } else { offset = this._writeHead(chunk, offset) + } } } cb() } Decoder.prototype._writeInterleavedHead = function (chunk, offset) { if (this._interleavedHeaderOffset === 0) { - debug('start of interleaved header'); - this._inInterleavedHead = true; + debug('start of interleaved header') + this._inInterleavedHead = true } - if (chunk.length-offset<4) { + if (chunk.length - offset < 4) { chunk.copy(this._interleavedHeader, this._interleavedHeaderOffset, offset) - this._interleavedHeaderOffset += chunk.length-offset; - return chunk.length; + this._interleavedHeaderOffset += chunk.length - offset + return chunk.length } - chunk.copy(this._interleavedHeader,this._interleavedHeaderOffset, offset, offset+INTERLEAVED_HEADER_BYTES); + chunk.copy(this._interleavedHeader, this._interleavedHeaderOffset, offset, offset + INTERLEAVED_HEADER_BYTES) - - debug('end of interleaved header'); - this._inInterleavedHead = false; + debug('end of interleaved header') + this._inInterleavedHead = false this._packet = new stream.PassThrough() - this._packet.channel = this._interleavedHeader.readInt8(1); - this._packet.size = this._interleavedHeader.readInt16BE(2); + this._packet.channel = this._interleavedHeader.readInt8(1) + this._packet.size = this._interleavedHeader.readInt16BE(2) this._interleavedHeaderOffset = 0 this._interleavedHeader = new Buffer(INTERLEAVED_HEADER_BYTES) - this._inInterleavedPacket = true; + this._inInterleavedPacket = true // _writeInterleavedPacket logic to handle back-pressure var self = this @@ -93,7 +93,7 @@ Decoder.prototype._writeInterleavedHead = function (chunk, offset) { this._end = null this._cb = null }) - this.emit('packet', this._packet); + this.emit('packet', this._packet) return offset + (INTERLEAVED_HEADER_BYTES - this._interleavedHeaderOffset) } @@ -104,10 +104,9 @@ Decoder.prototype._writeInterleavedPacket = function (chunk, offset, cb) { var bytesLeft = this._packet.size - this._interleavedPacketBytesWritten var end = bytesLeft < chunk.length - offset ? offset + bytesLeft : chunk.length - var drained = this._packet.write(chunk.slice(offset, end)) - this._interleavedPacketBytesWritten += end-offset; + this._interleavedPacketBytesWritten += end - offset if (this._interleavedPacketBytesWritten >= this._packet.size) { debug('end of packet') @@ -129,8 +128,8 @@ Decoder.prototype._writeInterleavedPacket = function (chunk, offset, cb) { Decoder.prototype._writeHead = function (chunk, offset) { if (this._headerOffset === 0) { - debug('start of header'); - this._inHead = true; + debug('start of header') + this._inHead = true } chunk.copy(this._header, this._headerOffset, offset) @@ -140,7 +139,7 @@ Decoder.prototype._writeHead = function (chunk, offset) { if (!~bodyStart) return chunk.length // still reading the header debug('end of header') - this._inHead = false; + this._inHead = false this._msg = new IncomingMessage(this._header)