Drain or cancel waiting tasks when the producer is destroyed

[rnowling-memphis]

The current implementation of async produce doesn't track the active tasks. Using a pattern from the Python asyncio documentation, tasks could be tracked so that when destroy() is called on the producer, all outstanding messages can be cancelled or at least finished.