rw - npm Package Compare versions

Comparing version 0.0.10 to 0.1.0

index.js

@@ -1,11 +0,13 @@
-exports.read = require("./lib/rw/read");
-exports.readSync = require("./lib/rw/read-sync");
-exports.write = require("./lib/rw/write");
-exports.writeSync = require("./lib/rw/write-sync");
-
+exports.reader = require("./lib/rw/reader");
 exports.writer = require("./lib/rw/writer");
-exports.lineWriter = require("./lib/rw/line-writer");
 
-exports.reader = require("./lib/rw/reader");
-exports.lineReader = require("./lib/rw/line-reader");
-exports.fixedReader = require("./lib/rw/fixed-reader");
+exports.fileReader = require("./lib/rw/file-reader");
+exports.fileWriter = require("./lib/rw/file-writer");
+
+exports.readFile = require("./lib/rw/read-file");
+exports.readFileSync = require("./lib/rw/read-file-sync");
+exports.writeFile = require("./lib/rw/write-file");
+exports.writeFileSync = require("./lib/rw/write-file-sync");
+
+exports.lineParser = require("./lib/rw/line-parser");
+exports.dsvParser = require("./lib/rw/dsv-parser");

lib/rw/reader.js

@@ -1,44 +0,114 @@
-var fs = require("fs");
+module.exports = function(open, read, close) {
+  var reader = {
+        bufferLength: reader_bufferLength,
+        fill: reader_fill,
+        read: reader_read,
+        end: reader_end,
+        ended: false
+      },
+      error = null,
+      fillCallback = null,
+      channel = null,
+      bufferOffset = 0,
+      bufferLength = 1 << 16,
+      bufferAvailable = 0,
+      buffer = new Buffer(bufferLength);
 
-module.exports = function(filePath) {
-  var error = null,
-      eof = false,
-      defers = [],
-      descriptor = null,
-      bufferLength = 1 << 14; // TODO options
+  process.nextTick(open.bind(null, function(newError, newChannel) {
+    error = newError;
+    channel = newChannel;
+    if (fillCallback) {
+      var oldFillCallback = fillCallback;
+      fillCallback = null;
+      reader_fill(oldFillCallback);
+    }
+  }));
 
-  function read(callback) {
-    if (!callback) throw new Error("callback is required");
+  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 there’s an error, return the error.
-    if (error) return void callback(error);
+  function reader_fill(callback) {
+    if (error) return void process.nextTick(callback.bind(reader, 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));
 
-    // If we’ve reached the end of the file, return null.
-    if (eof) return void callback(null, null);
+    // A fill is now in-progress.
+    fillCallback = callback;
 
-    // If there’s an operation already in-progress, wait for that to finish.
-    if (defers) return void defers.push(callback);
+    // If the channel is not yet open, wait for it.
+    if (!channel) return;
 
-    // Otherwise, read some bytes from the file.
-    defers = [];
-    fs.read(descriptor, new Buffer(bufferLength), 0, bufferLength, null, function(error_, bufferLength_, buffer) {
-      error = error_;
-      if (bufferLength_) callback(null, bufferLength_ < bufferLength ? buffer.slice(0, bufferLength_) : buffer);
-      else eof = true, fs.close(descriptor, callback);
-      var defers_ = defers;
-      defers = null;
-      defers_.forEach(read);
+    // Move any unread bytes to the front of the buffer before filling.
+    if (bufferOffset) {
+      if (bufferAvailable) buffer.copy(buffer, 0, bufferOffset, bufferOffset + bufferAvailable);
+      bufferOffset = 0;
+    }
+
+    // Fill the buffer. If no bytes are read, the channel has ended.
+    read(channel, buffer, bufferAvailable, bufferLength - bufferAvailable, function(newError, readLength) {
+      fillCallback = null;
+
+      // 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;
+
+        // Close the channel, ignoring any secondary errors.
+        var oldChannel = channel;
+        channel = null;
+        close(oldChannel, ignore);
+
+        return void callback(error);
+      }
+
+      // Otherwise mark the read bytes as available.
+      bufferAvailable += readLength;
+      callback(null);
     });
   }
 
-  // Open the file for reading.
-  fs.open(filePath, "r", 438 /*=0666*/, function(error_, descriptor_) {
-    error = error_;
-    descriptor = descriptor_;
-    var defers_ = defers;
-    defers = null;
-    defers_.forEach(read);
-  });
+  // Note: the returned data may not be read after the reader starts filling.
+  function reader_read(length) {
+    if (error) throw error;
+    if (fillCallback) throw new Error("cannot read while a fill is in progress");
+    if (length == null) length = bufferAvailable;
+    else if ((length |= 0) <= 0) throw new Error("invalid length: " + length);
 
-  return read;
+    // If all the requested bytes are not available, return null.
+    if (!bufferAvailable || length > bufferAvailable) return null;
+
+    // Otherwise, return the next slice of bytes from the buffer.
+    var oldBufferOffset = bufferOffset;
+    bufferAvailable -= length;
+    bufferOffset += length;
+    return oldBufferOffset || bufferOffset !== bufferLength
+        ? buffer.slice(oldBufferOffset, bufferOffset)
+        : buffer;
+  }
+
+  function reader_end(callback) {
+    if (error) throw error;
+    if (fillCallback) throw new Error("cannot end while a fill is in progress");
+    if (reader.ended) throw new Error("cannot end after already ended");
+    reader.ended = true;
+
+    // Close the channel.
+    var oldChannel = channel;
+    channel = null;
+    close(oldChannel, callback);
+  }
+
+  return reader;
 };
+
+// A no-op callback used to ignore secondary errors on close.
+function ignore(error) {}

lib/rw/writer.js

@@ -1,67 +0,113 @@
-var fs = require("fs");
+module.exports = function(open, write, close) {
+  var writer = {
+        encoding: writer_encoding,
+        bufferLength: writer_bufferLength,
+        drain: writer_drain,
+        write: writer_write,
+        end: writer_end,
+        ended: false
+      },
+      error = null,
+      drainCallback = null,
+      channel = null,
+      encoding = "utf8",
+      bufferUsed = 0,
+      bufferLength = 1 << 16,
+      buffers = [];
 
-module.exports = function(filePath) {
-  var error = null,
-      defers = [],
-      descriptor = null,
-      bufferLength = 1 << 14, // TODO options
-      buffer = new Buffer(bufferLength),
-      bufferIndex = 0;
+  process.nextTick(open.bind(null, function(newError, newChannel) {
+    error = newError;
+    channel = newChannel;
+    if (drainCallback) {
+      var oldDrainCallback = drainCallback;
+      drainCallback = null;
+      writer_drain(oldDrainCallback);
+    }
+  }));
 
-  function write(data, callback) {
-    if (!callback) throw new Error("callback is required");
+  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 there’s an error, return the error.
-    if (error) return void callback(error);
+  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;
+      return writer;
+    }
+    return bufferLength;
+  }
 
-    // If there’s an operation already in-progress, wait for that to finish.
-    if (defers) return void defers.push([data, callback]);
+  function writer_drain(callback) {
+    if (error) return void process.nextTick(callback.bind(writer, 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 (!buffers.length) return void process.nextTick(callback.bind(writer, null));
 
-    // If the file is done, flush the buffer and close the file descriptor.
-    if (data == null) {
-      error = new Error("already closed");
-      return void writeAll(descriptor, buffer, 0, bufferIndex, function(error_) {
-        if (error_) return void callback(error_);
-        fs.close(descriptor, callback);
-      });
-    }
+    // A drain is now in-progress.
+    drainCallback = callback;
 
-    // Always copy the input data into our private buffer,
-    // so that we ignore changes to the input after this function returns.
-    var bufferAdded = Math.min(bufferLength - bufferIndex, data.length);
-    data.copy(buffer, bufferIndex, 0, bufferAdded);
-    bufferIndex += bufferAdded;
+    // If the channel is not yet open, wait for it.
+    if (!channel) return;
 
-    // If the buffer isn’t full, callback immediately.
-    if (bufferIndex < bufferLength) return void callback(null);
+    // Write each buffer, in their original order.
+    buffers.reverse();
+    (function writeNext(newError) {
 
-    // Otherwise, write accumulated bytes to the file.
-    defers = [[data.slice(bufferAdded), callback]];
-    writeAll(descriptor, buffer, 0, bufferIndex, function(error_) {
-      error = error_;
-      bufferIndex = 0;
-      var defers_ = defers;
-      defers = null;
-      defers_.forEach(function(defer) { write(defer[0], defer[1]); });
+      // If an error occurs, stop writing, and shut it all down.
+      if (error = newError) {
+        drainCallback = null;
+
+        // Close the channel, ignoring any secondary errors.
+        var oldChannel = channel;
+        channel = null;
+        close(oldChannel, ignore);
+
+        return void callback(error);
+      }
+
+      // If there’s still data to write, write it. Otherwise callback.
+      var data = buffers.pop();
+      if (!data) return drainCallback = null, bufferUsed = 0, void callback(null);
+      write(channel, data, 0, data.length, writeNext);
+    })(null);
+  }
+
+  // Caution: the specified data must NOT be modified after draining starts.
+  function writer_write(data) {
+    if (error) throw error;
+    if (drainCallback) throw new Error("cannot write while a drain is in progress");
+    if (!(data instanceof Buffer)) data = new Buffer(data + "", encoding);
+    buffers.push(data);
+    bufferUsed += data.length;
+    return bufferUsed < bufferLength;
+  }
+
+  function writer_end(callback) {
+    if (error) throw error;
+    if (drainCallback) throw new Error("cannot end while a drain is in progress");
+    if (writer.ended) throw new Error("cannot end after already ended");
+
+    // Drain any buffered bytes.
+    writer_drain(function(error) {
+      if (error) return void callback(error);
+      writer.ended = true;
+
+      // Close the channel.
+      var oldChannel = channel;
+      channel = null;
+      close(oldChannel, callback);
     });
   }
 
-  // Open the file for writing.
-  fs.open(filePath, "w", 438 /*=0666*/, function(error_, descriptor_) {
-    error = error_;
-    descriptor = descriptor_;
-    var defers_ = defers;
-    defers = null;
-    defers_.forEach(function(defer) { write(defer[0], defer[1]); });
-  });
-
-  return write;
+  return writer;
 };
 
-function writeAll(descriptor, buffer, bufferIndex, bufferLength, callback) {
-  fs.write(descriptor, buffer, bufferIndex, bufferLength, null, function(error_, bufferLength_) {
-    if (error_) return void callback(error_.code === "EPIPE" ? null : error_); // ignore broken pipe, e.g., | head
-    if (bufferLength_ < bufferLength) return void writeAll(descriptor, buffer, bufferIndex + bufferLength_, bufferLength - bufferLength_, callback);
-    callback(null);
-  });
-}
+// A no-op callback used to ignore secondary errors on close.
+function ignore(error) {}

package.json

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

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Major refactor

Supply chain risk

Package has recently undergone a major refactor. It may be unstable or indicate significant internal changes. Use caution when updating to versions that include significant changes.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

30693

increased by41.49%

Number of package files

29

increased by11.54%

Lines of code

559

increased by64.9%

Worsened metrics

Number of medium supply chain risk alerts

1

increased byInfinity%

No dependency changes