rw - npm Package Compare versions

Comparing version 0.1.1 to 0.1.2

index.js

@@ -14,1 +14,3 @@ exports.reader = require("./lib/rw/reader");
 exports.dsvParser = require("./lib/rw/dsv-parser");
+
+exports.pipe = require("./lib/rw/pipe");

lib/rw/dsv-parser.js

@@ -5,39 +5,46 @@ module.exports = function() {
         encoding: parser_encoding,
+        transform: parser_transform,
         push: parser_push,
         pop: parser_pop
       },
-      buffer,
-      bufferOffset = 0,
-      bufferLength = 0,
-      encoding = "utf8",
-      delimiterCode = ",".charCodeAt(0),
-      row = [],
-      state = STATE_FIELD,
-      lineEmpty = true,
-      tokenOffset = 0,
-      tokenDefaultLength = 1 << 7,
-      tokenLength = tokenDefaultLength,
-      token = new Buffer(tokenLength);
+      delimiterCode = CODE_COMMA, // character code of delimiter
+      encoding = "utf8", // encoding for converting tokens to strings
+      transform = TRANSFORM_AUTO, // transform function
+      bufferOffset = 0, // number of bytes we’ve read from buffer
+      bufferLength = 0, // number of bytes available in buffer
+      buffer = new Buffer(0), // current buffer (not a copy)
+      state = STATE_FIELD, // parser finite state machine
+      row = [], // current row of fields
+      rowIndex = -1, // count of rows that have been parsed
+      lineEmpty = true, // true if the line was completely empty (to skip)
+      tokenOffset = 0, // number of bytes we’ve written for the current token
+      tokenDefaultLength = 1 << 7, // default token length to allocate
+      tokenLength = tokenDefaultLength, // current token length (doubles)
+      token = new Buffer(tokenLength); // current token accumulation buffer
 
   function parser_delimiter(newDelimiter) {
-    if (arguments.length) {
-      if (buffer != null) throw new Error("cannot change delimiter after pushing data");
-      if ((newDelimiter += "").length !== 1) throw new Error("invalid delimiter: " + newDelimiter);
-      delimiterCode = newDelimiter.charCodeAt(0);
-      return parser;
-    }
-    return String.fromCharCode(delimiterCode);
+    if (!arguments.length) return String.fromCharCode(delimiterCode);
+    if (buffer != null) throw new Error("cannot change delimiter after pushing data");
+    if ((newDelimiter += "").length !== 1) throw new Error("invalid delimiter: " + newDelimiter);
+    delimiterCode = newDelimiter.charCodeAt(0);
+    return parser;
   }
 
   function parser_encoding(newEncoding) {
-    if (arguments.length) {
-      if (buffer != null) throw new Error("cannot change encoding after pushing data");
-      if (newEncoding == null) newEncoding = null;
-      else if (!Buffer.isEncoding(newEncoding = newEncoding + "")) throw new Error("unknown encoding: " + newEncoding);
-      encoding = newEncoding;
-      return parser;
-    }
-    return encoding;
+    if (!arguments.length) return encoding;
+    if (buffer != null) throw new Error("cannot change encoding after pushing data");
+    if (!Buffer.isEncoding(newEncoding = newEncoding + "")) throw new Error("unknown encoding: " + newEncoding);
+    encoding = newEncoding;
+    return parser;
   }
 
+  function parser_transform(newTransform) {
+    if (!arguments.length) return transform === TRANSFORM_AUTO ? "auto" : transform === TRANSFORM_IDENTITY ? null : transform;
+    if (newTransform === null) newTransform = TRANSFORM_IDENTITY;
+    else if (newTransform === "auto") newTransform = TRANSFORM_AUTO;
+    else if (typeof newTransform != "function") throw new Error("invalid transform: " + newTransform);
+    transform = newTransform;
+    return parser;
+  }
+
   function parser_push(data) {
@@ -135,2 +142,10 @@ if (bufferOffset < bufferLength) throw new Error("cannot push before all lines are popped");
         if (oldLineEmpty) continue; // skip empty lines
+        ++rowIndex;
+        if (transform === TRANSFORM_AUTO) {
+          transform = autoTransform(oldRow);
+          --rowIndex;
+          continue;
+        }
+        oldRow = transform(oldRow, rowIndex);
+        if (oldRow == null) continue; // skip filtered rows
         return oldRow;
@@ -146,10 +161,22 @@ }
 
-var CODE_QUOTE = 34,
-    CODE_LINE_FEED = 10,
-    CODE_CARRIAGE_RETURN = 13;
+var CODE_QUOTE = '"'.charCodeAt(0),
+    CODE_COMMA = ",".charCodeAt(0),
+    CODE_LINE_FEED = "\n".charCodeAt(0),
+    CODE_CARRIAGE_RETURN = "\r".charCodeAt(0);
 
-var STATE_FIELD = 1, // inside an unquoted value
-    STATE_QUOTE = 2, // inside a quoted value
-    STATE_AFTER_QUOTE_IN_QUOTE = 3, // after a quote (") in a quoted value
-    STATE_AFTER_CARRIAGE_RETURN = 4, // after a carriage return in an unquoted value
-    STATE_END_OF_LINE = 5; // after the last field in a line
+var STATES = 0,
+    STATE_FIELD = ++STATES, // inside an unquoted value
+    STATE_QUOTE = ++STATES, // inside a quoted value
+    STATE_AFTER_QUOTE_IN_QUOTE = ++STATES, // after a quote (") in a quoted value
+    STATE_AFTER_CARRIAGE_RETURN = ++STATES, // after a carriage return in an unquoted value
+    STATE_END_OF_LINE = ++STATES; // after the last field in a line
+
+var TRANSFORM_AUTO = {},
+    TRANSFORM_IDENTITY = function(d) { return d; };
+
+function autoTransform(row) {
+  return new Function("d", "return {"
+      + row.map(function(name, i) {
+            return JSON.stringify(name) + ": d[" + i + "]";
+          }).join(",")+ "}");
+}

lib/rw/file-reader.js

@@ -15,4 +15,4 @@ var fs = require("fs");
 
-function read(descriptor, buffer, bufferOffset, bufferLength, callback) {
-  fs.read(descriptor, buffer, bufferOffset, bufferLength, null, callback);
+function read(descriptor, buffer, offset, length, callback) {
+  fs.read(descriptor, buffer, offset, length, null, callback);
 }
@@ -19,0 +19,0 @@

lib/rw/file-writer.js

@@ -16,6 +16,6 @@ var fs = require("fs");
 // if (error_ && error_.code === "EPIPE") error_ = null; // TODO ignore broken pipe and ignore subsequent writes
-function write(channel, buffer, bufferOffset, bufferLength, callback) {
-  fs.write(channel, buffer, bufferOffset, bufferLength, null, function(error, writeLength) {
+function write(channel, buffer, offset, length, callback) {
+  fs.write(channel, buffer, offset, length, null, function(error, writeLength) {
     if (error) return void callback(error);
-    if (writeLength < bufferLength) return void write(channel, buffer, bufferOffset + writeLength, bufferLength - writeLength, callback);
+    if (writeLength < length) return void write(channel, buffer, offset + writeLength, length - writeLength, callback);
     callback(null);
@@ -22,0 +22,0 @@ });

lib/rw/line-parser.js

@@ -7,17 +7,15 @@ module.exports = function() {
       },
-      buffer,
+      buffer = new Buffer(0),
       bufferOffset = 0,
       bufferLength = 0,
       encoding = "utf8",
-      fragments = null;
+      fragment = null,
+      state = STATE_DEFAULT;
 
   function parser_encoding(newEncoding) {
-    if (arguments.length) {
-      if (buffer != null) throw new Error("cannot change encoding after pushing data");
-      if (newEncoding == null) newEncoding = null;
-      else if (!Buffer.isEncoding(newEncoding = newEncoding + "")) throw new Error("unknown encoding: " + newEncoding);
-      encoding = newEncoding;
-      return parser;
-    }
-    return encoding;
+    if (!arguments.length) return encoding;
+    if (buffer != null) throw new Error("cannot change encoding after pushing data");
+    if (!Buffer.isEncoding(newEncoding = newEncoding + "")) throw new Error("unknown encoding: " + newEncoding);
+    encoding = newEncoding;
+    return parser;
   }
@@ -33,4 +31,3 @@
   function parser_pop(allowPartial) {
-    var oldBufferOffset = bufferOffset,
-        terminatorLength = 0;
+    var oldBufferOffset = bufferOffset;
 
@@ -40,37 +37,55 @@ // Find the next line terminator.
       var code = buffer[bufferOffset++];
+      if (state === STATE_MAYBE_CARRIAGE_RETURN_LINE_FEED) {
+        if (code === CODE_LINE_FEED) {
+          state = STATE_AFTER_CARRIAGE_RETURN_LINE_FEED;
+        } else {
+          state = STATE_AFTER_LINE_FEED; // treat bare \r as \n
+          --bufferOffset;
+        }
+        break;
+      }
       if (code === CODE_LINE_FEED) {
-        ++terminatorLength;
+        state = STATE_AFTER_LINE_FEED;
         break;
       }
       if (code === CODE_CARRIAGE_RETURN) {
-        ++terminatorLength;
-        if (buffer[bufferOffset] === CODE_LINE_FEED) ++bufferOffset, ++terminatorLength;
-        break;
+        state = STATE_MAYBE_CARRIAGE_RETURN_LINE_FEED;
+        continue;
       }
     }
 
-    // Slice the buffer up to the line terminator.
-    var fragment = buffer.slice(oldBufferOffset, bufferOffset - terminatorLength);
+    // Slice out the new data.
+    var newFragment = buffer.slice(oldBufferOffset, bufferOffset);
 
-    // If we read to the end of a line, return it!
-    if (terminatorLength || allowPartial) {
+    // Combine it with the old data, if any.
+    if (fragment != null) {
+      var oldFragment = newFragment;
+      newFragment = new Buffer(fragment.length + oldFragment.length);
+      fragment.copy(newFragment);
+      oldFragment.copy(newFragment, fragment.length);
+      fragment = null;
+    }
 
-      // Combine this with previously-read line fragments, if any.
-      if (fragments) {
-        fragments.push(fragment);
-        fragment = Buffer.concat(fragments);
-        fragments = null;
-      }
+    // Slice off the \r or \n.
+    if (state === STATE_AFTER_LINE_FEED) {
+      state = STATE_DEFAULT;
+      return newFragment.slice(0, -1);
+    }
 
-      // If this is the last line in a file not terminated with a new line,
-      // then return the line even though there’s no trailing terminator.
-      return !allowPartial || fragment.length ? fragment.toString(encoding) : null;
+    // Slice off the \r\n.
+    if (state === STATE_AFTER_CARRIAGE_RETURN_LINE_FEED) {
+      state = STATE_DEFAULT;
+      return newFragment.slice(0, -2);
     }
 
+    // Return the whole thing, if a trailing partial line is wanted.
+    if (allowPartial) {
+      state = STATE_DEFAULT;
+      return newFragment.length ? newFragment : null;
+    }
+
     // Otherwise, we’ve read part of a line. Copy the fragment so that the
     // source buffer can be modified without changing the fragment contents.
-    var fragmentCopy = new Buffer(fragment.length);
-    fragment.copy(fragmentCopy);
-    if (fragments) fragments.push(fragmentCopy);
-    else fragments = [fragmentCopy];
+    fragment = new Buffer(newFragment.length);
+    newFragment.copy(fragment);
     return null;
@@ -84,1 +99,6 @@ }
     CODE_CARRIAGE_RETURN = 13;
+
+var STATE_DEFAULT = 1,
+    STATE_AFTER_LINE_FEED = 2,
+    STATE_AFTER_CARRIAGE_RETURN_LINE_FEED = 3,
+    STATE_MAYBE_CARRIAGE_RETURN_LINE_FEED = 4;

lib/rw/reader.js

@@ -11,3 +11,3 @@ module.exports = function(open, read, close) {
       fillCallback = null,
-      channel = null,
+      channel = CHANNEL_OPEN,
       bufferOffset = 0,
@@ -18,34 +18,35 @@ bufferLength = 1 << 16,
 
-  process.nextTick(open.bind(null, function(newError, newChannel) {
-    error = newError;
-    channel = newChannel;
-    if (fillCallback) {
-      var oldFillCallback = fillCallback;
-      fillCallback = null;
-      reader_fill(oldFillCallback);
-    }
-  }));
-
   function reader_bufferLength(newBufferLength) {
-    if (arguments.length) {
-      if (bufferAvailable) throw new Error("cannot change buffer length while the buffer is not empty");
-      if (fillCallback) throw new Error("cannot change buffer length while the buffer is filling");
-      if ((newBufferLength |= 0) <= 0) throw new Error("invalid length: " + newBufferLength);
-      buffer = new Buffer(bufferLength = newBufferLength);
-      return reader;
-    }
-    return bufferLength;
+    if (!arguments.length) return bufferLength;
+    if (bufferAvailable) throw new Error("cannot change buffer length while the buffer is not empty");
+    if (fillCallback) throw new Error("cannot change buffer length while the buffer is filling");
+    if ((newBufferLength |= 0) <= 0) throw new Error("length <= 0: " + newBufferLength);
+    buffer = new Buffer(bufferLength = newBufferLength);
+    return reader;
   }
 
+  function reader_open() {
+    open(function(newError, newChannel) {
+      error = newError;
+      channel = newChannel;
+      fill(fillCallback);
+    });
+  }
+
   function reader_fill(callback) {
-    if (error) return void process.nextTick(callback.bind(reader, error));
+    if (error) return void process.nextTick(callback.bind(null, error));
     if (fillCallback) throw new Error("cannot fill while a fill is already in progress");
     if (reader.ended) throw new Error("cannot fill after ended");
-    if (bufferAvailable >= bufferLength) return void process.nextTick(callback.bind(reader, null));
+    fill(callback);
+  }
 
+  function fill(callback) {
+    if (!callback) throw new Error("missing callback");
+    if (channel === CHANNEL_OPEN) return fillCallback = callback, void reader_open();
+
     // A fill is now in-progress.
     fillCallback = callback;
 
-    // If the channel is not yet open, wait for it.
-    if (!channel) return;
+    // If there’s no space to read more, emulate reading zero bytes.
+    if (bufferAvailable >= bufferLength) return void process.nextTick(afterFill.bind(null, null, 0));
 
@@ -59,22 +60,18 @@ // Move any unread bytes to the front of the buffer before filling.
     // Fill the buffer. If no bytes are read, the channel has ended.
-    read(channel, buffer, bufferAvailable, bufferLength - bufferAvailable, function(newError, readLength) {
-      fillCallback = null;
+    read(channel, buffer, bufferAvailable, bufferLength - bufferAvailable, afterFill);
+  }
 
-      // If an error occurs, stop reading, and shut it all down.
-      // Or if no more bytes were available, then we’ve reached the end.
-      if ((error = newError) || !readLength) {
-        reader.ended = true;
+  function afterFill(newError, readLength) {
+    var oldFillCallback = fillCallback;
+    fillCallback = null;
 
-        // Close the channel, ignoring any secondary errors.
-        var oldChannel = channel;
-        channel = null;
-        close(oldChannel, ignore);
+    // If an error occurs, close, ignoring any secondary errors.
+    if (error = newError) return void close(channel, oldFillCallback.bind(null, error));
 
-        return void callback(error);
-      }
+    // If no more bytes were available, then we’ve reached the end, so close.
+    if (!readLength) return reader.ended = true, void close(channel, oldFillCallback);
 
-      // Otherwise mark the read bytes as available.
-      bufferAvailable += readLength;
-      callback(null);
-    });
+    // Otherwise mark the read bytes as available.
+    bufferAvailable += readLength;
+    oldFillCallback(null);
   }
@@ -87,3 +84,4 @@
     if (length == null) length = bufferAvailable;
-    else if ((length |= 0) <= 0) throw new Error("invalid length: " + length);
+    else if ((length |= 0) < 0) throw new Error("length < 0: " + length);
+    else if (length > bufferLength) throw new Error("length > bufferLength: " + length);
 
@@ -102,2 +100,3 @@ // If all the requested bytes are not available, return null.
 
+  // Close the channel.
   function reader_end(callback) {
@@ -107,8 +106,5 @@ if (error) throw error;
     if (reader.ended) throw new Error("cannot end after already ended");
+    if (!callback) callback = rethrow;
     reader.ended = true;
-
-    // Close the channel.
-    var oldChannel = channel;
-    channel = null;
-    close(oldChannel, callback);
+    close(channel, callback);
   }
@@ -119,3 +115,6 @@
 
-// A no-op callback used to ignore secondary errors on close.
-function ignore(error) {}
+function rethrow(error) {
+  if (error) throw error;
+}
+
+var CHANNEL_OPEN = {};

lib/rw/writer.js

@@ -12,3 +12,3 @@ module.exports = function(open, write, close) {
       drainCallback = null,
-      channel = null,
+      channel = CHANNEL_OPEN,
       encoding = "utf8",
@@ -20,63 +20,58 @@ bufferUsed = 0,
 
-  process.nextTick(open.bind(null, function(newError, newChannel) {
-    error = newError;
-    channel = newChannel;
-    if (drainCallback) {
-      var oldDrainCallback = drainCallback;
-      drainCallback = null;
-      writer_drain(oldDrainCallback);
-    }
-  }));
-
   function writer_encoding(newEncoding) {
-    if (arguments.length) {
-      if (!Buffer.isEncoding(newEncoding = newEncoding + "")) throw new Error("unknown encoding: " + newEncoding);
-      encoding = newEncoding;
-      return writer;
-    }
-    return encoding;
+    if (!arguments.length) return encoding;
+    if (!Buffer.isEncoding(newEncoding = newEncoding + "")) throw new Error("unknown encoding: " + newEncoding);
+    encoding = newEncoding;
+    return writer;
   }
 
   function writer_bufferLength(newBufferLength) {
-    if (arguments.length) {
-      if (bufferUsed) throw new Error("cannot change buffer length while the buffer is not empty");
-      if ((newBufferLength |= 0) <= 0) throw new Error("invalid length: " + newBufferLength);
-      bufferLength = newBufferLength;
-      bufferOverflowLength = bufferLength << 1;
-      buffer = new Buffer(bufferOverflowLength);
-      return writer;
-    }
-    return bufferLength;
+    if (!arguments.length) return bufferLength;
+    if (bufferUsed) throw new Error("cannot change buffer length while the buffer is not empty");
+    if ((newBufferLength |= 0) <= 0) throw new Error("length <= 0: " + newBufferLength);
+    bufferLength = newBufferLength;
+    bufferOverflowLength = bufferLength << 1;
+    buffer = new Buffer(bufferOverflowLength);
+    return writer;
   }
 
+  function writer_open() {
+    open(function(newError, newChannel) {
+      error = newError;
+      channel = newChannel;
+      drain(drainCallback);
+    });
+  }
+
   function writer_drain(callback) {
-    if (error) return void process.nextTick(callback.bind(writer, error));
+    if (error) return void process.nextTick(callback.bind(null, error));
     if (drainCallback) throw new Error("cannot drain while a drain is already in progress");
     if (writer.ended) throw new Error("cannot drain after ended");
-    if (!bufferUsed) return void process.nextTick(callback.bind(writer, null));
+    drain(callback);
+  }
 
+  function drain(callback) {
+    if (!callback) throw new Error("missing callback");
+    if (channel === CHANNEL_OPEN) return drainCallback = callback, void writer_open();
+
     // A drain is now in-progress.
     drainCallback = callback;
 
-    // If the channel is not yet open, wait for it.
-    if (!channel) return;
+    // If there’s nothing to drain, emulate draining zero bytes.
+    if (!bufferUsed) return void process.nextTick(afterDrain.bind(null, null));
 
-    // Write out the buffer.
-    write(channel, buffer, 0, bufferUsed, function(newError) {
-      drainCallback = null;
+    // Drain the buffer.
+    write(channel, buffer, 0, bufferUsed, afterDrain);
+  }
 
-      // If an error occurs, stop writing, and shut it all down.
-      if (error = newError) {
+  function afterDrain(newError) {
+    var oldDrainCallback = drainCallback;
+    drainCallback = null;
 
-        // Close the channel, ignoring any secondary errors.
-        var oldChannel = channel;
-        channel = null;
-        close(oldChannel, ignore);
+    // If an error occurs, close, ignoring any secondary errors.
+    if (error = newError) return void close(channel, oldDrainCallback.bind(null, error));
 
-        return void callback(error);
-      }
-
-      bufferUsed = 0;
-      callback(null);
-    });
+    // Otherwise mark the written bytes as drained.
+    bufferUsed = 0;
+    oldDrainCallback(null);
   }
@@ -87,2 +82,3 @@
     if (drainCallback) throw new Error("cannot write while a drain is in progress");
+    if (writer.ended) throw new Error("cannot write after ended");
     if (!(data instanceof Buffer)) data = new Buffer(data + "", encoding);
@@ -94,4 +90,5 @@ var oldBufferUsed = bufferUsed;
     if (bufferUsed > bufferOverflowLength) {
+      do bufferOverflowLength <<= 1; while (bufferUsed > bufferOverflowLength);
       var oldBuffer = buffer;
-      buffer = new Buffer(bufferOverflowLength <<= 1);
+      buffer = new Buffer(bufferOverflowLength);
       oldBuffer.copy(buffer, 0, 0, oldBufferUsed);
@@ -112,12 +109,9 @@ }
     if (writer.ended) throw new Error("cannot end after already ended");
+    if (!callback) callback = rethrow;
+    writer.ended = true;
 
-    // Drain any buffered bytes.
-    writer_drain(function(error) {
+    // Drain any buffered bytes, then close the channel.
+    drain(function(error) {
       if (error) return void callback(error);
-      writer.ended = true;
-
-      // Close the channel.
-      var oldChannel = channel;
-      channel = null;
-      close(oldChannel, callback);
+      close(channel, callback);
     });
@@ -129,3 +123,6 @@ }
 
-// A no-op callback used to ignore secondary errors on close.
-function ignore(error) {}
+function rethrow(error) {
+  if (error) throw error;
+}
+
+var CHANNEL_OPEN = {};

package.json

 {
   "name": "rw",
-  "version": "0.1.1",
+  "version": "0.1.2",
   "description": "Wrappers of fs.{read,write}File that work for /dev/std{in,out}.",
@@ -5,0 +5,0 @@ "keywords": [

README.md

@@ -38,3 +38,3 @@ # stdin & stdout, the right way
 
-But that’s a pain, since now your code has two different code paths for reading inputs depending on whether you’re reading a real file or stdin. And the code gets even more complex if you want to [read that file synchronously](https://github.com/mbostock/rw/blob/master/lib/rw/read-sync.js).
+But that’s a pain, since now your code has two different code paths for reading inputs depending on whether you’re reading a real file or stdin. And the code gets even more complex if you want to [read that file synchronously](https://github.com/mbostock/rw/blob/master/lib/rw/read-file-sync.js).
 
@@ -70,3 +70,3 @@ You could also try a different pattern for writing to stdout:
 ```js
-var contents = rw.readSync("/dev/stdin", "utf8");
+var contents = rw.readFileSync("/dev/stdin", "utf8");
 ```
@@ -77,3 +77,3 @@
 ```js
-rw.writeSync("/dev/stdout", contents, "utf8");
+rw.writeFileSync("/dev/stdout", contents, "utf8");
 ```
@@ -80,0 +80,0 @@

No alert changes

Improved metrics

Total package byte prevSize

43635

40.29%

Number of package files

40

37.93%

Lines of code

721

26.94%

No dependency changes