Tornado WebSockets 服务的测试需要借助websocket-client 这个包。
安装包
源码地址 https://github.com/liris/websocket-client/
有pip 安装
1pip install websocket-client
tornado 服务器端代码
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
import tornado.web
import tornado.websocket
import tornado.httpserver
import tornado.ioloop
class WebSocketHandler(tornado.websocket.WebSocketHandler):
def open(self):
pass
def on_message(self, message):
self.write_message(u"Your message was: " + message)
def on_close(self):
pass
class Application(tornado.web.Application):
def __init__(self):
handlers = [
(r'/websocket', WebSocketHandler)
]
tornado.web.Application.__init__(self, handlers)
if __name__ == '__main__':
ws_app = Application()
server = tornado.httpserver.HTTPServer(ws_app)
server.listen(8080)
tornado.ioloop.IOLoop.instance().start()
client 代码
1
2
3
4
5
6
7
8
9
10
11
12
import websocket
if __name__ == "__main__":
websocket.enableTrace(True)
ws = websocket.create_connection("ws://127.0.0.1:8080/websocket")
print("Sending 'Hello, World'...")
ws.send("Hello, World")
print("Sent")
print("Receiving...")
result = ws.recv()
print("Received {}".format(result))
ws.close()
运行输出
运行出现类似下面提示：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
--- request header ---
GET /websocket HTTP/1.1
Upgrade: websocket
Connection: Upgrade
Host: 127.0.0.1:8080
Origin: http://127.0.0.1:8080
Sec-WebSocket-Key: ur5A7d4CSA+5RK8rZOhPCQ==
Sec-WebSocket-Version: 13
-----------------------
--- response header ---
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: NwuEWqwSAfFcpoBEX9wajFPpJL0=
-----------------------
Sending 'Hello, World'...
send: '\x81\x8c\xbdP1\xdf\xf55]\xb3\xd2|\x11\x88\xd2"]\xbb'
Sent
Receiving...
Received Your message was: Hello, World
send: '\x88\x82\xd3\xf6\x89\xb7\xd0\x1e'
本文网址: https://pylist.com/topic/62.html 转摘请注明来源