add Consumer::receive_all

[svix-onelson]

Support for batched reads is not uniform among the various backends we have.
In some cases, partial batches will wait longer or shorter than the specified deadline.

The odd one out right now is RabbitMQ which supports configuring a channel with a pre-fetch limit, but you still have to just consume messages one-by-one (the batching is all under the hood). Additionally, there doesn't seem to be a way to specify a timeout or deadline (apparently this is left up to individual clients to decide what makes the most sense).

As such, RabbitMQ is under-served compared to the rest. The implementation for batch reads is entirely done in the client code (i.e. reading messages one-by-one, buffering them, then returning).

One possible solution is to add a max prefetch config option and use this when initializing a consumer channel, but this would also mean the max_messages argument would be meaningless for RabbitMQ, and using receive_all at all would probably be irrelevant. Callers should probably just use receive instead.

For now, I've left some comments in the rabbit impl and will leave this as future work.

---

Support for batched reads is not uniform among the various backends we have.

The in-memory and rabbitmq impls differ from the rest as these consumers are channel/stream based, with no explicit option for consuming messages in batches.

Using tokio::time::timeout for the first item, then looping to opportunistically try to fill up the batch with any remaining already ready messages, we can emulate a similar behavior as the other backends.

In all cases, receive_all should:

• not block longer (+/-) than the specified deadline, even if no items
  are available.
• not linger waiting for more items once it as received any.

Additionally, a new bit of config has been exposed for rabbitmq so callers can specify a prefetch count. This didn't seem to have any impact on the tests, but seems good to have.

[svix-gabriel]

now_or_never is pretty cool