From 58a889dc6b02150f0405a5142c3a7e9e0c2a5933 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Mon, 10 Aug 2026 09:33:01 +0800 Subject: [PATCH] =?UTF-8?q?modify=20amqtt=20broker=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/chongming/app/__init__.py | 3 +- src/chongming/app/core/lifespan_setup.py | 10 ++- verify_example_plugin.py | 109 +++++++++++++++++++++++ 3 files changed, 120 insertions(+), 2 deletions(-) create mode 100644 verify_example_plugin.py diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py index 071d4c0..f0e132d 100644 --- a/src/chongming/app/__init__.py +++ b/src/chongming/app/__init__.py @@ -41,7 +41,8 @@ async def lifespan(app: FastAPI): session = lifespan_setup.setup_zenoh(app) # 3. mqtt broker - lifespan_setup.setup_mqtt_broker(app, host=config[env]["mqtt"]["host"], port=config[env]["mqtt"]["port"]) + # lifespan_setup.setup_mqtt_broker(app, host=config[env]["mqtt"]["host"], port=config[env]["mqtt"]["port"]) + lifespan_setup.setup_mqtt_broker(app) # 4. WorkerManager manager = lifespan_setup.setup_worker_manager(app, session) diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index 6ff62d6..0e818ac 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -377,8 +377,16 @@ async def register_routes(app: FastAPI, manager: WorkerManager, custom_routers: app.include_router(r, prefix=prefix) logger.info(f"已注册 {len(custom_routers)} 个插件自定义路由（prefix={prefix}）") -def setup_mqtt_broker(app: FastAPI, host: str, port: int): +def setup_mqtt_broker(app: FastAPI): """启动 MQTT Broker 子进程（原子锁确保多 worker 仅启动一个）""" + env_config = config.get(env, {}) + mqtt_config = env_config.get("mqtt", {}) + if not mqtt_config.get("enabled", True): + logger.info("MQTT 已禁用") + return None + + host = mqtt_config.get("host", "localhost") + port = mqtt_config.get("port", 1883) if not _acquire_singleton_lock("mqtt_broker"): for _ in range(50): if is_port_in_use(port, host): diff --git a/verify_example_plugin.py b/verify_example_plugin.py new file mode 100644 index 0000000..5c95e20 --- /dev/null +++ b/verify_example_plugin.py @@ -0,0 +1,109 @@ +"""验证 example 插件 global_var.py / __init__.py 改造效果""" +import sys +from pathlib import Path + +SRC = Path(__file__).parent / "src" +sys.path.insert(0, str(SRC)) + +failures = [] + + +def check(name, cond, detail=""): + status = "PASS" if cond else "FAIL" + print(f"[{status}] {name}" + (f" — {detail}" if detail else "")) + if not cond: + failures.append(name) + + +# ========== 1. global_var 白名单 / require / __all__ ========== +from plugins.example import global_var as gv + +check("global_var.__all__ 包含 db", "db" in gv.__all__) +check("global_var.__all__ 包含 set_global_var", "set_global_var" in gv.__all__) +check("global_var.__all__ 包含 require", "require" in gv.__all__) +check("global_var 不再污染模型命名空间", not hasattr(gv, "AddGetRequest")) + +# 拼错变量名应抛 KeyError（改造前会静默创建新变量） +try: + gv.set_global_var("db2", object()) + check("set_global_var 未知变量应抛 KeyError", False, "未抛错，静默创建了 db2") +except KeyError: + check("set_global_var 未知变量应抛 KeyError", True, "db2 被拒绝") + +# require 在未初始化时应抛清晰错误 +try: + gv.require("db") + check("require 未初始化应抛 RuntimeError", False, "未抛错") +except RuntimeError: + check("require 未初始化应抛 RuntimeError", True) + +# set 后 require 应返回同一个对象 +marker = object() +gv.set_global_var("db", marker) +check("set 后 require 返回同一对象", gv.require("db") is marker) +check("模块属性 db 已被更新", gv.db is marker) + + +# ========== 2. __init__.py 导入链（router + auto_bind_models）========== +import plugins.example as plug + +check("plugins.example 导入成功", True) +check("create_workers 存在", callable(getattr(plug, "create_workers", None))) +check("cleanup 存在", callable(getattr(plug, "cleanup", None))) +check("_initialized 默认 False", plug._initialized is False) + +# ========== 3. create_workers 幂等性（mock session/config）========== +import asyncio + +# 用桩替代 _create_workers，避免真实 zenoh 会话 +calls = [] + + +def _stub_create_workers(session, config, count=1, worker_manager=None): + calls.append((session, config, count, worker_manager)) + return ["task-stub"] + + +plug._create_workers = _stub_create_workers # type: ignore[attr-defined] + + +async def main(): + cfg = {"data_dir": "/tmp/example_data"} + + # 第一次调用：应初始化 db + t1 = await plug.create_workers(session="s1", config=cfg, count=2) + check("首次 create_workers 调用桩", t1 == ["task-stub"]) + check("_initialized 置 True", plug._initialized is True) + from plugins.example.global_var import db as gv_db + check("global_var.db 已初始化", gv_db is not None) + + # 第二次调用：幂等跳过，不重复初始化（桩仍被调用来建 task，但 db 不重建） + t2 = await plug.create_workers(session="s2", config=cfg, count=2) + check("二次 create_workers 幂等返回", t2 == ["task-stub"]) + check("桩被调用两次（建 task）", len(calls) == 2) + from plugins.example.global_var import db as gv_db2 + check("db 对象未重建", gv_db2 is gv_db) + + # cleanup 后复位 + await plug.cleanup("worker/example/worker-i-1") + check("cleanup 后 _initialized 复位 False", plug._initialized is False) + from plugins.example.global_var import db as gv_db3 + check("cleanup 后 db 为 None", gv_db3 is None) + + # cleanup 后再启动：可重新初始化 + t3 = await plug.create_workers(session="s3", config=cfg, count=1) + check("cleanup 后可再次 create_workers", t3 == ["task-stub"]) + from plugins.example.global_var import db as gv_db4 + check("再次初始化后 db 非 None", gv_db4 is not None) + check("再次初始化 db 为新对象", gv_db4 is not gv_db) + + await plug.cleanup("worker/example/worker-i-1") + + +asyncio.run(main()) + +print("\n" + ("=" * 40)) +if failures: + print(f"❌ {len(failures)} 项未通过: {failures}") + sys.exit(1) +print("✅ 全部通过")