Considering accepting `Stream` and `Sink` as in parameters instead

[BusyJay]

Current API design returns Stream and Sink for streaming APIs, so that users can choose whichever executors they want. But it has limitations:

1. Usually only spawning the final futures into the client/context can achieve less context switch and maximum throughput.
2. If spawning the futures back to client/context, there are always code path executed multiple times unnecessary. For example, we always know Sink::start_send provided by grpcio can only accept one message at a time. All other messages can be accepted only after next CompletionQueue::next is executed at lease once. But a properly implemented future adapter usually will execute additional poll_complete or start_send to get a NotReady result, which will lock and poll unnecessary.

If we accept Stream and Sink as in parameter instead, we can control the timing more precisely and provide best practice out of box. Taking RouteGuide as an example, the current API is:

fn route_chat(&self) -> (ClientDuplexSender<RouteNote>, ClientDuplexReceiver<RouteNote>)

New API becomes:

fn route_chat(&self, nodes: impl Stream<Item=RouteNote>, sink: impl Sink<Item=RouteNote>)

However the new APIs may not be friendly for users as it requires a constructed Stream and Sink.