is this how you combine streams and do error handling in node?

hyperquest-trumpet-streams.js

var hyperquest = require("hyperquest");
var trumpet = require("trumpet");
var through = require("through");
var duplexer = require("duplexer");
var barrage = require("barrage");

function download(url) {
    var outstream = through();

    var tr = trumpet();

    tr.select("title").createStream().pipe(through(function (data) {
        outstream.queue({"dom": data.toString()});
    }))

    var request = hyperquest.get(url);

    dup = duplexer(tr, outstream);

    request.on("error", function (error) {
        dup.emit("error", error);
    });

    request.on("response", function (response) {
        //console.log("in request.on response, but it's not propagating?");
        dup.emit("response", response);
    });

    request.pipe(tr);

    return dup;

    //return barrage(hyperquest.get(url)).syphon(barrage(tr)).syphon(barrage(outstream))

    //return barrage(hyperquest.get(url)).syphon(barrage(duplexer(tr, outstream)));

    //return request.pipe(tr);

    //return request.pipe(duplexer(tr, outstream))

    //return request.pipe(tr).pipe(duplexer(tr, outstream));

    //return duplexer(duplexer(tr, request), outstream);
};

download("http://httpbinfjiafaklfda.org")
    .on("error", function (error) {
        console.log("1error: " + JSON.stringify(error));
    })
    .on("data", function (data) {
        console.log("1data: " + JSON.stringify(data));
    });

download("http://httpbin.org/404")
    .on("error", function (error) {
        console.log("2error: " + JSON.stringify(error));
    })
    .on("response", function (response) {
        // TODO: why is this never triggered?
        console.log("2response");
    })
    .pipe(through(function (data) { this.queue(data.toString()); }))
    .pipe(process.stdout)

    /*
    .on("data", function (data) {
        console.log("2data: " + JSON.stringify(data));
    });
    */

jail.js

var through = require("through");
var hyperquest = require("hyperquest");
var barrage = require("barrage");

function download(url) {
    var stream = through();

    stream.on("data", function (data) {
        // TODO: do streaming html parsing here ?
        console.log("received data: " + data);
        this.queue("some parsed result");
    });

    return barrage(hyperquest.get(url)).syphon(barrage(stream));
};

download("http://djifadlkfkjdaflkajflkada.com").on("error", function(error) {
    console.log("doesn't ever happen");
});

last.js

var through = require("through");
var hyperquest = require("hyperquest");
var through = require("through");
var duplexer = require("duplexer");
var trumpet = require("trumpet");

function download(url) {
    var stream = through();

    stream.on("response", function (response) {
        console.log("stream just caught response");
    });

    var tr = trumpet();

    tr.select("title").createStream().pipe(through(function (data) {
          stream.queue({"dom": data.toString()});
          // stream.queue("hello world");
    }));

    var request = hyperquest.get(url);

    request.on("error", function (error) {
        stream.emit("error", error);
    });

    request.on("response", function (response) {
        console.log("request.on response called..");
        //stream.emit("response", response);
        stream.emit("response", response);
    });

    var dup = duplexer(tr, stream);

    return duplexer(request.pipe(dup), 

    return dup;
};

var stream1 = download("http://dafkjldjfakda.com");

stream1.on("error", function (error) {
    console.log("yay, the error handler was called");
});

var stream2 = download("http://heybryan.org/");

stream2.on("response", function (response) {
    console.log("stream2 response called !!!");
});

stream2.pipe(through(function (data) {
    console.log("stream2 JSON.stringify data -> " + JSON.stringify(data));
}));

ugh.js

var hyperquest = require("hyperquest");
var through = require("through");
var trumpet = require("trumpet");
var duplexer = require("duplexer");

// curl http://heybryan.org/ | htmlparser

function download(url) {
    var tr = trumpet();
    return hyperquest.get(url).pause().pipe(tr)
};


download("http://jlkfdkadjfkla.com/")
    .on("error", function(error) {
        console.log("download.on error called");
    })
    .on("response", function(response) {
        console.log("download.on response called");
    })
    .pipe(process.stdout)
    .resume()

download("http://heybryan.org/")
    .on("error", function(error) {
        console.log("download.on error called");
    })
    .on("response", function(response) {
        console.log("download.on response called");
    })
    .pipe(process.stdout)
    .resume()

wtf.js

var trumpet = require("trumpet");
var hyperquest = require("hyperquest");
var through = require("through");
var duplexer = require("duplexer");
var concat = require("concat-stream");

function download(url) {
    var tr = trumpet();

    var dup = duplexer(tr, tr.select('title').createReadStream().pipe(concat(function (title) {
        this.emit("data", "title is: " + title.toString());
    })));

    var request = hyperquest.get(url)
        .on("error", function(error) {
            console.log("got error (1)");
            dup.emit("error", error);
        })
        .on("response", function(response) {
            console.log("got response (1)");
            dup.emit("response", response);
        })
        .pipe(tr)

    return dup;
};

download("http://heybryanfdjlkdajfkdjflakdfa.org")
.on("error", function(error) { console.log("got error (2)") })
.on("response", function(response) { console.log("got response (2)"); })
.pipe(through(function (data) { console.log(data); }))

download("http://heybryan.org/")
.on("error", function(error) { console.log("got error (2)") })
.on("response", function(response) { console.log("got response (2)"); })
.pipe(through(function (data) { console.log(data); }))