const assert = require('assert');

const crypto = require('crypto');

const stream = require('stream');

const util = require('util');

const zlib = require('zlib');

const Stream = stream.Stream;

// emit random bytes, and keep a shasum

function RandomReadStream(opt) {

Stream.call(this);

class RandomReadStream extends Stream {

constructor(opt) {

super();

this.readable = true;

this._paused = false;

this._processing = false;

this._hasher = crypto.createHash('sha1');

opt = opt || {};

// base block size.

opt.block = opt.block || 256 * 1024;

this.readable = true;

this._paused = false;

this._processing = false;

// total number of bytes to emit

opt.total = opt.total || 256 * 1024 * 1024;

this._remaining = opt.total;

this._hasher = crypto.createHash('sha1');

opt = opt || {};

// how variable to make the block sizes

opt.jitter = opt.jitter || 1024;

// base block size.

opt.block = opt.block || 256 * 1024;

this._opt = opt;

// total number of bytes to emit

opt.total = opt.total || 256 * 1024 * 1024;

this._remaining = opt.total;

this._process = this._process.bind(this);

// how variable to make the block sizes

opt.jitter = opt.jitter || 1024;

process.nextTick(this._process);

}

this._opt = opt;

util.inherits(RandomReadStream, Stream);

this._process = this._process.bind(this);

RandomReadStream.prototype.pause = function() {

this._paused = true;

this.emit('pause');

};

process.nextTick(this._process);

}

RandomReadStream.prototype.resume = function() {

// console.error("rrs resume");

this._paused = false;

this.emit('resume');

this._process();

};

pause() {

this._paused = true;

this.emit('pause');

}

RandomReadStream.prototype._process = function() {

if (this._processing) return;

if (this._paused) return;

resume() {

// console.error("rrs resume");

this._paused = false;

this.emit('resume');

this._process();

}

this._processing = true;

_process() {

if (this._processing) return;

if (this._paused) return;

if (!this._remaining) {

this._hash = this._hasher.digest('hex').toLowerCase().trim();

this._processing = false;

this._processing = true;

this.emit('end');

return;

}

if (!this._remaining) {

this._hash = this._hasher.digest('hex').toLowerCase().trim();

this._processing = false;

// figure out how many bytes to output

// if finished, then just emit end.

let block = this._opt.block;

const jitter = this._opt.jitter;

if (jitter) {

block += Math.ceil(Math.random() * jitter - (jitter / 2));

}

block = Math.min(block, this._remaining);

const buf = Buffer.allocUnsafe(block);

for (let i = 0; i < block; i++) {

buf[i] = Math.random() * 256;

}

this.emit('end');

return;

}

this._hasher.update(buf);

// figure out how many bytes to output

// if finished, then just emit end.

let block = this._opt.block;

const jitter = this._opt.jitter;

if (jitter) {

block += Math.ceil(Math.random() * jitter - (jitter / 2));

}

block = Math.min(block, this._remaining);

const buf = Buffer.allocUnsafe(block);

for (let i = 0; i < block; i++) {

buf[i] = Math.random() * 256;

}

this._remaining -= block;

this._hasher.update(buf);

console.error('block=%d\nremain=%d\n', block, this._remaining);

this._processing = false;

this._remaining -= block;

this.emit('data', buf);

process.nextTick(this._process);

};

this._processing = false;

this.emit('data', buf);

process.nextTick(this._process);

}

}

// a filter that just verifies a shasum

function HashStream() {

Stream.call(this);

class HashStream extends Stream {

constructor() {

super();

this.readable = this.writable = true;

this._hasher = crypto.createHash('sha1');

}

this.readable = this.writable = true;

this._hasher = crypto.createHash('sha1');

}

write(c) {

// Simulate the way that an fs.ReadStream returns false

// on *every* write, only to resume a moment later.

this._hasher.update(c);

process.nextTick(() => this.resume());

return false;

}

resume() {

this.emit('resume');

process.nextTick(() => this.emit('drain'));

}

util.inherits(HashStream, Stream);

HashStream.prototype.write = function(c) {

// Simulate the way that an fs.ReadStream returns false

// on *every* write like a jerk, only to resume a

// moment later.

this._hasher.update(c);

process.nextTick(this.resume.bind(this));

return false;

};

HashStream.prototype.resume = function() {

this.emit('resume');

process.nextTick(this.emit.bind(this, 'drain'));

};

HashStream.prototype.end = function(c) {

if (c) {

this.write(c);

end(c) {

if (c) {

this.write(c);

}

this._hash = this._hasher.digest('hex').toLowerCase().trim();

this.emit('data', this._hash);

this.emit('end');

}

this._hash = this._hasher.digest('hex').toLowerCase().trim();

this.emit('data', this._hash);

this.emit('end');

};

}

const inp = new RandomReadStream({ total: 1024, block: 256, jitter: 16 });

inp.pipe(gzip).pipe(gunz).pipe(out);

inp.on('data', function(c) {

console.error('inp data', c.length);

});

gzip.on('data', function(c) {

console.error('gzip data', c.length);

});

gunz.on('data', function(c) {

console.error('gunz data', c.length);

});

out.on('data', function(c) {

console.error('out data', c.length);

});

out.on('data', common.mustCall(function(c) {

console.error('hash=%s', c);

out.on('data', common.mustCall((c) => {

assert.strictEqual(c, inp._hash, 'hashes should match');

}));