refactoring and moved message stream processing