highland
highland copied to clipboard
caolan
errors and async consequence
Hello,
the current API for errors is
errors(function(err, push) {})
so we can push one or more values at the current location of the stream if we already know those values.
When you need an async operation in order to know how to react to an error, how would you organize your streams ? a consume custom transform ? a sort of push(stream) + sequence
I wonder if there is room for an API of the style
errors(function(err, push, next) {})
well, it's a bit tricky, i think. First the errors-handler just have two options really: either it just logs and does nothing else with the values in the stream or it rethrows by push(err)-ing it again. the errors implementation have called next already then, so calling next sometime late should probably be ok.
but, the thing, why i consider it tricky is the question: what kind of error is it. why do you need to do an additional async operation to react to it? is it because the error is part of the business logic, like often done in java with checked execeptions? this case, i wouldn't use error objects, but build special value objects (pretty much like _.nil ).
otherwise usual reaction to such non-business-logic errors is immediate shutdown (by pushing nil), transform+rethrow (by pushing err), log+ignore (by not-pushing) or retry, wich would imply a special implementation of the errors-handler function to be able to trigger the whole chain.
can you, @jamesrom , show some cases, where an async version is needed?
I guess that those are border line case between errors that you can forget and business logic. When I created the issue, I was trying to make sure the the error was logged and the resource in error locked (this is done with a promise) before accepting the stream to continue.
I think the use case is fair; even if you're not treading the "errors vs business logic" line, you might just want to wait and make sure your log gets written before moving on.
It's only a two line change from how errors is implemented now, and making your whole own consume to do the same thing seems cumbersome. The question is how to support both at the same time; we can just check the arity of f:
if (f.length > 2) {
f(err, push, next);
} else {
f(err, push);
next();
}
But is it bad to be relying on the arity to determine this?
If we do go this route, maybe we should also consider a third option, f(err), where either we automatically pass the errors along, or we push nothing.
I think what @greelgorke meant was that a stream error typically involves shutting down the whole stream and emitting nil. Providing a next call is pointless in this case, because the stream is considered ended at this point. This is how I personally use Highland streams.
That said, the very fact that this behavior isn't baked into Highland (like it is in RxJs, for example) means that probably not everyone treats stream errors the way I do. It may be worth it to make those peoples' lives a little easier, since it would not be a big effort on our part.
Regarding @LewisJEllis's proposal, my concern with relying on length is that it doesn't work with curried functions in a non-obvious way (i.e., no errors are thrown).
We could introduce an onError transform that does the f(err, push, next) case, then implement errors in terms of onError.
yes I think I remember reading that caolan does not favor using arity in general.
I use streams as many operations that need to be done. some are in error, some pass. But an error should not end the stream. @vqvu do you mean you only use stopOnError ?
+1 for onError. it seems like a good name for a more powerful errors.
I might try to implement this in a few days when I'll go back to my use case. I'll ping here before starting just in case someone needs it before and wants to send a PR
ok I see. As I said, I use highland as a stamp machine (a bit like a production line). Each transform has an operation that must be done with the token, and when an operation fails on a token, the token is extracted from the production line. so I use errors as checkpoints.
I would like the production line to be paused until the "extraction" is guaranteed to be finished.
I see. I handle situations like that by using flatMap and returning a stream that only emits nil when the "extraction" is done.
Your approach makes sense too though.
yes flatMap probably makes it more composable (all operations between 2 checkpoints can be 'hidden' in a function called by flatMap).
This is really old now, but exactly what I'm dealing with currently. @vqvu, @jeromew et al, what is the approach you use now-a-days?
My case is that I need to log errors in an async fashion (specifically to an AWS S3 bucket) and push the error back on the stream so that it can go and fail. Additionally, if the logging has issues, I push a new error that does a couple extra steps. The async stuff is within a promise wrapped by highland.
Is there an example in the docs of a pattern that covers this? Or alternative approaches?
I would just use consume for this. The onError transform suggested in this issue isn't implemented.
stream.consume((err, x, push, next) => {
if (err) {
// You can use toCallback if logErrorAsync must return a Highland stream,
// and you can't access the underlying promise.
logErrorAsync(err)
.then(() => next())
.catch((loggingError) => {
push(loggingError);
next();
});
push(err);
} else {
push(null, x);
if (x !== _.nil) {
next();
}
}
});
You can extract this into a through-transform if you need to use it in multiple places. There are examples in the docs.