added Tap operator for listening to stream sources