complete change in approach

  - necessitated a rewrite, but many bugs with small binary streams fixed
  - added tests

Author: seangarner

CHANGELOG.md

@@ -1,5 +1,11 @@
 # buffer-peek-stream changelog
 
+## 0.2.0 (2014/11/12)
+
+  - rewrite using a completely different approach because 0.1.x approach was too fragile and often
+    broke with edge cases
+  - api backwards incompatible with 0.1 - will bump to 1.0.0 only once api is stable
+
 ## 0.1.2 (2014/11/07)
 
   - fix data being emitted out of order on a large peeked stream

buffer-peek-stream.js

@@ -0,0 +1,79 @@
+var stream = require('stream');
+var util = require('util');
+
+function peek(source, bytes, callback) {
+  if (!callback) return peek(source, undefined, bytes);
+
+  var dest = new BufferPeekStream({peekBytes: bytes});
+
+  dest.once('peek', function (buffer) {
+    callback(null, buffer, dest);
+  });
+
+  return source.pipe(dest);
+}
+peek.BufferPeekStream = BufferPeekStream;
+
+module.exports = peek;
+
+
+function BufferPeekStream(opts) {
+  if (!opts) opts = {};
+
+  opts.highWaterMark = opts.peekBytes || 65536;
+
+  stream.Transform.call(this, opts);
+
+  this._peekState = {
+    buffer: [],
+    bytes: 0,
+    maxBytes: opts.peekBytes || 65536,
+    peeked: false
+  };
+}
+
+util.inherits(BufferPeekStream, stream.Transform);
+
+
+BufferPeekStream.prototype._transform = function _transform(chunk, enc, callback) {
+  var state = this._peekState;
+
+  // buffer incoming chunks until we have enough for our peek
+  state.buffer.push(chunk);
+  state.bytes += chunk.length;
+
+  // get more?
+  if (state.bytes >= state.maxBytes) _peek(this, callback);
+    else callback();
+};
+
+
+BufferPeekStream.prototype._flush = function _flush(callback) {
+  if (this._peekState.peeked) callback();
+    else _peek(this, callback);
+};
+
+
+function _peek(stream, callback) {
+  var state = stream._peekState;
+
+  var buffer = Buffer.concat(state.buffer);
+
+  // emit exactly the number of bytes we wanted to peek
+  stream.emit('peek', buffer.slice(0, state.maxBytes));
+
+  stream.push(buffer);
+
+  state.buffer = null;
+  state.bytes = null;
+  state.peeked = true;
+
+  stream._transform = passthrough;
+
+  callback();
+}
+
+function passthrough(chunk, enc, callback) {
+  this.push(chunk);
+  callback();
+}

lib/buffer-peek-stream.js

@@ -1,17 +1,18 @@
 {
   "name": "buffer-peek-stream",
-  "version": "0.1.3",
+  "version": "0.2.0",
   "description": "Transform stream that lets you inspect the start of a ReadStream before deciding what to do with it",
-  "main": "lib/index.js",
+  "main": "buffer-peek-stream.js",
   "scripts": {
-    "test": "mocha test.js"
+    "test": "node_modules/mocha/bin/mocha test.js"
   },
   "repository": {
     "type": "git",
     "url": "https://github.com/seangarner/node-buffer-peek-stream.git"
   },
   "keywords": [
     "stream",
+    "streams",
     "peek",
     "parse",
     "buffer",
@@ -24,8 +25,11 @@
   },
   "homepage": "https://github.com/seangarner/node-buffer-peek-stream",
   "devDependencies": {
-    "chai": "^1.9.2",
+    "chai": "^1.10.0",
+    "concat-stream": "^1.4.6",
+    "dev-null-stream": "0.0.1",
     "mocha": "^2.0.1",
-    "randomstring": "^1.0.3"
+    "randstream": "^0.3.2",
+    "truncate-stream": "^1.0.1"
   }
 }

lib/index.js

@@ -0,0 +1,98 @@
+var fs = require('fs');
+var expect = require('chai').expect;
+var RandStream = require('randstream');
+var TruncateStream = require('truncate-stream');
+var DevNullStream = require('dev-null-stream');
+var concat = require('concat-stream');
+
+var peek = require('./buffer-peek-stream');
+
+
+function make(size, mode) {
+  return (new RandStream({mode: mode || 'pseudo'})).pipe(new TruncateStream({maxBytes: size}));
+}
+
+describe('peek', function() {
+  this.timeout(250);
+
+  it('should callback with a buffer', function (done) {
+    var source = make(50000);
+    peek(source, 50000, function (err, buffer) {
+      if (err) return done(err);
+      expect(buffer).to.be.an.instanceof(Buffer);
+      done();
+    }).pipe(new DevNullStream());
+  });
+
+  it('should callback with exactly the number of bytes requested', function (done) {
+    peek(make(50000), 1000, function (err, buffer) {
+      if (err) return done(err);
+      expect(buffer).to.have.lengthOf(1000);
+      done();
+    });
+  });
+
+  it('should callback with all bytes when peeking more than is available', function (done) {
+    peek(make(1000), 5000, function (err, buffer) {
+      if (err) return done(err);
+      expect(buffer).to.have.lengthOf(1000);
+      done();
+    });
+  });
+
+  it('should callback with a stream which receives all bytes', function (done) {
+    var source = make(50000);
+    peek(source, 1000, function (err, buffer, stream) {
+      if (err) return done(err);
+      stream.pipe(concat(function (data) {
+        expect(data).to.have.lengthOf(50000);
+        done();
+      }));
+    });
+  });
+
+  it('should return a stream which receives all bytes', function (done) {
+    peek(make(5000), 1000, function () {}).pipe(concat(function (data) {
+      expect(data).to.have.lengthOf(5000);
+      done();
+    }));
+  });
+
+  it('should return the same stream as it calls back', function (done) {
+    var res = peek(make(5000), 1000, function (err, buffer, stream) {
+      if (err) return done(err);
+      expect(stream).to.equal(res);
+      done();
+    });
+  });
+
+  it('should peek 65536 bytes by default', function (done) {
+    peek(make(100000), function (err, buffer) {
+      if (err) return done(err);
+      expect(buffer).to.have.lengthOf(65536);
+      done();
+    });
+  });
+
+  it('should work when peeked more once in a pipeline', function (done) {
+    peek(make(100000), 50000, function (err, first, stream) {
+      if (err) return done(err);
+      expect(first).to.have.lengthOf(50000);
+      peek(stream, 40000, function (err, second, stream) {
+        if (err) return done(err);
+        expect(second).to.have.lengthOf(40000);
+        expect(second).to.eql(first.slice(0, 40000));
+        stream.pipe(concat(function (data) {
+          expect(data).to.have.lengthOf(100000);
+          expect(first).to.eql(data.slice(0, 50000));
+          expect(second).to.eql(data.slice(0, 40000));
+          done();
+        }));
+      });
+    });
+  });
+
+  //TODO: peeking inside gzip data (transform)
+
+  //TODO: peeking a http response
+});