chuanqi/apilot/stubs/networkctl.py

"""LuaNetworkController stub + msg 对象构造 + 分发。"""
from __future__ import annotations

from typing import Any

from apilot.wire import Frame
from apilot.stubs.netclient import NetClientStub

MSG_LUA = """
local function mkmsg(recog, msgId, p1, p2, p3, body)
    local pos = 1
    local reader = {}
    function reader:ReadString(n)
        local s = string.sub(body, pos, pos + n - 1)
        pos = pos + n
        return s
    end
    local m = {}
    function m:GetMsgId() return msgId end
    function m:GetHeader()
        return {recog=recog, param1=p1, param2=p2, param3=p3, msgId=msgId}
    end
    function m:GetDataLength() return #body end
    function m:GetData() return reader end
    return m
end
return mkmsg
"""


class NetworkCtlStub:
    def __init__(self, nc: NetClientStub) -> None:
        self.nc = nc
        self._tail = 0
        self.handlers: dict[int, Any] = {}
        self._installed: set[int] = set()  # 已 install 的 lua runtime id

    def _next_tail(self) -> int:
        self._tail += 1
        return self._tail

    def send_msg(self, msg_id: int, recog: int, p1: int, p2: int, p3: int,
                 data_str: "bytes | str | None", data_len: int) -> None:
        if data_str is None:
            body = b""
        elif isinstance(data_str, str):
            body = data_str.encode("utf-8")
        else:
            body = bytes(data_str)
        self.nc.send_frame(Frame(recog=recog, msg_id=msg_id, p1=p1, p2=p2, p3=p3,
                                 tail=self._next_tail(), body=body))

    def register_lua_handler(self, msg_id: int, cb: Any) -> None:
        self.handlers[int(msg_id)] = cb

    def unregister_all(self) -> None:
        self.handlers.clear()

    def install_to_lua(self, lua: Any) -> None:
        g = lua.globals()
        g["AP_networkCtl"] = self
        # 冒号调用 networkCtl:SendMsg(...) 兼容:__index 返回 (self, ...) 闭包
        lua.execute("""
networkCtl = networkCtl or {}
setmetatable(networkCtl, {__index = function(_, k)
    local self = AP_networkCtl
    local map = {SendMsg="send_msg",
                 RegisterLuaHandler="register_lua_handler",
                 UnregisterAllLuaHandler="unregister_all"}
    local pyname = map[k]
    if pyname then
        return function(_, ...) return self[pyname](self, ...) end
    end
    return function(...) return nil end
end})
LuaNetworkController = {Inst = function() return networkCtl end}
""")
        lua.execute("AP_mkmsg = (function() " + MSG_LUA + " end)()")
        self._installed.add(id(lua))


def dispatch_frame(lua: Any, ctl: NetworkCtlStub, f: Frame) -> None:
    cb = ctl.handlers.get(f.msg_id)
    if cb is None:
        return
    if id(lua) not in ctl._installed:
        ctl.install_to_lua(lua)
    mkmsg = lua.globals()["AP_mkmsg"]
    body = f.body.decode("latin-1")
    msg = mkmsg(f.recog, f.msg_id, f.p1, f.p2, f.p3, body)
    cb(msg)