So the situation is that I have a variety of datasources that are providing a stream of messages from e