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
|
#!/usr/bin/env python3
import asyncio
import dummy_killer
import tornado.ioloop
import tornado.web
import tornado.httpserver
import ssl
import argparse
import os
class MainHandler(tornado.web.RequestHandler):
@tornado.gen.coroutine
def get(self, path):
if path == '/empty':
# Return an empty reply
self.set_header("Content-Type", "text/plain")
self.write("")
elif path == '/error_403':
# Return a 403 HTTP error
raise tornado.web.HTTPError(403)
elif path == '/timeout':
# Wait for 4 seconds before returning an empty reply
yield tornado.gen.sleep(4)
self.set_header("Content-Type", "text/plain")
self.write("")
elif path == '/request':
# Return a string 'hello world'
self.set_header("Content-Type", "text/plain")
self.write("hello world")
elif path == '/map-simple':
# Return a string 'hello map'
self.set_header("Content-Type", "text/plain")
self.write("hello map")
elif path == '/map-query':
# Parse the 'key' argument from the HTTP request
key = self.get_query_argument("key", default=None)
if key == 'au':
# Return a string 'hit' if 'key' is equal to 'au'
self.set_header("Content-Type", "text/plain")
self.write("1.0")
else:
# Return a 404 HTTP error if 'key' is not equal to 'au'
raise tornado.web.HTTPError(404)
elif path == '/settings':
self.set_header("Content-Type", "application/json")
self.write("{\"actions\": { \"reject\": 1.0}, \"symbols\": { \"EXTERNAL_SETTINGS\": 1.0 }}")
else:
raise tornado.web.HTTPError(404)
@tornado.gen.coroutine
def post(self, path):
if path == '/empty':
# Return an empty reply
self.set_header("Content-Type", "text/plain")
self.write("")
elif path == '/error_403':
# Return a 403 HTTP error
raise tornado.web.HTTPError(403)
elif path == '/request':
# Return a string 'hello post'
self.set_header("Content-Type", "text/plain")
self.write("hello post")
elif path == '/timeout':
# Wait for 4 seconds before returning an empty reply
yield tornado.gen.sleep(4)
self.set_header("Content-Type", "text/plain")
self.write("")
elif path == '/map-simple':
# Return a string 'hello map'
self.set_header("Content-Type", "text/plain")
self.write("hello map")
elif path == '/map-query':
# Parse the 'key' argument from the HTTP request
key = self.get_query_argument("key", default="")
if key == 'au':
# Return a string 'hit' if 'key' is equal to 'au'
self.set_header("Content-Type", "text/plain")
self.write("hit")
else:
# Return a 404 HTTP error if 'key' is not equal to 'au'
raise tornado.web.HTTPError(404)
elif path == '/settings':
self.set_header("Content-Type", "application/json")
self.write("{\"actions\": { \"reject\": 1.0}, \"symbols\": { \"EXTERNAL_SETTINGS\": 1.0 }}")
else:
raise tornado.web.HTTPError(404)
def head(self, path):
self.set_header("Content-Type", "text/plain")
if path == "/redirect1":
# Send an HTTP redirect to the bind address of the server
self.redirect(f"{self.request.protocol}://{self.request.host}/hello")
elif path == "/redirect2":
# Send an HTTP redirect to the bind address of the server
self.redirect(f"{self.request.protocol}://{self.request.host}/redirect1")
elif self.path == "/redirect3":
# Send an HTTP redirect to the bind address of the server
self.redirect(f"{self.request.protocol}://{self.request.host}/redirect4")
elif self.path == "/redirect4":
# Send an HTTP redirect to the bind address of the server
self.redirect(f"{self.request.protocol}://{self.request.host}/redirect3")
else:
self.send_response(200)
self.set_header("Content-Type", "text/plain")
def make_app():
return tornado.web.Application([
(r"(/[^/]+)", MainHandler),
])
async def main():
parser = argparse.ArgumentParser()
parser.add_argument("--bind", "-b", default="localhost", help="bind address")
parser.add_argument("--port", "-p", type=int, default=18080, help="bind port")
parser.add_argument("--keyfile", "-k", help="server private key file")
parser.add_argument("--certfile", "-c", help="server certificate file")
parser.add_argument("--pidfile", "-pf", help="path to the PID file")
args = parser.parse_args()
# Create the Tornado application
app = make_app()
# If keyfile and certfile are provided, create an HTTPS server.
# Otherwise, create an HTTP server.
if args.keyfile and args.certfile:
ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_ctx.load_cert_chain(args.certfile, args.keyfile)
server = tornado.httpserver.HTTPServer(app, ssl_options=ssl_ctx)
else:
server = tornado.httpserver.HTTPServer(app)
# Write the PID to the specified PID file, if provided
if args.pidfile:
dummy_killer.write_pid(args.pidfile)
# Start the server
server.bind(args.port, args.bind)
server.start(1)
await asyncio.Event().wait()
if __name__ == "__main__":
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
|
