move the source around for clarity
introduce control() operation and subsume fork() and fanout()
possible api for sending a reified a stream