How to use the mz/fs.createWriteStream function in mz

async upload() {
    const { ctx } = this;
    const files = ctx.request.files;
    ctx.logger.warn('files: %j', files);

    try {
      for (const file of files) {
        const filename = file.filename.toLowerCase();
        const targetPath = path.join(this.config.baseDir, 'app/public', filename);
        const source = fs.createReadStream(file.filepath);
        const target = fs.createWriteStream(targetPath);
        await pump(source, target);
        ctx.logger.warn('save %s to %s', file.filepath, targetPath);
      }
    } finally {
      // delete those request tmp files
      await ctx.cleanupRequestFiles();
    }

    const fields = [];
    for (const k in ctx.request.body) {
      fields.push({
        key: k,
        value: ctx.request.body[k],
      });
    }

exports.getStream = async url => {
  const dir = path.join(__dirname, 'tmp');
  await mkdirp(dir);
  const file = path.join(dir, utility.md5(url) + path.extname(url));
  if (await fs.exists(file)) {
    return fs.createReadStream(file);
  }

  const writeStream = fs.createWriteStream(file);
  await urllib.request(url, {
    writeStream,
    timeout: 10000,
    followRedirect: true,
  });
  return fs.createReadStream(file);
};

yield new Promise((resolve, reject) => {
    const path = Path.join(config.UPLOAD_PATH, _encodePath(data.filePath));
    console.log('saving file to', path);
    const stream = fs.createWriteStream(path);
    this.req.pipe(stream);
    this.req.on('end', () => {
      files[data.filePath] = data;
      resolve();
    });
    this.req.on('error', reject);
  });
  this.body = 'fake s3 upload ok';

function writeFile(fileName: string, data: string | Buffer, instrumenter: NodeJS.ReadWriteStream | null = null): Promise {
  if (Buffer.isBuffer(data)) {
    return fs.writeFile(fileName, data);
  } else if (instrumenter) {
    instrumenter.pipe(fs.createWriteStream(fileName, 'utf8'));
    return writeToStream(data, instrumenter);
  } else {
    return fs.writeFile(fileName, data, 'utf8');
  }
}

return new Promise((resolve, reject) => {
    let out = fs.createWriteStream(filename);
    stream
      .on('data', chunk => {
        if (md5checksum != null) {
          hasher.update(chunk);
        }
      })
      .pipe(out)
      .on('error', err => {
        reject(err);
      })
      .on('finish', () => {
        let actualChecksum = hasher.digest('hex');
        if (md5checksum != null) {
          if (actualChecksum !== md5checksum) {
            reject(new Error(`Incorrect md5sum (expected ${md5checksum}, got ${actualChecksum})`))
            return;

await new Promise( ( resolve, reject ) =>
                data.pipe( fs.createWriteStream( file ) ).on( 'error', reject ).on( 'finish', resolve )
            );

.then(res => {
        const stream = fs.createWriteStream(destination);
        res.body.pipe(stream);
        return new Promise((resolve, reject) => {
          res.body.on('end', () => resolve(destination));
          res.body.on('error', err => reject(err));
        });
      });
  }

async function downloadFile (forceDownload = false) {
  if (forceDownload || !await fs.existsSync(productsFile) || !await fs.existsSync(termsFile)) {
    if (forceDownload || !await fs.existsSync(offersFile)) {
      const stream = fs.createWriteStream(offersFile)
      request.get(offers).pipe(stream)
      await new Promise(fulfill => stream.on('finish', fulfill))
    }
    await exec(`cat ${offersFile} | npx JSONStream 'products.*' > ${productsFile}`)
    await exec(`cat ${offersFile} | npx JSONStream 'terms.OnDemand.*' > ${termsFile}`)
  }
  return true
}

yield function (done) {
      fs.createReadStream(path)
      .on('error', done)
      .pipe(zlib.createGzip())
      .on('error', done)
      .pipe(fs.createWriteStream(tmp))
      .on('error', done)
      .on('finish', done)
    }