How can I redirect a stream based in an "internal state"

Question

How can I redirect a stream based in an "internal state"

I'm writing a plugin for gulp that uses a web service and depending on the response do one or other thing. The algorithm is something like this:

stream1 = through.obj(function(src, enc, cb) {
 if src.is_a_buffer()
     http_request(options)
     http_request.on('response', function () {
        if (statusCode = 200) {
            /* Normal course, everything works fine here */
            do_something()
            return cb()
        } else {
            /* Exception course, although the stream2 is created, is never executed */
            stream1.pipe(stream2())
        }

}, function (cb) {
    cb()
});

stream2 = through.obj(function(src,enc,cb) {
     do_other_stuff()
     stream2.push(src)
     return cb()
}, function (cb) {
    cb()
});

When I run the code stream2 never excecutes. Since i'm new to node streams, I think i misunderstood something. Can any of you guys help me out understanding what am I getting wrong here?

javascript
node.js
stream
gulp

Answer 1

When you call stream1.pipe(stream2()), stream1 has already emitted data (possibly all of it); making that call won't pass the execution on to stream2. There are a couple of ways this can be handled depending on your needs:

NOTE: I'm just modifying the original pseudocode here

Option 1:

Don't bother with a stream2 and just call do_other_stuff() directly:

stream1 = through.obj(function(src, enc, cb) {
 if src.is_a_buffer()
     http_request(options)
     http_request.on('response', function () {
        if (statusCode = 200) {
            /* Normal course, everything works fine here */
            do_something()
            cb()
        } else {
            do_other_stuff()
            cb()
        }

}, function (cb) {
    cb()
});

Option 2:

If you need stream2 for other purposes, pull the through.obj() callback out in to its own callable function and call it directly from your else clause.

stream1 = through.obj(function(src, enc, cb) {
 if src.is_a_buffer()
     http_request(options)
     http_request.on('response', function () {
        if (statusCode = 200) {
            /* Normal course, everything works fine here */
            do_something()
            return cb()
        } else {
            processStream2(src, enc, cb)
        }

}, function (cb) {
    cb()
});

function processStream2(src, enc, cb) {
     do_other_stuff()
     return cb()
}

stream2 = through.obj(processStream2, function (cb) {
    cb()
});

I hope that helps :)