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
|
"""Test service that accepts connections and reads all data off them."""
from __future__ import print_function
import argparse
import os
import sys
from trollius import *
ARGS = argparse.ArgumentParser(description="TCP data sink example.")
ARGS.add_argument(
'--tls', action='store_true', dest='tls',
default=False, help='Use TLS with a self-signed cert')
ARGS.add_argument(
'--iocp', action='store_true', dest='iocp',
default=False, help='Use IOCP event loop (Windows only)')
ARGS.add_argument(
'--host', action='store', dest='host',
default='127.0.0.1', help='Host name')
ARGS.add_argument(
'--port', action='store', dest='port',
default=1111, type=int, help='Port number')
ARGS.add_argument(
'--maxsize', action='store', dest='maxsize',
default=16*1024*1024, type=int, help='Max total data size')
server = None
args = None
def dprint(*args):
print('sink:', *args, file=sys.stderr)
class Service(Protocol):
def connection_made(self, tr):
dprint('connection from', tr.get_extra_info('peername'))
dprint('my socket is', tr.get_extra_info('sockname'))
self.tr = tr
self.total = 0
def data_received(self, data):
if data == b'stop':
dprint('stopping server')
server.close()
self.tr.close()
return
self.total += len(data)
dprint('received', len(data), 'bytes; total', self.total)
if self.total > args.maxsize:
dprint('closing due to too much data')
self.tr.close()
def connection_lost(self, how):
dprint('closed', repr(how))
@coroutine
def start(loop, host, port):
global server
sslctx = None
if args.tls:
import ssl
# TODO: take cert/key from args as well.
here = os.path.join(os.path.dirname(__file__), '..', 'tests')
sslctx = SSLContext(ssl.PROTOCOL_SSLv23)
if not BACKPORT_SSL_CONTEXT:
sslctx.options |= ssl.OP_NO_SSLv2
sslctx.load_cert_chain(
certfile=os.path.join(here, 'ssl_cert.pem'),
keyfile=os.path.join(here, 'ssl_key.pem'))
server = yield From(loop.create_server(Service, host, port, ssl=sslctx))
dprint('serving TLS' if sslctx else 'serving',
[s.getsockname() for s in server.sockets])
yield From(server.wait_closed())
def main():
global args
args = ARGS.parse_args()
if args.iocp:
from trollius.windows_events import ProactorEventLoop
loop = ProactorEventLoop()
set_event_loop(loop)
else:
loop = get_event_loop()
try:
loop.run_until_complete(start(loop, args.host, args.port))
finally:
loop.close()
if __name__ == '__main__':
main()
|
