Escribir flujo node.js en una variable de cadena

Question

Estoy pirateando un programa node.js que captura correos SMTP y actúa sobre los datos de correo. La biblioteca del nodo 'smtp-protocol' proporciona los datos del correo en vivo como una secuencia, y como novato node.js no estoy seguro de cómo escribir esa secuencia en una variable de cadena. En este momento tengo que escribir en la salida estándar utilizando la línea:

stream.pipe(process.stdout, { end : false }); 

Como dije, tengo que conseguir estos datos de corriente al grabar en una variable de cadena en su lugar, lo que voy a utilizar a continuación, una vez que el flujo ha terminado.

Ayuda muy apreciada!

Answer 1

La clave es utilizar estos dos Stream events:

• Evento: 'data'
• Evento: 'final'

Para stream.on('data', ...) que debe recolectar sus datos de datos ya sea en un búfer (si es binario) o en una cadena.

Para on('end', ...) debe llamar a una devolución de llamada con el búfer completo, o si puede alinearlo y utilizar devolver utilizando una biblioteca Promises.

Answer 2

Espero que esto sea más útil que la respuesta anterior (que ahora tiene un enlace roto).

También tenga en cuenta que la concatenación de cadenas no es una forma eficaz de recoger las partes de cuerda, pero se usa por simplicidad (y tal vez su código no se preocupa por la eficiencia)

var string = '' 
stream.on('readable',function(buffer){ 
    var part = buffer.read().toString(); 
    string += part; 
    console.log('stream data ' + part); 
}); 


stream.on('end',function(){ 
console.log('final output ' + string); 
}); 

Answer 3

De los nodejs documentation que debiera hacer esto - recuerda siempre una cadena sin conocer la codificación es sólo un montón de bytes:

var readable = getReadableStreamSomehow(); 
readable.setEncoding('utf8'); 
readable.on('data', function(chunk) { 
    assert.equal(typeof chunk, 'string'); 
    console.log('got %d characters of string data', chunk.length); 
}) 

Answer 4

estoy usando por lo general esta función sencilla para transformar una secuencia en una cadena:

function streamToString(stream, cb) { 
    const chunks = []; 
    stream.on('data', (chunk) => { 
    chunks.push(chunk.toString()); 
    }); 
    stream.on('end',() => { 
    cb(chunks.join('')); 
    }); 
} 

Ejemplo de uso:

let stream = fs.createReadStream('./myFile.foo'); 
streamToString(stream, (data) => { 
    console.log(data); // data is now my string variable 
}); 

Answer 5

Ninguna de las anteriores trabajado para mí. Necesitaba usar el objeto Buffer:

const chunks = []; 

    readStream.on("data", function (chunk) { 
    chunks.push(chunk); 
    }); 

    // Send the buffer or you can put it into a var 
    readStream.on("end", function() { 
    res.send(Buffer.concat(chunks)); 
    }); 

Answer 6

corrientes no tienen un simple .toString() función (que entiendo), ni algo así como una función .toStringAsync(cb) (que no entiendo).

así que creé mi propia función auxiliar:

var streamToString = function(stream, callback) { 
    var str = ''; 
    stream.on('data', function(chunk) { 
    str += chunk; 
    }); 
    stream.on('end', function() { 
    callback(str); 
    }); 
} 

// how to use: 
streamToString(myStream, function(myStr) { 
    console.log(myStr); 
}); 

Answer 7

Esto funcionó para mí y se basa en Node v6.7.0 docs:

let output = ''; 
stream.on('readable', function() { 
    let read = stream.read(); 
    if (read !== null) { 
     // New stream data is available 
     output += read.toString(); 
    } else { 
     // Stream is now finished when read is null. 
     // You can callback here e.g.: 
     callback(null, output); 
    } 
}); 

stream.on('error', function(err) { 
    callback(err, null); 
}) 

Answer 8

¿Qué pasa con algo así como un reductor de flujo?

Aquí hay un ejemplo de cómo utilizar las clases de ES6.

var stream = require('stream') 

class StreamReducer extends stream.Writable { 
    constructor(chunkReducer, initialvalue, cb) { 
    super(); 
    this.reducer = chunkReducer; 
    this.accumulator = initialvalue; 
    this.cb = cb; 
    } 
    _write(chunk, enc, next) { 
    this.accumulator = this.reducer(this.accumulator, chunk); 
    next(); 
    } 
    end() { 
    this.cb(null, this.accumulator) 
    } 
} 

// just a test stream 
class EmitterStream extends stream.Readable { 
    constructor(chunks) { 
    super(); 
    this.chunks = chunks; 
    } 
    _read() { 
    this.chunks.forEach(function (chunk) { 
     this.push(chunk); 
    }.bind(this)); 
    this.push(null); 
    } 
} 

// just transform the strings into buffer as we would get from fs stream or http request stream 
(new EmitterStream(
    ["hello ", "world !"] 
    .map(function(str) { 
    return Buffer.from(str, 'utf8'); 
    }) 
)).pipe(new StreamReducer(
    function (acc, v) { 
    acc.push(v); 
    return acc; 
    }, 
    [], 
    function(err, chunks) { 
    console.log(Buffer.concat(chunks).toString('utf8')); 
    }) 
);