refactored source/sink/pipe
added FiltedSource and FilteredSink