а если это стрим от контроллера (а он от контроллера) - то стрим не завершиться пока не закроется исходный контроллер и подписка не отдаст все что там выпущено
Лучший вариант для тебя - сделать стримтрансформер, который будет принимать функцию редьюсер и функцию пареобразования) Ну или что проще - использовать asyncExpand или whereType.map