index.js 1.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374
  1. var Transform = require('stream').Transform;
  2. var $u = require('util');
  3. $u.inherits(StreamSlicer, Transform);
  4. function StreamSlicer(options) {
  5. if (!(this instanceof StreamSlicer))
  6. return new StreamSlicer(options);
  7. Transform.call(this, options);
  8. this._buffer = [];
  9. this._currentLength = 0;
  10. if (options && options.sliceBy)
  11. this._sliceBy = options.sliceBy;
  12. else
  13. this._sliceBy = '\n';
  14. if (options && options.replaceWith)
  15. this.replaceWith = new Buffer(options.replaceWith);
  16. }
  17. StreamSlicer.prototype._transform = function(chunk, encoding, callback) {
  18. chunk = String(chunk);
  19. var start = 0;
  20. var index = -1;
  21. while ((index = chunk.indexOf(this._sliceBy, start)) > -1 ) {
  22. var miniChunk = chunk.substring(start, index);
  23. this._append( miniChunk );
  24. this._separatorFlush();
  25. start = index + this._sliceBy.length;
  26. }
  27. var trailing = chunk.substring(start);
  28. if (trailing.length > 0)
  29. this._append( trailing );
  30. callback();
  31. };
  32. StreamSlicer.prototype._append = function ( str ) {
  33. var chunk = new Buffer(str);
  34. this._buffer.push(chunk);
  35. this._currentLength += chunk.length;
  36. };
  37. StreamSlicer.prototype._separatorFlush = function (transformFlush) {
  38. if (this.replaceWith && !transformFlush) {
  39. this._buffer.push(this.replaceWith);
  40. this._currentLength += this.replaceWith.length;
  41. }
  42. var data = Buffer.concat(this._buffer, this._currentLength);
  43. this._buffer = [];
  44. this._currentLength = 0;
  45. this.push(data);
  46. this.emit('slice', data);
  47. };
  48. StreamSlicer.prototype._flush = function (callback) {
  49. this._separatorFlush(true);
  50. if (callback)
  51. callback();
  52. };
  53. module.exports = StreamSlicer;