prevent memory leak when writing before ready

index.js

@@ -54,12 +54,12 @@ function openFile (file, sonic) {
       sonic.emit('ready')
     }
 
-    if (sonic._reopening) {
+    if (sonic._reopening || sonic.destroyed) {
       return
     }
 
     // start
-    if (!sonic._writing && sonic._len > sonic.minLength && !sonic.destroyed) {
+    if ((!sonic._writing && sonic._len > sonic.minLength) || sonic._flushPending) {
       sonic._actualWrite()
     }
   }
@@ -103,6 +103,7 @@ function SonicBoom (opts) {
   this._ending = false
   this._reopening = false
   this._asyncDrainScheduled = false
+  this._flushPending = false
   this._hwm = Math.max(minLength || 0, 16387)
   this.file = null
   this.destroyed = false
@@ -337,16 +338,22 @@ function writeBuffer (data) {
 }
 
 function callFlushCallbackOnDrain (cb) {
+  this._flushPending = true
   const onDrain = () => {
     // only if _fsync is false to avoid double fsync
     if (!this._fsync) {
-      fs.fsync(this.fd, cb)
+      fs.fsync(this.fd, (err) => {
+        this._flushPending = false
+        cb(err)
+      })
     } else {
+      this._flushPending = false
       cb()
     }
     this.off('error', onError)
   }
   const onError = (err) => {
+    this._flushPending = false
     cb(err)
     this.off('drain', onDrain)
   }

test/flush.test.js

@@ -373,4 +373,47 @@ function buildTests (test, sync) {
 
     t.ok(stream.write('hello world\n'))
   })
+
+  test('call flush cb when writing and trying to flush before ready (on async)', (t) => {
+    t.plan(4)
+
+    const fakeFs = Object.create(fs)
+    const SonicBoom = proxyquire('../', {
+      fs: fakeFs
+    })
+
+    fakeFs.open = fsOpen
+
+    const dest = file()
+    const stream = new SonicBoom({
+      fd: dest,
+      // only async as sync is part of the constructor so the user will not be able to call write/flush
+      // before ready
+      sync: false,
+
+      // to not trigger write without calling flush
+      minLength: 4096
+    })
+
+    stream.on('ready', () => {
+      t.pass('ready emitted')
+    })
+
+    function fsOpen (...args) {
+      process.nextTick(() => {
+        // try writing and flushing before ready and in the middle of opening
+        t.pass('fake fs.open called')
+        t.ok(stream.write('hello world\n'))
+
+        // calling flush
+        stream.flush((err) => {
+          if (err) t.fail(err)
+          else t.pass('flush cb called')
+        })
+
+        fakeFs.open = fs.open
+        fs.open(...args)
+      })
+    }
+  })
 }