Sink impl for AsyncSender?

[silvanshade]

Would it be possible to add a SendSink for AsyncSender similar to ReceiveStream for AsyncReceiver? This would be nice to have and flume also supports this.

You could add a "sink" feature with an optional dependency on futures-sink to enable this.

I took a brief look at adding it myself but I'm not familiar enough with the internals to know the best way to implement SendSink.

In particular, I'm not entirely sure how to implement Sink::poll_flush and Sink::poll_close. I suspect that I need to use Signal to do this efficiently and somehow calculate when queue is empty but I haven't looked into it further.