switch to using consume streams