Skip to content

Instantly share code, notes, and snippets.

@hexrain
Created Jul 23, 2019
Embed
What would you like to do?
Server example
import asyncio
import ssl
async def handle_echo(reader, writer):
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')
print("Received %r from %r" % (message, addr))
print("Send: %r" % message)
writer.write(data)
await writer.drain()
loop = asyncio.get_event_loop()
sc = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
sc.load_cert_chain('localhost.crt', 'localhost.key')
coro = asyncio.start_server(handle_echo, '127.0.0.1', 5000, loop=loop, ssl=sc)
server = loop.run_until_complete(coro)
try:
loop.run_forever()
except KeyboardInterrupt:
pass
# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment