Retry on ebusy (pinojs#170)

* Retry on EBUSY for better Windows compatibility

Signed-off-by: Matteo Collina <hello@matteocollina.com>

* added docs and test

Signed-off-by: Matteo Collina <hello@matteocollina.com>

---------

Signed-off-by: Matteo Collina <hello@matteocollina.com>

README.md

@@ -68,7 +68,7 @@ The options are:
 * `mode`: specify the creating file `mode` (see [fs.open()](https://nodejs.org/api/fs.html#fsopenpath-flags-mode-callback) from Node.js core).
 * `mkdir`: ensure directory for dest file exists when `true` (default `false`).
 * `retryEAGAIN(err, writeBufferLen, remainingBufferLen)`: a function that will be called when sonic-boom
-    write/writeSync/flushSync encounters a EAGAIN error. If the return value is
+    write/writeSync/flushSync encounters a EAGAIN or EBUSY error. If the return value is
     true sonic-boom will retry the operation, otherwise it will bubble the
     error. `err` is the error that caused this function to be called,
     `writeBufferLen` is the length of the buffer sonic-boom tried to write, and

index.js

@@ -126,7 +126,7 @@ function SonicBoom (opts) {
 
   this.release = (err, n) => {
     if (err) {
-      if (err.code === 'EAGAIN' && this.retryEAGAIN(err, this._writingBuf.length, this._len - this._writingBuf.length)) {
+      if ((err.code === 'EAGAIN' || err.code === 'EBUSY') && this.retryEAGAIN(err, this._writingBuf.length, this._len - this._writingBuf.length)) {
         if (this.sync) {
           // This error code should not happen in sync mode, because it is
           // not using the underlining operating system asynchronous functions.
@@ -377,7 +377,8 @@ SonicBoom.prototype.flushSync = function () {
         this._bufs.shift()
       }
     } catch (err) {
-      if (err.code !== 'EAGAIN' || !this.retryEAGAIN(err, buf.length, this._len - buf.length)) {
+      const shouldRetry = err.code === 'EAGAIN' || err.code === 'EBUSY'
+      if (shouldRetry && !this.retryEAGAIN(err, buf.length, this._len - buf.length)) {
         throw err
       }
 

test/retry.test.js

@@ -320,3 +320,95 @@ test('retryEAGAIN receives remaining buffer if exceeds maxWrite', (t) => {
     t.pass('close emitted')
   })
 })
+
+test('retry on EBUSY', (t) => {
+  t.plan(7)
+
+  const fakeFs = Object.create(fs)
+  fakeFs.write = function (fd, buf, enc, cb) {
+    t.pass('fake fs.write called')
+    fakeFs.write = fs.write
+    const err = new Error('EBUSY')
+    err.code = 'EBUSY'
+    process.nextTick(cb, err)
+  }
+  const SonicBoom = proxyquire('..', {
+    fs: fakeFs
+  })
+
+  const dest = file()
+  const fd = fs.openSync(dest, 'w')
+  const stream = new SonicBoom({ fd, sync: false, minLength: 0 })
+
+  stream.on('ready', () => {
+    t.pass('ready emitted')
+  })
+
+  t.ok(stream.write('hello world\n'))
+  t.ok(stream.write('something else\n'))
+
+  stream.end()
+
+  stream.on('finish', () => {
+    fs.readFile(dest, 'utf8', (err, data) => {
+      t.error(err)
+      t.equal(data, 'hello world\nsomething else\n')
+    })
+  })
+  stream.on('close', () => {
+    t.pass('close emitted')
+  })
+})
+
+test('emit error on async EBUSY', (t) => {
+  t.plan(11)
+
+  const fakeFs = Object.create(fs)
+  fakeFs.write = function (fd, buf, enc, cb) {
+    t.pass('fake fs.write called')
+    fakeFs.write = fs.write
+    const err = new Error('EBUSY')
+    err.code = 'EBUSY'
+    process.nextTick(cb, err)
+  }
+  const SonicBoom = proxyquire('..', {
+    fs: fakeFs
+  })
+
+  const dest = file()
+  const fd = fs.openSync(dest, 'w')
+  const stream = new SonicBoom({
+    fd,
+    sync: false,
+    minLength: 12,
+    retryEAGAIN: (err, writeBufferLen, remainingBufferLen) => {
+      t.equal(err.code, 'EBUSY')
+      t.equal(writeBufferLen, 12)
+      t.equal(remainingBufferLen, 0)
+      return false
+    }
+  })
+
+  stream.on('ready', () => {
+    t.pass('ready emitted')
+  })
+
+  stream.once('error', err => {
+    t.equal(err.code, 'EBUSY')
+    t.ok(stream.write('something else\n'))
+  })
+
+  t.ok(stream.write('hello world\n'))
+
+  stream.end()
+
+  stream.on('finish', () => {
+    fs.readFile(dest, 'utf8', (err, data) => {
+      t.error(err)
+      t.equal(data, 'hello world\nsomething else\n')
+    })
+  })
+  stream.on('close', () => {
+    t.pass('close emitted')
+  })
+})