基于WebSockets状态机协议的实现
项目描述
本存储库包含一个纯Python实现的WebSocket协议栈。它是从头开始编写的,以便可以嵌入到您选择的任何程序中,确保无论您的编程范式如何,都可以通过WebSocket进行通信,正如RFC6455中定义的那样。
此存储库不提供解析层、网络层或任何并发规则。相反,它是一个纯内存解决方案,以数据操作和WebSocket帧为定义。完全支持RFC6455和通过RFC7692的WebSocket压缩扩展。
wsproto支持Python 3.6.1或更高版本。
要安装它,只需运行
$ pip install wsproto
用法
假设您有一些形式的网络套接字可用。wsproto客户端连接会自动生成一个HTTP请求来启动WebSocket握手。要创建WebSocket客户端连接
from wsproto import WSConnection, ConnectionType
from wsproto.events import Request
ws = WSConnection(ConnectionType.CLIENT)
ws.send(Request(host='echo.websocket.org', target='/'))
要创建WebSocket服务器连接
from wsproto.connection import WSConnection, ConnectionType
ws = WSConnection(ConnectionType.SERVER)
每次您发送消息、调用ping或简单地接收传入数据时,wsproto可能会以一些需要发送的输出数据响应
some_socket.send(ws.bytes_to_send())
两种连接类型都需要接收传入数据
ws.receive_data(some_byte_string_of_data)
如果数据包含任何WebSocket消息或状态更改,wsproto将发出事件
for event in ws.events():
if isinstance(event, Request):
# only client connections get this event
ws.send(AcceptConnection())
elif isinstance(event, CloseConnection):
# guess nobody wants to talk to us any more...
elif isinstance(event, TextMessage):
print('We got text!', event.data)
elif isinstance(event, BytesMessage):
print('We got bytes!', event.data)
有关事件的全列表,请参阅我们的文档!(https://wsproto.readthedocs.io/en/latest/api.html#events)
测试
它完全且严格通过了autobahn测试套件,在客户端和服务器模式以及使用permessage-deflate的情况下。
如果您想运行合规性测试,请进入合规性目录,然后进入测试客户端模式,在一个shell中运行Autobahn测试服务器
$ wstest -m fuzzingserver -s ws-fuzzingserver.json
然后在另一个shell中运行测试客户端
$ python test_client.py
要测试服务器模式,运行测试服务器
$ python test_server.py
然后在另一个shell中运行Autobahn测试客户端
$ wstest -m fuzzingclient -s ws-fuzzingclient.json
文档
贡献
wsproto欢迎任何人的贡献!与许多其他项目不同,我们很高兴接受外观贡献和小贡献,除了大型功能请求和更改。
在您贡献之前(无论是通过打开一个问题还是提交一个pull请求),请阅读贡献指南。
许可协议
wsproto是在MIT许可证下提供的。有关更多详情,请参阅存储库中的LICENSE文件。
项目详情
下载文件
下载适合您平台的文件。如果您不确定选择哪个,请了解更多关于安装包的信息。