sonic-boom - npm Package Compare versions

Comparing version 3.5.0 to 3.6.0

index.js

@@ -339,11 +339,49 @@ 'use strict'
 
-function flush () {
+function callFlushCallbackOnDrain (cb) {
+  const onDrain = () => {
+    // only if _fsync is false to avoid double fsync
+    if (!this._fsync) {
+      fs.fsync(this.fd, cb)
+    } else {
+      cb()
+    }
+    this.off('error', onError)
+  }
+  const onError = (err) => {
+    cb(err)
+    this.off('drain', onDrain)
+  }
+
+  this.once('drain', onDrain)
+  this.once('error', onError)
+}
+
+function flush (cb) {
+  if (cb != null && typeof cb !== 'function') {
+    throw new Error('flush cb must be a function')
+  }
+
   if (this.destroyed) {
-    throw new Error('SonicBoom destroyed')
+    const error = new Error('SonicBoom destroyed')
+    if (cb) {
+      cb(error)
+      return
+    }
+
+    throw error
   }
 
-  if (this._writing || this.minLength <= 0) {
+  if (this.minLength <= 0) {
+    cb?.()
     return
   }
 
+  if (cb) {
+    callFlushCallbackOnDrain.call(this, cb)
+  }
+
+  if (this._writing) {
+    return
+  }
+
   if (this._bufs.length === 0) {
@@ -356,11 +394,30 @@ this._bufs.push('')
 
-function flushBuffer () {
+function flushBuffer (cb) {
+  if (cb != null && typeof cb !== 'function') {
+    throw new Error('flush cb must be a function')
+  }
+
   if (this.destroyed) {
-    throw new Error('SonicBoom destroyed')
+    const error = new Error('SonicBoom destroyed')
+    if (cb) {
+      cb(error)
+      return
+    }
+
+    throw error
   }
 
-  if (this._writing || this.minLength <= 0) {
+  if (this.minLength <= 0) {
+    cb?.()
     return
   }
 
+  if (cb) {
+    callFlushCallbackOnDrain.call(this, cb)
+  }
+
+  if (this._writing) {
+    return
+  }
+
   if (this._bufs.length === 0) {
@@ -367,0 +424,0 @@ this._bufs.push([])

package.json

 {
   "name": "sonic-boom",
-  "version": "3.5.0",
+  "version": "3.6.0",
   "description": "Extremely fast utf8 only stream implementation",
@@ -5,0 +5,0 @@ "main": "index.js",

README.md

@@ -86,3 +86,3 @@ # sonic-boom
 
-### SonicBoom#flush()
+### SonicBoom#flush([cb])
 
@@ -92,2 +92,4 @@ Writes the current buffer to the file if a write was not in progress.
 
+call the callback when the flush operation is completed. when failed the callback is called with an error.
+
 ### SonicBoom#reopen([file])
@@ -94,0 +96,0 @@

test/flush.test.js

@@ -7,2 +7,3 @@ 'use strict'
 const { file, runTests } = require('./helper')
+const proxyquire = require('proxyquire')
 
@@ -104,2 +105,274 @@ runTests(buildTests)
   })
+
+  test('call flush cb after flushed', (t) => {
+    t.plan(4)
+
+    const dest = file()
+    const fd = fs.openSync(dest, 'w')
+    const stream = new SonicBoom({ fd, minLength: 4096, sync })
+
+    stream.on('ready', () => {
+      t.pass('ready emitted')
+    })
+
+    t.ok(stream.write('hello world\n'))
+    t.ok(stream.write('something else\n'))
+
+    stream.flush((err) => {
+      if (err) t.fail(err)
+      else t.pass('flush cb called')
+    })
+  })
+
+  test('only call fsyncSync and not fsync when fsync: true', (t) => {
+    t.plan(6)
+
+    const fakeFs = Object.create(fs)
+    const SonicBoom = proxyquire('../', {
+      fs: fakeFs
+    })
+
+    const dest = file()
+    const fd = fs.openSync(dest, 'w')
+    const stream = new SonicBoom({
+      fd,
+      sync,
+      fsync: true,
+      minLength: 4096
+    })
+
+    stream.on('ready', () => {
+      t.pass('ready emitted')
+    })
+
+    fakeFs.fsync = function (fd, cb) {
+      t.fail('fake fs.fsync called while should not')
+      cb()
+    }
+    fakeFs.fsyncSync = function (fd) {
+      t.pass('fake fsyncSync called')
+    }
+
+    function successOnAsyncOrSyncFn (isSync, originalFn) {
+      return function (...args) {
+        t.pass(`fake fs.${originalFn.name} called`)
+        fakeFs[originalFn.name] = originalFn
+        return fakeFs[originalFn.name](...args)
+      }
+    }
+
+    if (sync) {
+      fakeFs.writeSync = successOnAsyncOrSyncFn(true, fs.writeSync)
+    } else {
+      fakeFs.write = successOnAsyncOrSyncFn(false, fs.write)
+    }
+
+    t.ok(stream.write('hello world\n'))
+    stream.flush((err) => {
+      if (err) t.fail(err)
+      else t.pass('flush cb called')
+
+      process.nextTick(() => {
+        // to make sure fsync is not called as well
+        t.pass('nextTick after flush called')
+      })
+    })
+  })
+
+  test('call flush cb with error when fsync failed', (t) => {
+    t.plan(5)
+
+    const fakeFs = Object.create(fs)
+    const SonicBoom = proxyquire('../', {
+      fs: fakeFs
+    })
+
+    const dest = file()
+    const fd = fs.openSync(dest, 'w')
+    const stream = new SonicBoom({
+      fd,
+      sync,
+      minLength: 4096
+    })
+
+    stream.on('ready', () => {
+      t.pass('ready emitted')
+    })
+
+    const err = new Error('other')
+    err.code = 'other'
+
+    function onFsyncOnFsyncSync (isSync, originalFn) {
+      return function (...args) {
+        Error.captureStackTrace(err)
+        t.pass(`fake fs.${originalFn.name} called`)
+        fakeFs[originalFn.name] = originalFn
+        const cb = args[args.length - 1]
+
+        cb(err)
+      }
+    }
+
+    // only one is called depends on sync
+    fakeFs.fsync = onFsyncOnFsyncSync(false, fs.fsync)
+
+    function successOnAsyncOrSyncFn (isSync, originalFn) {
+      return function (...args) {
+        t.pass(`fake fs.${originalFn.name} called`)
+        fakeFs[originalFn.name] = originalFn
+        return fakeFs[originalFn.name](...args)
+      }
+    }
+
+    if (sync) {
+      fakeFs.writeSync = successOnAsyncOrSyncFn(true, fs.writeSync)
+    } else {
+      fakeFs.write = successOnAsyncOrSyncFn(false, fs.write)
+    }
+
+    t.ok(stream.write('hello world\n'))
+    stream.flush((err) => {
+      if (err) t.equal(err.code, 'other')
+      else t.fail('flush cb called without an error')
+    })
+  })
+
+  test('call flush cb even when have no data', (t) => {
+    t.plan(2)
+
+    const dest = file()
+    const fd = fs.openSync(dest, 'w')
+    const stream = new SonicBoom({ fd, minLength: 4096, sync })
+
+    stream.on('ready', () => {
+      t.pass('ready emitted')
+
+      stream.flush((err) => {
+        if (err) t.fail(err)
+        else t.pass('flush cb called')
+      })
+    })
+  })
+
+  test('call flush cb even when minLength is 0', (t) => {
+    t.plan(1)
+
+    const dest = file()
+    const fd = fs.openSync(dest, 'w')
+    const stream = new SonicBoom({ fd, minLength: 0, sync })
+
+    stream.flush((err) => {
+      if (err) t.fail(err)
+      else t.pass('flush cb called')
+    })
+  })
+
+  test('call flush cb with an error when trying to flush destroyed stream', (t) => {
+    t.plan(1)
+
+    const dest = file()
+    const fd = fs.openSync(dest, 'w')
+    const stream = new SonicBoom({ fd, minLength: 4096, sync })
+    stream.destroy()
+
+    stream.flush((err) => {
+      if (err) t.pass(err)
+      else t.fail('flush cb called without an error')
+    })
+  })
+
+  test('call flush cb with an error when failed to flush', (t) => {
+    t.plan(5)
+
+    const fakeFs = Object.create(fs)
+    const SonicBoom = proxyquire('../', {
+      fs: fakeFs
+    })
+
+    const dest = file()
+    const fd = fs.openSync(dest, 'w')
+    const stream = new SonicBoom({
+      fd,
+      sync,
+      minLength: 4096
+    })
+
+    stream.on('ready', () => {
+      t.pass('ready emitted')
+    })
+
+    const err = new Error('other')
+    err.code = 'other'
+
+    function onWriteOrWriteSync (isSync, originalFn) {
+      return function (...args) {
+        Error.captureStackTrace(err)
+        t.pass(`fake fs.${originalFn.name} called`)
+        fakeFs[originalFn.name] = originalFn
+
+        if (isSync) throw err
+        const cb = args[args.length - 1]
+
+        cb(err)
+      }
+    }
+
+    // only one is called depends on sync
+    fakeFs.write = onWriteOrWriteSync(false, fs.write)
+    fakeFs.writeSync = onWriteOrWriteSync(true, fs.writeSync)
+
+    t.ok(stream.write('hello world\n'))
+    stream.flush((err) => {
+      if (err) t.equal(err.code, 'other')
+      else t.fail('flush cb called without an error')
+    })
+
+    stream.end()
+
+    stream.on('close', () => {
+      t.pass('close emitted')
+    })
+  })
+
+  test('call flush cb when finish writing when currently in the middle', (t) => {
+    t.plan(4)
+
+    const fakeFs = Object.create(fs)
+    const SonicBoom = proxyquire('../', {
+      fs: fakeFs
+    })
+
+    const dest = file()
+    const fd = fs.openSync(dest, 'w')
+    const stream = new SonicBoom({
+      fd,
+      sync,
+
+      // to trigger write without calling flush
+      minLength: 1
+    })
+
+    stream.on('ready', () => {
+      t.pass('ready emitted')
+    })
+
+    function onWriteOrWriteSync (originalFn) {
+      return function (...args) {
+        stream.flush((err) => {
+          if (err) t.fail(err)
+          else t.pass('flush cb called')
+        })
+
+        t.pass(`fake fs.${originalFn.name} called`)
+        fakeFs[originalFn.name] = originalFn
+        return originalFn(...args)
+      }
+    }
+
+    // only one is called depends on sync
+    fakeFs.write = onWriteOrWriteSync(fs.write)
+    fakeFs.writeSync = onWriteOrWriteSync(fs.writeSync)
+
+    t.ok(stream.write('hello world\n'))
+  })
 }

types/index.d.ts

@@ -41,3 +41,3 @@ // Type definitions for sonic-boom 0.7
      */
-    flush(): void;
+    flush(cb?: (err?: Error) => unknown): void;
 
@@ -44,0 +44,0 @@ /**

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

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

78655

increased by10.68%

Lines of code

2515

increased by11.73%

Number of lines in readme file

152

increased by1.33%

No dependency changes