pipe - replace use of DELAY() with a better solution, add features.