=== Message

Refactor this code so that this consumed stream pipeline is not reused.