Refactor to make sources/sinks operate on Pijps.