import asyncio import io import struct import functools import ssl as ssl_module import socket import collections import time import traceback class MyEchoClientProtocol(asyncio.Protocol): def connection_made(self, transport): print('connection_made', transport) query = 'hello world' transport.write(query.encode('latin-1')) def data_received(self, data): print('data_received', data) def connection_lost(self, exc): print('connection_lost', exc) async def create_connection(ssl_obj, timeout=60): loop = asyncio.get_event_loop() connector = loop.create_connection(MyEchoClientProtocol, '127.0.0.1', 5000, ssl=ssl_obj) connector = asyncio.ensure_future(connector) tr, pr = await asyncio.wait_for(connector, timeout=timeout, loop=loop) return tr, pr async def main(timeout, ssl_obj): async def _acquire_impl(): try: proxy = await create_connection(ssl_obj) except Exception: raise else: return proxy res = await asyncio.wait_for(_acquire_impl(), timeout=timeout, loop=loop) return res async def test_cancel(): sc = ssl_module.create_default_context(ssl_module.Purpose.SERVER_AUTH, cafile='localhostc.crt') sc.check_hostname = False sc.verify_mode = ssl_module.CERT_NONE for i in range(10): # try 50 times timeout = 0.003 try: tr, pr = await main( timeout=timeout, ssl_obj=sc ) tr.close() except asyncio.TimeoutError as e: print('timeouterror', repr(e)) await asyncio.sleep(600) import asyncio loop = asyncio.get_event_loop() loop.run_until_complete(test_cancel()) loop.run_forever() loop.close()