stream proccessing refactoring