Просто чтобы уточнить ответ Зильвины.Я покажу вам все, как я получил его на работу.
const athena = require('./athena')
const redis = require('./redis')
const zlib = require('zlib')
const Stream = require('stream')
exports.persistStream = (config, query, name, transform) => {
return new Promise((resolve, reject) => {
let recordCount = 0
var transformStream = new Stream.Transform({ writableObjectMode: true, readableObjectMode: true})
transformStream._transform = function (chunk, encoding, done) {
recordCount++
if (transform) chunk = transform(chunk)
let jsonChunk = JSON.stringify([chunk])
switch (true) {
case recordCount === 1:
jsonChunk = jsonChunk.slice(0, jsonChunk.length - 1); break
default:
jsonChunk = ',' + jsonChunk.slice(1, jsonChunk.length - 1); break
}
this.push(jsonChunk)
done();
};
transformStream._final = function (done) {
this.push(']')
done()
}
const gzip = zlib.createGzip()
let buffers = []
var stream = athena.execute(query)
.toStream()
.pipe(transformStream)
.pipe(gzip)
gzip.on('data', (chunk) => {
buffers.push(chunk)
})
gzip.on('end', function () {
let buffer = Buffer.concat(buffers)
redis.set(name, buffer.toString('base64'), (err, response) => {
zlib.gzip(config, (err, buff) => {
redis.set(name + ' Config', buff.toString('base64'), (err, response) => {
if (err) {
console.log(err)
reject()
} else {
console.log(name + ' succeeded')
resolve()
}
})
})
})
})
stream.on('error', (err) => {
console.log(err)
reject()
})
})
}