用于python的websocket客户端。支持Hybi13。
dataplicity_websocket_client的Python项目详细描述
Dataplicity WebSocket客户端
这是https://github.com/websocket-client/websocket-client的叉子
它包含对安全WebSocket问题的热修复。除非需要修复，否则您可能需要正式版本（https://pypi.python.org/pypi/websocket-client/0.40.0）
WebSocket客户端
我们的存储库已移动到https://github.com/websocket client/websocket client
websocket客户端模块是用于python的websocket客户端。这为websocket提供了低级api。所有的api都是同步函数。
websocket客户端只支持hybi-13。
许可证
- LGPL
安装
此模块在Python2.7和Python3.x上进行测试。
键入“python setup.py install”或“pip install websocket client”进行安装。
小心！
从v0.16.0开始，我们可以通过python 3的“pip install websocket client”进行安装。
此模块取决于
- six
- backports.ssl_match_hostname for Python 2.x
python 3如何
现在，我们在0.14.0版的单个源代码上支持Python3。谢谢，@battlemidget和@ralphbean。
http代理
支持通过http代理访问websocket。 代理服务器必须允许“连接”方法到WebSocket端口。 默认squid设置为“只允许连接https端口”。
websocket客户端的当前实现是通过代理使用“connect”方法。
示例
importwebsocketws=websocket.WebSocket()ws.connect("ws://example.com/websocket",http_proxy_host="proxy_host_name",http_proxy_port=3128):
示例
长寿命连接
这个例子类似于使用javascript的浏览器中websocket代码的外观。
importwebsocketimportthreadimporttimedefon_message(ws,message):printmessagedefon_error(ws,error):printerrordefon_close(ws):print"### closed ###"defon_open(ws):defrun(*args):foriinrange(3):time.sleep(1)ws.send("Hello %d"%i)time.sleep(1)ws.close()print"thread terminating..."thread.start_new_thread(run,())if__name__=="__main__":websocket.enableTrace(True)ws=websocket.WebSocketApp("ws://echo.websocket.org/",on_message=on_message,on_error=on_error,on_close=on_close)ws.on_open=on_openws.run_forever()
短命一次性发送接收
这是如果你想传达一个短消息，并断开立即完成。
fromwebsocketimportcreate_connectionws=create_connection("ws://echo.websocket.org/")print"Sending 'Hello, World'..."ws.send("Hello, World")print"Sent"print"Receiving..."result=ws.recv()print"Received '%s'"%resultws.close()
如果要自定义套接字选项，请设置sockopt。
sockopt示例
fromwebsocketimportcreate_connectionws=create_connection("ws://echo.websocket.org/",sockopt=((socket.IPPROTO_TCP,socket.TCP_NODELAY),))
更高级：自定义类
如果您想自己处理细节问题，也可以为连接编写自己的类。
fromwebsocketimportcreate_connection,WebSocketclassMyWebSocket(WebSocket):defrecv_frame(self):frame=super().recv_frame()print('yay! I got this frame: ',frame)returnframews=create_connection("ws://echo.websocket.org/",sockopt=((socket.IPPROTO_TCP,socket.TCP_NODELAY),),class_=MyWebSocket)
常见问题
如何禁用ssl证书验证？
请将sslopt设置为{“cert_reqs”：ssl.cert_none}。
WebSocketApp示例
ws=websocket.WebSocketApp("wss://echo.websocket.org")ws.run_forever(sslopt={"cert_reqs":ssl.CERT_NONE})
创建连接示例
ws=websocket.create_connection("wss://echo.websocket.org",sslopt={"cert_reqs":ssl.CERT_NONE})
WebSocket示例
ws=websocket.WebSocket(sslopt={"cert_reqs":ssl.CERT_NONE})ws.connect("wss://echo.websocket.org")
如何禁用主机名验证。
请将sslopt设置为{“check\u hostname”：false}。 （从v0.18.0开始）
WebSocketApp示例
ws=websocket.WebSocketApp("wss://echo.websocket.org")ws.run_forever(sslopt={"check_hostname":False})
创建连接示例
ws=websocket.create_connection("wss://echo.websocket.org",sslopt={"check_hostname":False})
WebSocket示例
ws=websocket.WebSocket(sslopt={"check_hostname":False})ws.connect("wss://echo.websocket.org")
sni支持可用于python 2.7.9+和3.2+。它将尽可能自动启用。
子协议。
服务器需要支持子协议，请这样设置子协议。
子目录示例
ws=websocket.create_connection("ws://exapmle.com/websocket",subprotocols=["binary","base64"])
wsdump.py
py是一个简单的websocket测试（调试）工具。
echo.websocket.org示例：
$ wsdump.py ws://echo.websocket.org/ Press Ctrl+C to quit > Hello, WebSocket < Hello, WebSocket > How are you? < How are you?
用法
用法：
wsdump.py [-h] [-v [VERBOSE]] ws_url
WebSocket简单转储工具
- 位置参数：
- 网址websocket url。例如ws://echo.websocket.org/
- 可选参数：
-h, --help show this help message and exit - websocketapp
-v VERBOSE, --verbose VERBOSE set verbose mode. If set to 1, show opcode. If set to 2, enable to trace websocket module
示例：
$ wsdump.py ws://echo.websocket.org/ $ wsdump.py ws://echo.websocket.org/ -v $ wsdump.py ws://echo.websocket.org/ -vv
欢迎加入QQ群-->： 979659372
