Through stream that calls a cb when no data has been written after the first x seconds.
var Timeout = require('through-timeout');
source
.pipe(Timeout({
objectMode: true,
duration: 10000
}, function () {
// oh no!
source.destroy();
}))
.pipe(destination);MIT