strong-tie/inbound-calls

1'use strict'2 3const { test } = require('tap')4const fs = require('fs')5const proxyquire = require('proxyquire')6const SonicBoom = require('../')7const { file, runTests } = require('./helper')8 9runTests(buildTests)10 11function buildTests (test, sync) {12  // Reset the umask for testing13  process.umask(0o000)14 15  test('write things to a file descriptor', (t) => {16    t.plan(6)17 18    const dest = file()19    const fd = fs.openSync(dest, 'w')20    const stream = new SonicBoom({ fd, sync })21 22    stream.on('ready', () => {23      t.pass('ready emitted')24    })25 26    t.ok(stream.write('hello world\n'))27    t.ok(stream.write('something else\n'))28 29    stream.end()30 31    stream.on('finish', () => {32      fs.readFile(dest, 'utf8', (err, data) => {33        t.error(err)34        t.equal(data, 'hello world\nsomething else\n')35      })36    })37    stream.on('close', () => {38      t.pass('close emitted')39    })40  })41 42  test('write things in a streaming fashion', (t) => {43    t.plan(8)44 45    const dest = file()46    const fd = fs.openSync(dest, 'w')47    const stream = new SonicBoom({ fd, sync })48 49    stream.once('drain', () => {50      fs.readFile(dest, 'utf8', (err, data) => {51        t.error(err)52        t.equal(data, 'hello world\n')53        t.ok(stream.write('something else\n'))54      })55 56      stream.once('drain', () => {57        fs.readFile(dest, 'utf8', (err, data) => {58          t.error(err)59          t.equal(data, 'hello world\nsomething else\n')60          stream.end()61        })62      })63    })64 65    t.ok(stream.write('hello world\n'))66 67    stream.on('finish', () => {68      t.pass('finish emitted')69    })70    stream.on('close', () => {71      t.pass('close emitted')72    })73  })74 75  test('can be piped into', (t) => {76    t.plan(4)77 78    const dest = file()79    const fd = fs.openSync(dest, 'w')80    const stream = new SonicBoom({ fd, sync })81    const source = fs.createReadStream(__filename, { encoding: 'utf8' })82 83    source.pipe(stream)84 85    stream.on('finish', () => {86      fs.readFile(__filename, 'utf8', (err, expected) => {87        t.error(err)88        fs.readFile(dest, 'utf8', (err, data) => {89          t.error(err)90          t.equal(data, expected)91        })92      })93    })94    stream.on('close', () => {95      t.pass('close emitted')96    })97  })98 99  test('write things to a file', (t) => {100    t.plan(6)101 102    const dest = file()103    const stream = new SonicBoom({ dest, sync })104 105    stream.on('ready', () => {106      t.pass('ready emitted')107    })108 109    t.ok(stream.write('hello world\n'))110    t.ok(stream.write('something else\n'))111 112    stream.end()113 114    stream.on('finish', () => {115      fs.readFile(dest, 'utf8', (err, data) => {116        t.error(err)117        t.equal(data, 'hello world\nsomething else\n')118      })119    })120    stream.on('close', () => {121      t.pass('close emitted')122    })123  })124 125  test('minLength', (t) => {126    t.plan(8)127 128    const dest = file()129    const stream = new SonicBoom({ dest, minLength: 4096, sync })130 131    stream.on('ready', () => {132      t.pass('ready emitted')133    })134 135    t.ok(stream.write('hello world\n'))136    t.ok(stream.write('something else\n'))137 138    const fail = t.fail139    stream.on('drain', fail)140 141    // bad use of timer142    // TODO refactor143    setTimeout(function () {144      fs.readFile(dest, 'utf8', (err, data) => {145        t.error(err)146        t.equal(data, '')147 148        stream.end()149 150        stream.on('finish', () => {151          fs.readFile(dest, 'utf8', (err, data) => {152            t.error(err)153            t.equal(data, 'hello world\nsomething else\n')154          })155        })156      })157    }, 100)158 159    stream.on('close', () => {160      t.pass('close emitted')161    })162  })163 164  test('write later on recoverable error', (t) => {165    t.plan(8)166 167    const fakeFs = Object.create(fs)168    const SonicBoom = proxyquire('../', {169      fs: fakeFs170    })171 172    const dest = file()173    const fd = fs.openSync(dest, 'w')174    const stream = new SonicBoom({ fd, minLength: 0, sync })175 176    stream.on('ready', () => {177      t.pass('ready emitted')178    })179    stream.on('error', () => {180      t.pass('error emitted')181    })182 183    if (sync) {184      fakeFs.writeSync = function (fd, buf, enc) {185        t.pass('fake fs.writeSync called')186        throw new Error('recoverable error')187      }188    } else {189      fakeFs.write = function (fd, buf, ...args) {190        t.pass('fake fs.write called')191        setTimeout(() => args.pop()(new Error('recoverable error')), 0)192      }193    }194 195    t.ok(stream.write('hello world\n'))196 197    setTimeout(() => {198      if (sync) {199        fakeFs.writeSync = fs.writeSync200      } else {201        fakeFs.write = fs.write202      }203 204      t.ok(stream.write('something else\n'))205 206      stream.end()207      stream.on('finish', () => {208        fs.readFile(dest, 'utf8', (err, data) => {209          t.error(err)210          t.equal(data, 'hello world\nsomething else\n')211        })212      })213      stream.on('close', () => {214        t.pass('close emitted')215      })216    }, 0)217  })218 219  test('emit write events', (t) => {220    t.plan(7)221 222    const dest = file()223    const stream = new SonicBoom({ dest, sync })224 225    stream.on('ready', () => {226      t.pass('ready emitted')227    })228 229    let length = 0230    stream.on('write', (bytes) => {231      length += bytes232    })233 234    t.ok(stream.write('hello world\n'))235    t.ok(stream.write('something else\n'))236 237    stream.end()238 239    stream.on('finish', () => {240      fs.readFile(dest, 'utf8', (err, data) => {241        t.error(err)242        t.equal(data, 'hello world\nsomething else\n')243        t.equal(length, 27)244      })245    })246    stream.on('close', () => {247      t.pass('close emitted')248    })249  })250 251  test('write multi-byte characters string over than maxWrite', (t) => {252    const fakeFs = Object.create(fs)253    const MAX_WRITE = 65535254    fakeFs.write = function (fd, buf, ...args) {255      // only write byteLength === MAX_WRITE256      const _buf = Buffer.from(buf).subarray(0, MAX_WRITE).toString()257      fs.write(fd, _buf, ...args)258      setImmediate(args[args.length - 1], null, MAX_WRITE)259      fakeFs.write = function (fd, buf, ...args) {260        fs.write(fd, buf, ...args)261      }262    }263    const SonicBoom = proxyquire('../', {264      fs: fakeFs265    })266    const dest = file()267    const fd = fs.openSync(dest, 'w')268    const stream = new SonicBoom({ fd, minLength: 0, sync, maxWrite: MAX_WRITE })269    let buf = Buffer.alloc(MAX_WRITE).fill('x')270    buf = '🌲' + buf.toString()271    stream.write(buf)272    stream.end()273 274    stream.on('finish', () => {275      fs.readFile(dest, 'utf8', (err, data) => {276        t.error(err)277        t.equal(data, buf)278        t.end()279      })280    })281    stream.on('close', () => {282      t.pass('close emitted')283    })284    stream.on('error', () => {285      t.pass('error emitted')286    })287  })288}289 290test('write buffers that are not totally written', (t) => {291  t.plan(9)292 293  const fakeFs = Object.create(fs)294  fakeFs.write = function (fd, buf, ...args) {295    t.pass('fake fs.write called')296    fakeFs.write = function (fd, buf, ...args) {297      t.pass('calling real fs.write, ' + buf)298      fs.write(fd, buf, ...args)299    }300    process.nextTick(args[args.length - 1], null, 0)301  }302  const SonicBoom = proxyquire('../', {303    fs: fakeFs304  })305 306  const dest = file()307  const fd = fs.openSync(dest, 'w')308  const stream = new SonicBoom({ fd, minLength: 0, sync: false })309 310  stream.on('ready', () => {311    t.pass('ready emitted')312  })313 314  t.ok(stream.write('hello world\n'))315  t.ok(stream.write('something else\n'))316 317  stream.end()318 319  stream.on('finish', () => {320    fs.readFile(dest, 'utf8', (err, data) => {321      t.error(err)322      t.equal(data, 'hello world\nsomething else\n')323    })324  })325  stream.on('close', () => {326    t.pass('close emitted')327  })328})329 330test('write enormously large buffers async', (t) => {331  t.plan(3)332 333  const dest = file()334  const fd = fs.openSync(dest, 'w')335  const stream = new SonicBoom({ fd, minLength: 0, sync: false })336 337  const buf = Buffer.alloc(1024).fill('x').toString() // 1 MB338  let length = 0339 340  for (let i = 0; i < 1024 * 512; i++) {341    length += buf.length342    stream.write(buf)343  }344 345  stream.end()346 347  stream.on('finish', () => {348    fs.stat(dest, (err, stat) => {349      t.error(err)350      t.equal(stat.size, length)351    })352  })353  stream.on('close', () => {354    t.pass('close emitted')355  })356})357 358test('make sure `maxWrite` is passed', (t) => {359  t.plan(1)360  const dest = file()361  const stream = new SonicBoom({ dest, maxLength: 65536 })362  t.equal(stream.maxLength, 65536)363})364 365test('write enormously large buffers async atomicly', (t) => {366  const fakeFs = Object.create(fs)367  const SonicBoom = proxyquire('../', {368    fs: fakeFs369  })370 371  const dest = file()372  const fd = fs.openSync(dest, 'w')373  const stream = new SonicBoom({ fd, minLength: 0, sync: false })374 375  const buf = Buffer.alloc(1023).fill('x').toString()376 377  fakeFs.write = function (fd, _buf, ...args) {378    if (_buf.length % buf.length !== 0) {379      t.fail('write called with wrong buffer size')380    }381 382    setImmediate(args[args.length - 1], null, _buf.length)383  }384 385  for (let i = 0; i < 1024 * 512; i++) {386    stream.write(buf)387  }388 389  setImmediate(() => {390    for (let i = 0; i < 1024 * 512; i++) {391      stream.write(buf)392    }393 394    stream.end()395  })396 397  stream.on('close', () => {398    t.pass('close emitted')399    t.end()400  })401})402 403test('write should not drop new data if buffer is not full', (t) => {404  t.plan(2)405  const fakeFs = Object.create(fs)406  const SonicBoom = proxyquire('../', {407    fs: fakeFs408  })409 410  const dest = file()411  const fd = fs.openSync(dest, 'w')412  const stream = new SonicBoom({ fd, minLength: 101, maxLength: 102, sync: false })413 414  const buf = Buffer.alloc(100).fill('x').toString()415 416  fakeFs.write = function (fd, _buf, ...args) {417    t.equal(_buf.length, buf.length + 2)418    setImmediate(args[args.length - 1], null, _buf.length)419    fakeFs.write = () => t.error('shouldnt call write again')420    stream.end()421  }422 423  stream.on('drop', (data) => {424    t.error('should not drop')425  })426 427  stream.write(buf)428  stream.write('aa')429 430  stream.on('close', () => {431    t.pass('close emitted')432  })433})434 435test('write should drop new data if buffer is full', (t) => {436  t.plan(3)437  const fakeFs = Object.create(fs)438  const SonicBoom = proxyquire('../', {439    fs: fakeFs440  })441 442  const dest = file()443  const fd = fs.openSync(dest, 'w')444  const stream = new SonicBoom({ fd, minLength: 101, maxLength: 102, sync: false })445 446  const buf = Buffer.alloc(100).fill('x').toString()447 448  fakeFs.write = function (fd, _buf, ...args) {449    t.equal(_buf.length, buf.length)450    setImmediate(args[args.length - 1], null, _buf.length)451    fakeFs.write = () => t.error('shouldnt call write more than once')452  }453 454  stream.on('drop', (data) => {455    t.equal(data.length, 3)456    stream.end()457  })458 459  stream.write(buf)460  stream.write('aaa')461 462  stream.on('close', () => {463    t.pass('close emitted')464  })465})466