-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathasyncio_echo_server_protocol.py
More file actions
64 lines (51 loc) · 1.76 KB
/
asyncio_echo_server_protocol.py
File metadata and controls
64 lines (51 loc) · 1.76 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import asyncio
import logging
import sys
SERVER_ADDRESS = ("localhost", 10000)
logging.basicConfig(
level=logging.DEBUG, format="%(name)s: %(message)s", stream=sys.stderr
)
log = logging.getLogger("main")
event_loop = asyncio.get_event_loop()
class EchoServer(asyncio.Protocol):
def __init__(self):
self.transport = None
self.address = None
self.log = None
def connection_made(self, transport):
self.transport = transport
self.address = transport.get_extra_info("peername")
self.log = logging.getLogger("EchoServer_{}_{}".format(*self.address))
self.log.debug("connection accepted")
def data_received(self, data):
self.log.debug("received {!r}".format(data))
self.transport.write(data)
self.log.debug("sent {!r}".format(data))
def eof_received(self):
self.log.debug("received EOF")
if self.transport.can_write_eof():
self.transport.write_eof()
def connection_lost(self, error):
if error:
self.log.error("ERROR: {}".format(error))
else:
self.log.debug("closing")
super().connection_lost(error)
# Create the server and let the loop finish the coroutine before
# starting the real event loop
factory = event_loop.create_server(EchoServer, *SERVER_ADDRESS)
try:
server = event_loop.run_until_complete(factory)
except OSError:
print("Failed to create server")
sys.exit(1)
log.debug("starting up on {} port {}".format(*SERVER_ADDRESS))
# Enter the event loop permanently to handle all connections
try:
event_loop.run_forever()
finally:
log.debug("closing server")
server.close()
event_loop.run_until_complete(server.wait_closed())
log.debug("closing event loop")
event_loop.close()