I have a writable stream piped into a readable stream. The writable stream
at some point fails and emits an error. Me, as a readable stream
implementor, how am I supposed to handle this situation? Do I receive any
event from the writable stream?
I have this example that shows what I'm trying to explain.
--------------------------------------
var fs = require ("fs");
var rs = fs.createReadStream ("in", { highWaterMark: 1 })
var ws = fs.createWriteStream ("out");
var i = 5;
var write = ws._write;
ws._write = function (){
if (!--i) this.emit ("error", new Error ());
else write.apply (this, arguments)
}
rs.on ("error", function (){
console.log("rs error")
})
rs.on ("end", function (){
console.log("rs end")
})
rs.on ("close", function (){
console.log("rs close")
})
ws.on ("error", function (error){
console.log(error)
})
ws.on ("finish", function (error){
console.log("ws finish")
})
ws.on ("open", function (error){
console.log("ws open")
})
ws.on ("pipe", function (error){
console.log("ws pipe")
})
ws.on ("unpipe", function (error){
console.log("ws unpipe")
})
rs.pipe (ws);
--------------------------------------
File "in" has the content "1234567890".
File "out" ends with the content "1234".
The console output is:
ws pipe
ws open
ws unpipe
[Error]
So, what I don't understand is what is occurring under the hood. I've been
following the source code and what happens is:
- When rs calls to pipe(), an "error" handler (Line
545<https://github.com/joyent/node/blob/master/lib/_stream_readable.js#L545>)
and an "unpipe" handler (Line
489<https://github.com/joyent/node/blob/master/lib/_stream_readable.js#L489>)
are attached to the ws.
- When ws emits an error, the previous error handler is executed and it
calls to unpipe() (Line
576<https://github.com/joyent/node/blob/master/lib/_stream_readable.js#L576>
).
- Then the ws is unpiped from the rs, and emits an "unpipe" event (Line
628<https://github.com/joyent/node/blob/master/lib/_stream_readable.js#L628>).
This event calls the previous "unpipe" handler.
- When the "unpipe" handler is executed it calls to cleanup() (Line
509<https://github.com/joyent/node/blob/master/lib/_stream_readable.js#L509>).
It basically removes some attached listeners.
So... as you can see the readable stream never emits any event, just like
the result of the previous example (none of the ws listeners are executed).
I don't understand this, the readable stream magically stops emitting data
just because the destination stream is unpiped, but I don't see in the code
where the Readable stream notifies its implementor
(ReadStream<https://github.com/joyent/node/blob/master/lib/fs.js#L1415>)
to close the underlying resource and stop pushing more data into it.
My readable stream implementation gets the data from internet. If the
writable stream fails I need to close the socket. Right now I simply have
an abort() function that is called when the writable stream fails:
var rs = getMyReadStream ();
var ws = fs.createWriteStream ("out");
ws.on ("error", function (){
rs.abort ();
})
rs.pipe (ws);
I want to remove the call to abort(), it should be called implicitly from
inside the readable stream. Maybe should I listen to an "unpipe" event?
(the ReadStream
<https://github.com/joyent/node/blob/master/lib/fs.js#L1415>doesn't
do this)
--
--
Job Board: http://jobs.nodejs.org/
Posting guidelines:
https://github.com/joyent/node/wiki/Mailing-List-Posting-Guidelines
You received this message because you are subscribed to the Google
Groups "nodejs" group.
To post to this group, send email to [email protected]
To unsubscribe from this group, send email to
[email protected]
For more options, visit this group at
http://groups.google.com/group/nodejs?hl=en?hl=en
---
You received this message because you are subscribed to the Google Groups
"nodejs" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
For more options, visit https://groups.google.com/groups/opt_out.
