basic stream processing functions