fix(stream): flush enqueued entries on every pull on direct streams

fix(stream): flush enqueued entries on every pull on direct streams #14019