Solid Gold!!!

conditional-pipe.js

#!/usr/bin/env node
'use strict';
const AWS = require('aws-sdk');
if (process.env.AWS_PROFILE) { // eslint-disable-line
  AWS.config.credentials = new AWS.SharedIniFileCredentials({
    profile: process.env.AWS_PROFILE // eslint-disable-line
  });
  AWS.config.logger = console;
}
const s3opts = {};
// process.env.AWS_S3_ENDPOINT stuff is for using with MINIO or other
// S3 compatible storage.
// https://github.com/minio/cookbook/blob/master/docs/aws-sdk-for-javascript-with-minio.md
if (process.env.AWS_S3_ENDPOINT) { // eslint-disable-line
  s3opts.endpoint = process.env.AWS_S3_ENDPOINT; // eslint-disable-line
  s3opts.s3ForcePathStyle = true;
}
const s3 = new AWS.S3(s3opts);
const { Transform } = require('stream');
const pumpify = require('pumpify');
const split2 = require('split2');
class AllowLines extends Transform {
  constructor(options) {
    options = options || {};
    options.objectMode = true;
    super(options);
    this.allow = options.allow || [];
  }
  _transform(string, x, done) {
    if (this.allow.includes(string)) {
      this.emit('matched');
      this.push(`${string}\n`);
    }
    // done(new Error('AGGGH!'));
    done();
  }
}
const allowA = [
  '2019/09/01/03/18/45/hexgbg649-ingest-0-355354a4-cc67-11e9-b572-0cc47aa318b6.gz',
  '2019/09/01/03/28/34/hexgbg649-ingest-2-9459e5b6-cc68-11e9-ae81-0cc47aa3197c.gz',
  '2019/09/01/03/28/35/hexgbg649-ingest-2-94b5d344-cc68-11e9-b494-0cc47aa3186e.gz',
  '2019/09/30/17/37/19/hexgbg649-ingest-6-f3b0ce92-e3a8-11e9-b8e9-0cc47aa318b4.gz',
  '2019/09/30/17/36/51/hexgbg649-ingest-2-1f3cfc67-7920-4198-9021-0dc56585ef88.gz',
  '2019/09/30/17/37/01/hexgbg649-ingest-2-97e35e15-8890-44aa-b4e2-98243bb4a1c0.gz',
  '2019/09/30/17/37/03/hexgbg649-ingest-2-ea0b47fa-e3a8-11e9-91e0-0cc47aa319ca.gz'
];
const run = async () => {
  const src = require('fs').createReadStream('parsed-keys/hexgbg649-112.txt');
  const allowStr = new AllowLines({ allow: allowA });
  const str = pumpify(src, split2(), allowStr);
  await new Promise((resolve, reject) => {
    // throw new Error('?????');
    let matched = false;
    str.on('finish', () => {
      console.log('finish');
      if (!matched) {
        console.log('finish resolve');
        resolve();
      }
    });
    str.on('error', (err) => {
      console.log('on error');
      if (!matched) {
        console.log('on error reject');
        reject(err);
      }
    });
    // end never happens if there's no data.
    // str.on('end', () => {
    //   console.log('end');
    //   if (!matched) {
    //     console.log('end resolve');
    //     resolve();
    //   }
    // });
    allowStr.once('matched', async () => {
      matched = true;
      const params = {
        Bucket: process.env.BUCKET, // eslint-disable-line
        Key: 'conditional-pipe',
        Body: str
      };
      await s3.upload(params).promise().then(() => {
        console.log('upload then');
        resolve();
      }).catch((err) => {
        console.log('upload catch');
        reject(err);
      });
    });
  });
  console.log('exiting run');
  return 'some results';
};
run()
  .then((result) => { console.log('run done', result); })
  .catch((err) => { console.error('run error', err.stack); });