数据包监听方法
监听特定数据包
- 在插件主类中使用
self.ListenPacket(packet_id, callback)监听数据包 - 示例:
from tooldelta import Plugin
from tooldelta.constants import PacketIDS
class MyPlugin(Plugin):
...
def __init__(self, frame):
super().__init__(frame)
self.ListenPacket(PacketIDS.Text, self.on_packet)
def on_packet(self, packet):
self.print(packet["SourceName"], packet["Message"])2
3
4
5
6
7
8
9
10
11
- 关于
ListenPacket:参数 类型 说明 packet_id int 数据包 ID callback (dict) -> bool 监听回调函数,返回表示是否拦截数据包
提示
数据包格式可参考 此处,ToolDelta 会将数据包中的类型全部转换为 Python 基本类型。
对于更贴近 ToolDelta 解析的属性名和属性类型,可参考 Golang 实现 中各个数据包的结构体,它们会被直接序列化为 msgpack(一般情况下) 或 json 格式(备用)供 Python 读取。
警告
由于 Python 的 json 模块性能羸弱,在监听所有数据包时,很可能导致 ToolDelta 严重卡顿。
在 1.0.0 版本中,ToolDelta 移除了监听所有数据包的方法。
监听二进制数据包
部分数据包(如区块数据)体积过大,不适合解析为字典,ToolDelta 以二进制形式处理它们。可使用 self.ListenBytesPacket(pkID, cb, priority=0) 监听,此时回调接收的是一个 BaseBytesPacket 子类实例(而非 dict),返回 True 表示拦截该数据包。
关于
ListenBytesPacket:参数 类型 说明 pkID PacketIDS | list[PacketIDS] 数据包 ID,可传入单个 ID 或其列表 cb (BaseBytesPacket) -> bool 监听回调函数,返回表示是否拦截数据包 priority int 可选参数,优先级,数值越大越先执行,默认为 0 示例:
from tooldelta import Plugin
from tooldelta.constants import PacketIDS
class MyPlugin(Plugin):
...
def __init__(self, frame):
super().__init__(frame)
self.ListenBytesPacket(PacketIDS.LevelChunk, self.on_level_chunk)
def on_level_chunk(self, packet):
# packet 为 LevelChunk 实例
...2
3
4
5
6
7
8
9
10
11
12
当前 ToolDelta 支持的二进制数据包及其对应类如下(均位于 tooldelta/mc_bytes_packet/):
| 数据包 | 对应类 |
|---|---|
| SubChunkRequest | SubChunkRequest |
| LevelChunk | LevelChunk |
| SubChunk | SubChunk |
| StructureTemplateDataResponse | StructureTemplateDataResponse |
警告
ListenPacket 与 ListenBytesPacket 不可混用:对二进制数据包 ID 使用 ListenPacket,或对普通字典数据包 ID 使用 ListenBytesPacket,都会抛出异常。请根据需要监听的数据包类型选择对应的方法。
监听框架内广播
可使用 self.ListenInternalBroadcast(broadcast_name, cb, priority=0) 监听指定名字的框架内广播事件,回调接收一个 InternalBroadcast 对象,其返回值会回传给广播的发送者。
- 关于
ListenInternalBroadcast:参数 类型 说明 broadcast_name str 广播事件名 cb (InternalBroadcast) -> Any 监听回调函数,返回值会回传给发送者 priority int 可选参数,优先级,数值越大越先执行,默认为 0
插件间通信的完整用法与示例请见 插件间通信与事件广播。
等待下一个数据包
- 等待并获得租赁服发送至 ToolDelta 的下一个指定 ID 的数据包。
- 示例:
from tooldelta.constants import PacketIDS
class MyPlugin(Plugin):
def do_something(self):
# 等待下一个 Text 数据包
pk = self.frame.packet_handler.wait_next_packet(PacketIDS.Text, 30)
if pk is None:
# 在 30s 内等不到 Text 数据包
# 做点什么...
pass
else:
# 做点什么... (输出数据包内容)
print(pk)2
3
4
5
6
7
8
9
10
11
12
13
14
关于
wait_next_packet:参数 类型 说明 pkID PacketIDS 要等待的包ID timeout float 等待超时时间,单位为秒 返回类型 说明 dict 数据包内容 None 获取超时
添加字典数据包监听器
packet_handler.add_dict_packet_listener(packet_id, cb, priority=0)直接向 PacketHandler 注册一个字典数据包监听器。
| 参数 | 类型 | 说明 |
|---|---|---|
| packet_id | PacketIDS | 数据包 ID |
| cb | (dict) -> bool | 监听回调函数,返回表示是否拦截数据包 |
| priority | int | 可选参数,优先级,数值越大越先执行,默认为 0 |
添加二进制数据包监听器
packet_handler.add_bytes_packet_listener(packet_id, cb, priority=0)直接向 PacketHandler 注册一个二进制数据包监听器。
| 参数 | 类型 | 说明 |
|---|---|---|
| packet_id | PacketIDS | 数据包 ID |
| cb | (BaseBytesPacket) -> bool | 监听回调函数,返回表示是否拦截数据包 |
| priority | int | 可选参数,优先级,数值越大越先执行,默认为 0 |