基于websockets状态机的协议实现
wsproto的Python项目详细描述
纯Python,纯状态机WebSocket实现
这个存储库包含一个websocket协议的纯python实现 堆叠。它是从头开始编写的,可以嵌入到任何程序中 选择使用,确保可以通过WebSocket进行通信,如中所定义 RFC6455,不管你的程序是什么 范式。
此存储库不提供解析层、网络层或任何规则 关于并发性。相反,它是一个纯粹的内存解决方案,定义为 数据操作和WebSocket框架。RFC6455和压缩扩展 通过RFC7692的WebSocket已完全 支持。
wsproto支持python 3.6.1或更高版本。
要安装它,只需运行:
$ pip install wsproto
用法
假设您有某种形式的网络套接字可用。wsproto客户端 连接自动生成一个HTTP请求来启动WebSocket 握手。创建WebSocket客户端连接:
fromwsprotoimportWSConnection,ConnectionTypefromwsproto.eventsimportRequestws=WSConnection(ConnectionType.CLIENT)ws.send(Request(host='echo.websocket.org',target='/'))
要创建WebSocket服务器连接:
fromwsproto.connectionimportWSConnection,ConnectionTypews=WSConnection(ConnectionType.SERVER)
每次您发送消息或呼叫ping时,或者仅当您收到传入的消息时 数据,wsproto可能会响应一些您必须发送的传出数据:
some_socket.send(ws.bytes_to_send())
两种连接类型都需要接收传入数据:
ws.receive_data(some_byte_string_of_data)
如果数据包含任何websocket消息或状态更改,wsproto将发出事件:
foreventinws.events():ifisinstance(event,Request):# only client connections get this eventws.send(AcceptConnection())elifisinstance(event,CloseConnection):# guess nobody wants to talk to us any more...elifisinstance(event,TextMessage):print('We got text!',event.data)elifisinstance(event,BytesMessage):print('We got bytes!',event.data)
查看我们的文档以获取活动的完整列表 <;https://wsproto.readthedocs.io/en/latest/api.html events>;!
测试
它通过了高速公路测试套件,在客户端和 服务器模式和使用permessage deflate。
如果要运行符合性测试,请转到符合性目录并 然后要测试客户端模式,请在一个shell中运行高速公路测试服务器:
$ 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欢迎任何人的捐款与许多其他项目不同,我们 乐于接受化妆品和小额捐款,此外 对大型功能的请求和更改
在你投稿之前(通过发行或提交请求书的方式), 请read the contribution guidelines。
许可证
wsproto在mit许可下可用。有关详细信息,请参见 LICENSE存储库中的文件。