helper function for stream callbacks