chore(createWriteStream): rewrite in ES5 (#26)

lib/api/createWriteStream.js

+var Writable = require('stream').Writable;
+
+var BigInt = require('../tools/bigint');
+var request = require('../tools/smb2-forge').request;
+var constants = require('../structures/constants');
+
+// Where does it come from?!
+var maxPacketSize = 0x00010000 - 0x71;
+
+module.exports = function createWriteStream(path, options, cb) {
+  if (typeof options === 'function') {
+    cb = options;
+    options = {};
+  }
+
+  var createDisposition;
+  var flags = options != null && options.flags;
+  if (flags == null) {
+  } else if (flags === 'r+') {
+    createDisposition = constants.FILE_OPEN;
+  } else if (flags === 'w' || flags === 'w+') {
+    createDisposition = constants.FILE_OVERWRITE_IF;
+  } else if (flags === 'wx' || flags === 'wx+') {
+    createDisposition = constants.FILE_CREATE;
+  }
+
+  var connection = this;
+  request(
+    'create',
+    { createDisposition: createDisposition, path: path },
+    connection,
+    function(err, file) {
+      if (err != null) {
+        return cb(err);
+      }
+
+      var offset = new BigInt(8, (options != null && options.start) || 0);
+
+      var close = request.bind(undefined, 'close', file, connection);
+
+      function write(buffer, i, cb) {
+        var j = i + maxPacketSize;
+        var chunk = buffer.slice(i, j)
+        request(
+          'write',
+          {
+            Buffer: chunk,
+            FileId: file.FileId,
+            Offset: offset.toBuffer(),
+          },
+          connection,
+          function(err, result) {
+            if (err != null) {
+              return cb(err);
+            }
+            offset = offset.add(chunk.length)
+            if (j < buffer.length) {
+              return write(buffer, j, cb);
+            }
+            cb();
+          }
+        );
+      }
+
+      var stream = new Writable();
+      stream._destroy = function(err, cb) {
+        close(function(err2) {
+          if (err != null) {
+            return cb(err2);
+          }
+          cb(err);
+        });
+      };
+      stream._final = close
+      stream._write = function(chunk, encoding, next) {
+        write(
+          Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk, encoding),
+          0,
+          next
+        );
+      };
+      cb(null, stream);
+    }
+  );
+};

lib/api/src/createWriteStream.js

-import Bigint from '../tools/bigint';
-import Bluebird from 'bluebird';
-import { request } from '../tools/smb2-forge';
-import { Writable } from 'stream';
-
-import {
-  FILE_OPEN,
-  FILE_OPEN_IF,
-  FILE_OVERWRITE_IF,
-  FILE_CREATE,
-} from '../structures/constants';
-
-const requestAsync = Bluebird.promisify(request);
-
-const maxPacketSize = new Bigint(8, 0x00010000 - 0x71);
-
-function* fibonacci() {
-  let a = 1;
-  let b = 2;
-
-  for (;;) {
-    const c = a;
-    a = b;
-    b = c + a;
-    yield c;
-  }
-}
-
-class SmbWritableStream extends Writable {
-  constructor(connection, file, options = {}) {
-    super(options);
-
-    const { encoding = 'utf8', start = 0 } = options;
-
-    this.connection = connection;
-    this.encoding = encoding;
-    this.file = file;
-    this.offset = new Bigint(8, start);
-  }
-
-  async _write(chunk, encoding, next) {
-    encoding = this.encoding || encoding;
-    chunk = Buffer.isBuffer(chunk) ? chunk : new Buffer(chunk, encoding);
-
-    while (chunk.length > 0) {
-      const packetSize = Math.min(maxPacketSize.toNumber(), chunk.length);
-      const packet = chunk.slice(0, packetSize);
-      chunk = chunk.slice(packetSize);
-      const offset = new Bigint(this.offset);
-      this.offset = this.offset.add(packetSize);
-
-      const retryInterval = fibonacci();
-      let pending = true;
-
-      while (pending) {
-        try {
-          await requestAsync(
-            'write',
-            {
-              FileId: this.file.FileId,
-              Offset: offset.toBuffer(),
-              Buffer: packet,
-            },
-            this.connection
-          );
-
-          pending = false;
-        } catch (error) {
-          if (error.code === 'STATUS_PENDING') {
-            await new Promise((resolve, reject) => {
-              setTimeout(resolve, retryInterval.next().value);
-            });
-          } else {
-            throw error;
-          }
-        }
-      }
-    }
-
-    next();
-  }
-
-  async end(...args) {
-    try {
-      super.end(...args);
-    } finally {
-      await requestAsync('close', this.file, this.connection);
-    }
-  }
-}
-
-export default function(path, options, cb) {
-  if (typeof options === 'function') {
-    cb = options;
-    options = {};
-  }
-
-  let createDisposition;
-  const flags = options && options.flags;
-
-  if (flags === 'r') {
-    createDisposition = FILE_OPEN;
-  } else if (flags === 'r+') {
-    createDisposition = FILE_OPEN_IF;
-  } else if (flags === 'w' || flags === 'w+') {
-    createDisposition = FILE_OVERWRITE_IF;
-  } else if (flags === 'wx' || flags === 'w+x') {
-    createDisposition = FILE_CREATE;
-  }
-
-  request('create', { path, createDisposition }, this, (err, file) => {
-    if (err) {
-      cb(err);
-    } else {
-      cb(null, new SmbWritableStream(this, file, options));
-    }
-  });
-}