use flow instead of stream