Utilizo secuencias legibles y de transformación que luego uso for await .
No puedo encontrar una manera de procesar los errores de transmisión de la persona que llama para que puedan ser capturados en la función de la persona que llama.
Por ejemplo, si transform lanza, da como resultado un error no detectado.
Si agrego un detector de errores para transform , naturalmente no propaga el error a la captura principal.
function handler(err) { // Log the error here // If throw from here it won't be cauch in main // How do I propagate it to main? } function getStream() { // transform1 and transform2 are custom transform streams readable.pipe(csvParser).pipe(transform1).pipe(transform2); readable.on('error', handler); csvParser.on('error', handler); transform1.on('error', handler); transform2.on('error', handler); return transform2; } async function main() { try { const stream = getStream(); for await(const chunk of stream) { // process chunk } } catch (ex) { // how to catch transform errors here? } }¿Hay alguna forma de hacerlo?
Terminé resolviendo esto con Promise.race :
function readStream(stream) { return Promise.race([ new Promise((_, reject) => stream.once('error', reject)), iterate(stream) ]); } async function iterate(stream) { for await (const object of stream) { // ... } } Esto se rechazará con cualquier error de la secuencia o de la función de iterate , o se resolverá con el resultado de iterate , lo que ocurra primero.
Según tengo entendido, async/await es en gran medida azúcar sintáctico además de promesas. Específicamente, creo que dado que ReadableStream.pipe sigue un patrón basado en eventos ( on("error", errorHandler) ) en lugar de un patrón basado en promesas, el try { ... await ... } catch (ex) { ... } la construcción no va a manejar los errores asincrónicos "lanzados" dentro de ReadableStream.pipe tan bien como cabría esperar.
Suponiendo que mi comprensión es correcta (y para ser honesto, no uso la sintaxis aysnc/await con frecuencia, por lo que es posible que también me esté perdiendo algo), una solución sencilla es volver a agregar un controlador de eventos basado en devolución de llamada como este:
function handleError(err) { ... } readable.once("error", handleError); // and if you want, re-use that handler within your catch block, like: // `} catch (ex) { handleError(ex); }` // but I'm not sure how often that will come up if you follow this patternpero es posible que ya lo sepas.
De lo contrario, si realmente desea utilizar la construcción aysnc/await/try/catch-style en este caso, podría utilizar algo como util.promisify para convertir la API basada en eventos on("error", handler) que ReadableStream es usando promesas. Esta publicación de StackOverflow: ¿Cómo usar Async await usando util promisify? - cubre ese tema con más profundidad, pero (en mi opinión) parece un montón de aros para saltar solo para evitar agregar readable.once("error" /* ...whatever you'd otherwise have in your catch block ... */)
En resumen, creo que debido a que ReadableStream.pipe no está diseñado en torno a promesas, la sintaxis async/await no es suficiente (en sí misma) para garantizar que los errores asíncronos que podrían emitirse como eventos de error sean atrapados por el bloque de prueba/captura. Debe manejar esos errores de la manera tradicional, ya sea directamente (registrando un controlador para los eventos de error emitidos por su readable , en cuyo caso las cosas de esperar y probar/atrapar no son directamente aplicables) o "indirectamente" ( mediante la creación de un adaptador que hace que esos eventos emitidos burbujeen como el caso de captura en una promesa resuelta, en cuyo caso puede hacer que parezca un intento/captura de estilo síncrono usando async/await).
Creo que una función de transmisión de transformación no debería generar una excepción. En su lugar, debería emitir un evento de error o pasar el error a la función de devolución de llamada.
Aquí hay un ejemplo.
Envuelve todo en el método transform con un bloque try catch. Y se propaga a la función principal.
const { Transform } = require("stream"); //Added on 1st edit function handler(ex) { console.error("Logged By Handler: ", ex); } function getStream() { // A simple transform stream to test const transform = new Transform({ transform(chunk, encoding, callback) { try { chunk = chunk.toString(); // if chunk == "err" then we want to throw an error // to simulate a real life error if (chunk === "err\n") return callback("Oops! Sth failed in the transform stream.", null); // or this.emit("error", "Oops! Sth failed in the transform stream."); this.push(chunk.toUpperCase()); // Simulating exception throw new Error(`Fatal error.`); callback(); } catch (ex) { handler(ex); callback(ex, null); } }, }); process.stdin.pipe(transform); // readable -> transform return transform; } async function main(transform) { try { for await (const chunk of transform) process.stdout.write(chunk); } catch (ex) { console.error("Handled by main:", ex); } } main(getStream());