Prevent incomplete task leak in recv.

If `self.worker` completes before `next_message`, then `next_message` will remain scheduled forever. This is a pr to explicitly cancel the coroutine if the worker completes. A new `next_message` will naturally be scheduled if/when recv is called again. 

It also uses `asyncio.async` instead of `Task` in accordance with the [docs](https://docs.python.org/3/library/asyncio-task.html#asyncio.Task).