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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
"""
Example of a simple TCP server that is written in (mostly) coroutine
style and uses asyncio.streams.start_server() and
asyncio.streams.open_connection().
Note that running this example starts both the TCP server and client
in the same process. It listens on port 1234 on 127.0.0.1, so it will
fail if this port is currently in use.
"""
from __future__ import print_function
import sys
import trollius as asyncio
import asyncio.streams
from trollius import From, Return
class MyServer:
"""
This is just an example of how a TCP server might be potentially
structured. This class has basically 3 methods: start the server,
handle a client, and stop the server.
Note that you don't have to follow this structure, it is really
just an example or possible starting point.
"""
def __init__(self):
self.server = None # encapsulates the server sockets
# this keeps track of all the clients that connected to our
# server. It can be useful in some cases, for instance to
# kill client connections or to broadcast some data to all
# clients...
self.clients = {} # task -> (reader, writer)
def _accept_client(self, client_reader, client_writer):
"""
This method accepts a new client connection and creates a Task
to handle this client. self.clients is updated to keep track
of the new client.
"""
# start a new Task to handle this specific client connection
task = asyncio.Task(self._handle_client(client_reader, client_writer))
self.clients[task] = (client_reader, client_writer)
def client_done(task):
print("client task done:", task, file=sys.stderr)
del self.clients[task]
task.add_done_callback(client_done)
@asyncio.coroutine
def _handle_client(self, client_reader, client_writer):
"""
This method actually does the work to handle the requests for
a specific client. The protocol is line oriented, so there is
a main loop that reads a line with a request and then sends
out one or more lines back to the client with the result.
"""
while True:
data = (yield From(client_reader.readline()))
data = data.decode("utf-8")
if not data: # an empty string means the client disconnected
break
parts = data.rstrip().split(' ')
cmd = parts[0]
args = parts[1:]
if cmd == 'add':
arg1 = float(args[0])
arg2 = float(args[1])
retval = arg1 + arg2
client_writer.write("{0!r}\n".format(retval).encode("utf-8"))
elif cmd == 'repeat':
times = int(args[0])
msg = args[1]
client_writer.write("begin\n".encode("utf-8"))
for idx in range(times):
client_writer.write("{0}. {1}\n".format(idx+1, msg)
.encode("utf-8"))
client_writer.write("end\n".encode("utf-8"))
else:
print("Bad command {0!r}".format(data), file=sys.stderr)
# This enables us to have flow control in our connection.
yield From(client_writer.drain())
def start(self, loop):
"""
Starts the TCP server, so that it listens on port 1234.
For each client that connects, the accept_client method gets
called. This method runs the loop until the server sockets
are ready to accept connections.
"""
self.server = loop.run_until_complete(
asyncio.streams.start_server(self._accept_client,
'127.0.0.1', 12345,
loop=loop))
def stop(self, loop):
"""
Stops the TCP server, i.e. closes the listening socket(s).
This method runs the loop until the server sockets are closed.
"""
if self.server is not None:
self.server.close()
loop.run_until_complete(self.server.wait_closed())
self.server = None
def main():
loop = asyncio.get_event_loop()
# creates a server and starts listening to TCP connections
server = MyServer()
server.start(loop)
@asyncio.coroutine
def client():
reader, writer = yield From(asyncio.streams.open_connection(
'127.0.0.1', 12345, loop=loop))
def send(msg):
print("> " + msg)
writer.write((msg + '\n').encode("utf-8"))
def recv():
msgback = (yield From(reader.readline()))
msgback = msgback.decode("utf-8").rstrip()
print("< " + msgback)
raise Return(msgback)
# send a line
send("add 1 2")
msg = yield From(recv())
send("repeat 5 hello")
msg = yield From(recv())
assert msg == 'begin'
while True:
msg = yield From(recv())
if msg == 'end':
break
writer.close()
yield From(asyncio.sleep(0.5))
# creates a client and connects to our server
try:
loop.run_until_complete(client())
server.stop(loop)
finally:
loop.close()
if __name__ == '__main__':
main()
|