Dynamically add/delete sources

duplicate with When stream-eos holding in message structure