Support substituting receive() with a different AsyncSequence #127
Description
Is your feature request related to a problem? Please describe.
The current Transport API mandates that the receive() function is an AsyncThrowingStream. Unfortunately, this type doesn't support backpressure. Meaning that if the messages are handled slower than they arrive, this will start (slowly or quickly) blowing up memory. At large volumes, this can cause an unintentional Denial of Service.
Describe the solution you'd like
If we can have an associatedtype instead of the AsyncThrowingStream, Swift(NIO) based libraries can provide their own implementations that support backpressure. That would eliminate this problem.
associatedtype Messages: AsyncStream<Data, any Error>