Improve connection handling

[boimart1]

Issues that were encountered during usage:

• If an AmqpConnection was closed due to a read/write error with the client, the AmqpConnection object stayed registered in the AmqpServer, polluting the server with dead connection objects.
• If an AmqpServer had several AmqpConnections, during shutdown, it only properly closed half of the connections and left the other half with pending tasks, which caused very long error messages when shutting down the app.

Improvements:

To resolve the first issue, the connection automatically unregisters itself when it's closed (due to the reader being closed). There are a few conditions under which a connection should be closed:

1. if the reader reads EOF,
2. if the reader throws an exception,
3. if the AmqpServer is shutting down and notifies the AmqpConnection with close().

To handle all of these cases properly:

• When the AmqpServer starts a connection, it is now done with an async callback rather than a sync callback. The async callback starts the connection and awaits connection.run_until_complete().
• When the AmqpServer shuts down, a call to connection.close() is done for each connection, which will simply cancel the reader.
• Added AmqpConnection.run_until_closed(), which awaits the reader. If the reader task was finished due to the loop ending (reader got an EOF) or an exception, the error is logged here, and a call to its own connection.close() is done. If the reader was finished due to an external call to connection.close(), then run_until_done() simply returns.

The second issue was due to the for connection in self._connections: loop during shutdown, where the connection.close() would end up removing itself from the self._connections list, basically invalidating the iterator and causing some connections to be skipped. To resolve this, tasks are created all at once and then awaited simultaneously with asyncio.gather().