今だと自前のオペレーターを作れば良いと思いますが、制約が結構あるので大変そう。 AsyncStream が出てきたら割と何でも出来るようになると思います。