Replace AsyncRead impl of Object with Stream of Bytes

[paolobarbolini]

This replaces the AsyncRead implementation of Object with a Stream<Item = io::Result<Bytes>>.

The reason for doing this breaking change is that because the AsyncRead impl is a wrapper of a Consumer (which in turn uses a Subscriber internally) the data already came in the form of a Bytes Stream, so it was the wrong abstraction to force the user to go through the naive AsyncRead implementation, at times probably needing to reconvert that back to a Stream because of many async crates preferring them to read/write style APIs.

This removes 2 memcpys from the stack (copying into the VecDeque and copying out of the VecDeque).

Users who need the AsyncRead implementation can use tokio_util::io::StreamReader, which has an even more efficient implementation than the nats.rs one by doing only one memcpy.

As a side note, this also replaces what I thought was a messy implementation (and in some places unexpected, like throwing an error if a second read had been launched after exhausting the stream, or the use of unwrap in creating the Consumer) with a more structured and idiomatic one.

[Jarema]

I like how the new way simplifies the codebase, and cleans it up. I just wonder if we should introduce such a hefty breaking change that will affect anyone using Object Store.

EDIT: Did you have an opportunity to bench it against the current solution?

[paolobarbolini]

I haven't. I was mainly triggered by the AsyncReader 😃. I'll look into it.

[thomastaylor312]

Just chiming in here to say that I'd love to see this land. Having it be a stream feels much easier (and idiomatic) to work with

[Jarema]

Happy to revisit this one, but requires performance check.