Фильтрация объекта Stream в node.js

Мне кажется, что элегантный способ обработки определенных типов данных в Node.js - это цепочка объектов обработки, таких как конвейеры UNIX.

Например, grep:

function Grep(pattern) {
    ...
}
util.inherits(Grep, stream.Stream);

Grep.prototype.???? = ???????  // What goes here?

grep = new Grep(/foo/);

process.stdin.pipe(grep);
myStream.pipe(process.stdout);

Однако это Мне совсем не ясно, как различные методы Stream должны быть переопределены, чтобы это работало.

Как я могу создать объект Stream, который просто копирует его входные данные в свои выходные? Предположительно, с этим ответом более сложные потоки фильтрации станут тривиальными.

Обновление: кажется, что следующее должно работать (выражено в CoffeeScript, поэтому я не заполняю это поле синтаксисом JS!):

class Forwarder extends stream.Stream
    write: (chunk, encoding) ->
        @emit 'data', chunk
    end: (chunk, encoding) =>
        if chunk?
            @emit 'data', chunk
        @emit 'end'

fwd = new Forwarder()
fwd.pipe(process.stdout);
process.stdin.pipe(fwd);
process.stdin.resume();

Однако добавление чего-либо к этому сценарию ничего не выводит. Явный вызов fwd.write () в скрипте вызывает вывод на стандартный вывод.