IMO your bidi streaming layer should should be entirely transport- and data- agnostic. You can build those in a layer up if you actually need them. See rSocket[0] or my own omnistreams[1].

[0]: https://rsocket.io/

[1]: https://github.com/omnistreams