Add support for coroutine functions as listener callbacks

[elprans]

The Connection.add_listener(), Connection.add_log_listener() and
Connection.add_termination_listener() now allow coroutine functions as
callbacks.

Fixes: #567.

[theo-brown]

Hi, I've also been working on this! Much less than you though, you've done a lot!

I've got a couple of questions, not necessarily criticisms just looking for clarification (I'm fairly new to async).

1. Why did you use both create_task and call_soon, rather than just create_task? I thought they both run at the top of the next event loop.
2. What is the perceived need for the _Callback class, beyond checking whether something is async?

[elprans]

1. Why did you use both create_task and call_soon, rather than just create_task?

Because create_task works on coroutines and call_soon on regular callables.

2. What is the perceived need for the _Callback class, beyond checking whether something is async?

To avoid calling inspect.iscoroutinefunction on every callback invocation, since this can be easily determined when the callback is set up.

[theo-brown]

Because create_task works on coroutines and call_soon on regular callables.

My mistake, I incorrectly thought that create_task could be used on both.

[theo-brown]

Does this suggestion offer an improvement to your exception handling, or is there no impact?

  for cb in self._listeners[channel]:
      self._loop.create_task(self._call_listener(cb, ...)

async def _call_listener(cb, ...)
  try:
      if cb.is_async:
          await cb.cb(con_ref, pid, channel, payload)
      else:
          cb.cb(con_ref, pid, channel, payload)
    except:
        ....

[elprans]

Does this suggestion offer an improvement to your exception handling, or is there no impact?

The event loop already logs exceptions in tasks and callbacks fairly well, and, if anything, I think custom invocation of call_exception_handler there might be a vestige of old days of bad asyncio or something. That said, handling it this way for both types of callbacks is consistent at least.

[theo-brown]

Oh nice, that's so much tidier.
Would it be worth swapping the test queue from q3.put_nowait() to await q3.put()?