refactor, introduce generalized give.flow abstraction for a pipe implementation