modify: 统一插件风格
This commit is contained in:
@@ -33,6 +33,7 @@ config
|
||||
|
||||
# .agents
|
||||
.github
|
||||
.zcode
|
||||
|
||||
# lib
|
||||
lib
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
基于 **FastAPI + Zenoh + MQTT + Vue 3** 的**配置驱动微服务网关平台**。
|
||||
|
||||
通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型与 FastAPI 路由，同时暴露 **HTTP** 和 **MQTT** 双通道。底层基于 Zenoh 分布式消息协议实现 RPC 通信，Worker 插件可独立开发、热加载、水平扩展。
|
||||
通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型与 FastAPI 路由，同时暴露 **HTTP** 和 **MQTT** 双通道。底层基于 Zenoh 分布式消息协议实现 RPC 通信，Worker 插件可独立开发、按副本数水平扩展。
|
||||
|
||||
---
|
||||
|
||||
@@ -13,8 +13,9 @@
|
||||
| **FastAPI + Uvicorn** | 高性能异步 Web 框架，自动生成 OpenAPI 文档 |
|
||||
| **Zenoh 服务发现** | 基于 liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 超时重试 |
|
||||
| **MQTT 集成** | 内嵌 amqtt Broker，MQTT ↔ Worker 双向桥接 |
|
||||
| **动态路由生成** | 扫描 `config/workers/*.toml` 自动注册 HTTP 路由 |
|
||||
| **动态路由生成** | 扫描 `config/workers/*/*.toml` 自动注册 HTTP 路由 |
|
||||
| **Worker 框架** | `@rpc_method` 装饰器 + `Plugin` 基类统一生命周期 + Pydantic 模型绑定 |
|
||||
| **启动契约校验** | TOML 路由 ↔ `@rpc_method` ↔ 生成模型三方一致性校验，fail fast |
|
||||
| **自动化模型生成** | `gen_models` 根据 TOML 配置生成请求/响应模型 |
|
||||
| **插件脚手架** | `gen_plugin` 一键生成风格统一的 Worker 插件骨架 |
|
||||
| **JWT 鉴权** | Rust 实现，路由级 `auth_required` 控制 |
|
||||
@@ -36,15 +37,15 @@ chongming-zenoh/
|
||||
├── config/ # 配置文件目录
|
||||
│ ├── zenoh_config.json5 # Zenoh 会话配置
|
||||
│ └── workers/ # Worker 路由定义（TOML）
|
||||
│ ├── config.toml # 插件注册列表
|
||||
│ ├── auth/ # 鉴权插件（login / register / me / …）
|
||||
│ ├── config.toml # 旧式全局插件注册（仅作兜底兼容）
|
||||
│ ├── auth/ # 鉴权插件（config.toml = 插件注册 + 运行时配置）
|
||||
│ ├── doubao/ # 豆包视觉理解插件
|
||||
│ ├── aibox_sophon_adapter/ # AI Box 设备心跳监控插件
|
||||
│ └── example/ # 示例插件（四则运算 / MQTT 端点）
|
||||
├── src/
|
||||
│ ├── chongming/ # 核心应用
|
||||
│ │ ├── app/
|
||||
│ │ │ ├── core/ # WorkerManager, RouteGenerator, MQTTBridge, auth
|
||||
│ │ │ ├── core/ # WorkerManager, RouteGenerator, MQTTBridge, contracts(契约校验), auth
|
||||
│ │ │ └── routes/ # 内置路由（文件上传等）
|
||||
│ │ └── scripts/ # build, gen_models, gen_plugin
|
||||
│ ├── chongming-web/ # Vue 3 前端
|
||||
@@ -211,17 +212,34 @@ async def upload(file: UploadFile = File(...)):
|
||||
```toml
|
||||
# config/workers/aibox_sophon_adapter/config.toml
|
||||
[[workers.mqtt_topics]]
|
||||
topic = "/board_ping" # 支持 +（单级）/ #（多级，须在末尾）通配符
|
||||
topic = "box/aibox_sophon/board_ping" # 支持 +（单级）/ #（多级，须在末尾）通配符
|
||||
subject = "aibox_sophon_adapter.board_ping" # 可选，{1} {2}... 对应通配符捕获值
|
||||
response_topic = "/board_ping/response" # 可选，响应主题；默认 "<实际主题>/response"
|
||||
```
|
||||
|
||||
- `subject` 缺省为 `<插件类型>.<topic 最后一段>`；`response_topic` 缺省为 `<实际匹配主题>/response`。
|
||||
- 网关发现插件时自动收集所有 `[[workers.mqtt_topics]]`，MQTT Bridge 一并订阅并路由。
|
||||
- 请求负载同样支持 `_method` / `_timeout` / `_cacheable` 控制字段。
|
||||
- 多条自定义主题按配置顺序匹配，先匹配者优先；内置 `chongming/...` 接口最后兜底。
|
||||
- 全局 `[<env>.mqtt].custom_topics` 仍可作为通用兜底（不依赖具体插件时使用，可选）。
|
||||
|
||||
### 请求负载格式（信封协议）
|
||||
|
||||
MQTT 请求负载为信封结构，控制字段与业务数据分离：
|
||||
|
||||
```json
|
||||
{
|
||||
"meta": {
|
||||
"method": "GET", // 可选，默认 GET；MQTT 专用端点需显式 "MQTT"
|
||||
"timeout": 30, // 可选，RPC 超时（秒），默认 30
|
||||
"cacheable": false // 可选，是否允许结果缓存，默认 false
|
||||
},
|
||||
"data": { "a": 3, "b": 5 } // 业务参数，传给 Worker
|
||||
}
|
||||
```
|
||||
|
||||
- 非信封格式的负载会被网关丢弃并告警（严格协议，设备侧请按此格式发布）。
|
||||
- HTTP 通道由网关自动组装信封（注入 method / user_info），无需调用方关心。
|
||||
|
||||
### 快速测试
|
||||
|
||||
```bash
|
||||
@@ -243,14 +261,19 @@ python publisher.py example add '{"a":3,"b":5}' # 调用 add
|
||||
### 添加新 Worker 插件
|
||||
|
||||
```text
|
||||
1. 创建 src/plugins/<name>/ — __init__.py, global_var.py, router/, models/
|
||||
2. 注册到 config/workers/config.toml 的 [[workers]]
|
||||
1. 创建 src/plugins/<name>/ — __init__.py（Plugin 子类）, global_var.py, router/, models/
|
||||
2. 创建 config/workers/<name>/config.toml，在 [workers] 节注册插件
|
||||
（type / module / factory / cleanup / count + [workers.config] 运行时配置）
|
||||
3. 创建 config/workers/<name>/*.toml 端点定义
|
||||
4. 运行 uv run python -m chongming.scripts.gen_models 生成模型
|
||||
5. 实现 router/<op>.py，引用 models/pydantic.py 中的模型
|
||||
6. 更新 router/__init__.py 导入
|
||||
```
|
||||
|
||||
> 推荐用脚手架一步生成骨架：`uv run gen_plugin <name> [operation ...]`。
|
||||
> 启动时网关会做**契约校验**：TOML 路由缺少对应 `@rpc_method` 将导致启动失败，
|
||||
> 模型缺失/孤儿等不一致以警告日志提示。
|
||||
|
||||
### 为现有插件添加新端点
|
||||
|
||||
```text
|
||||
@@ -303,7 +326,7 @@ uv pip install -e utils/<name> # 安装/更新工具包
|
||||
|
||||
- **应用配置**：根目录 `config.toml`，包含环境、服务器、安全、MQTT 等全局设置
|
||||
- **Zenoh 配置**：`config/zenoh_config.json5`
|
||||
- **Worker 注册**：`config/workers/config.toml`，定义加载的插件及其副本数
|
||||
- **插件注册**：`config/workers/<name>/config.toml` 的 `[workers]` 节，定义加载方式、副本数与运行时配置（旧全局 `config/workers/config.toml` 仅作兜底兼容）
|
||||
- **路由定义**：`config/workers/<name>/*.toml`，每个端点一个文件
|
||||
|
||||
详细配置说明见 [`config/README.md`](config/README.md)。
|
||||
@@ -339,7 +362,7 @@ uv pip install -e utils/<name> # 安装/更新工具包子包
|
||||
|
||||
| 字段 | 位置 | 说明 |
|
||||
| -------------------- | --------------------- | --------------------------------------- |
|
||||
| `count` | `workers/config.toml` | Worker 副本数 |
|
||||
| `count` | `<插件>/config.toml` | Worker 副本数（`[workers]` 节） |
|
||||
| `router_prefix` | `<插件>/config.toml` | API 路由前缀 |
|
||||
| `defaults.timeout` | `<插件>/config.toml` | 默认超时（秒） |
|
||||
| `defaults.cacheable` | `<插件>/config.toml` | 是否缓存 RPC 结果 |
|
||||
|
||||
+8
-4
@@ -4,9 +4,10 @@ MQTT Publisher — 向 chongming Worker 发送 MQTT 请求
|
||||
用法:
|
||||
python publisher.py <worker_type> <operation> [json_payload]
|
||||
|
||||
示例:
|
||||
python publisher.py example add '{"a": 3, "b": 5, "_method": "GET"}'
|
||||
python publisher.py doubao analyze '{"text": "hello", "_method": "POST"}'
|
||||
示例（信封协议，meta 可选）:
|
||||
python publisher.py example add '{"meta": {"method": "GET"}, "data": {"a": 3, "b": 5}}'
|
||||
python publisher.py example add '{"a": 3, "b": 5}' # 扁平载荷自动包装（method=GET）
|
||||
python publisher.py example add '{"meta": {"method": "MQTT"}, "data": {"a": 3, "b": 5}}'
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
@@ -24,7 +25,7 @@ PREFIX = "chongming"
|
||||
# 默认参数
|
||||
DEFAULT_WORKER = "example"
|
||||
DEFAULT_OP = "add"
|
||||
DEFAULT_PAYLOAD = {"a": 3, "b": 5, "_method": "GET"}
|
||||
DEFAULT_PAYLOAD = {"meta": {"method": "GET"}, "data": {"a": 3, "b": 5}}
|
||||
|
||||
async def main():
|
||||
worker_type = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_WORKER
|
||||
@@ -33,6 +34,9 @@ async def main():
|
||||
|
||||
topic = f"{PREFIX}/{worker_type}/{operation}"
|
||||
payload = json.loads(payload_str)
|
||||
# 扁平载荷自动包装为信封（method 默认 GET），已含 meta/data 则原样使用
|
||||
if not (isinstance(payload, dict) and ("meta" in payload or "data" in payload)):
|
||||
payload = {"meta": {"method": "GET"}, "data": payload}
|
||||
|
||||
client = MQTTClient()
|
||||
await client.connect("mqtt://localhost:1883/")
|
||||
|
||||
+1
-1
@@ -56,11 +56,11 @@ chongming-request = { git = "http://124.71.68.6:3000/chongming/chongming-utils-r
|
||||
chongming-jwt = { git = "http://124.71.68.6:3000/chongming/chongming-utils-jwt.git" }
|
||||
chongming-config = { git = "http://124.71.68.6:3000/chongming/chongming-utils-config.git" }
|
||||
chongming-password = { git = "http://124.71.68.6:3000/chongming/chongming-utils-password.git" }
|
||||
chongming-worker = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git" }
|
||||
bytedance = { git = "http://124.71.68.6:3000/chongming/bytedance.git" }
|
||||
chongming-stream-manager = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git" }
|
||||
chongming-database = { git = "http://124.71.68.6:3000/chongming/chongming-utils-database.git" }
|
||||
chongming-mqtt = { git = "http://124.71.68.6:3000/chongming/chongming-mqtt.git" }
|
||||
chongming-worker = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git"}
|
||||
|
||||
[project.scripts]
|
||||
build = "chongming.scripts.build:build"
|
||||
|
||||
@@ -50,6 +50,9 @@ async def lifespan(app: FastAPI):
|
||||
# 5. 加载 workers（同时自动发现插件自定义路由）
|
||||
worker_tasks, custom_routers = await lifespan_setup.load_workers(app, session, manager)
|
||||
|
||||
# 5.5 契约校验（TOML 路由 ↔ @rpc_method ↔ 生成模型），失败则启动终止
|
||||
lifespan_setup.validate_contracts(app)
|
||||
|
||||
# 6. 注册路由（含 TOML 动态路由 + 插件自定义路由）
|
||||
await lifespan_setup.register_routes(app, manager, custom_routers)
|
||||
|
||||
|
||||
@@ -0,0 +1,168 @@
|
||||
"""
|
||||
启动契约校验（config/workers TOML ↔ @rpc_method 注册表 ↔ gen_models 生成模型）
|
||||
|
||||
挂载于 lifespan 的 load_workers 与 register_routes 之间，此时插件模块已导入
|
||||
（@rpc_method 注册表就绪）且 TOML 路由尚未注册到 FastAPI。
|
||||
|
||||
策略:
|
||||
- 硬失败（汇总后 raise，启动终止）:
|
||||
HTTP 路由（GET/POST/PUT/DELETE/PATCH 且 internal != true）在注册表中
|
||||
缺少对应 subject#METHOD 的 @rpc_method → 该路由运行时必然 503，提前失败。
|
||||
- 软警告（仅日志汇总，不阻断启动）:
|
||||
[MQTT] / internal 路由缺少注册、已注册方法无 TOML 路由、
|
||||
TOML 声明 params/response_model 但生成模型缺失（gen_models 过期）、孤儿模型。
|
||||
"""
|
||||
import importlib
|
||||
from typing import Any, Dict, List, Optional, Set
|
||||
|
||||
from chongming_worker import get_all_rpc_methods
|
||||
|
||||
from .route_generator import load_worker_routes_from_dir
|
||||
from .logger import get_logger
|
||||
|
||||
logger = get_logger("contracts")
|
||||
|
||||
# 参与 HTTP 路由注册的方法（与 route_generator 的注册行为一致）
|
||||
HTTP_METHODS = ("GET", "POST", "PUT", "DELETE", "PATCH")
|
||||
|
||||
|
||||
def _model_name(base: str, method: str, suffix: str) -> str:
|
||||
"""与 gen_models._model_name 保持一致的模型命名: <Op><Method><Suffix>"""
|
||||
method_title = "Mqtt" if method == "MQTT" else method.title()
|
||||
return f"{base.title().replace('-', '').replace('_', '')}{method_title}{suffix}"
|
||||
|
||||
|
||||
def _load_plugin_models(worker_type: str) -> Optional[Set[str]]:
|
||||
"""
|
||||
加载插件 models.pydantic 中所有 Pydantic 模型的类名集合。
|
||||
|
||||
返回 None 表示模块不存在（插件无生成模型，跳过模型相关校验）。
|
||||
注: gen_models 生成的 `<Plugin>Response` 联合类型——多模型时为 UnionType
|
||||
（自动被 isinstance(type) 过滤），单模型时是类的别名（与真实模型同一对象，
|
||||
按对象去重排除），两者都不会被误报为孤儿模型。
|
||||
"""
|
||||
from pydantic import BaseModel
|
||||
|
||||
for mod_name in (
|
||||
f"plugins.{worker_type}.models.pydantic",
|
||||
f"src.plugins.{worker_type}.models.pydantic",
|
||||
):
|
||||
try:
|
||||
mod = importlib.import_module(mod_name)
|
||||
except ImportError:
|
||||
continue
|
||||
names: Set[str] = set()
|
||||
seen_objs: Set[int] = set()
|
||||
for name in dir(mod):
|
||||
obj = getattr(mod, name)
|
||||
if (
|
||||
isinstance(obj, type)
|
||||
and issubclass(obj, BaseModel)
|
||||
and obj is not BaseModel
|
||||
and id(obj) not in seen_objs
|
||||
):
|
||||
seen_objs.add(id(obj))
|
||||
names.add(name)
|
||||
return names
|
||||
return None
|
||||
|
||||
|
||||
def validate_contracts(plugins: List[Dict[str, Any]]) -> None:
|
||||
"""
|
||||
对每个发现的插件执行三方一致性校验。
|
||||
|
||||
Args:
|
||||
plugins: _discover_plugins() 返回的插件配置列表
|
||||
|
||||
Raises:
|
||||
RuntimeError: 存在硬失败项（HTTP 路由缺失 @rpc_method 注册）时
|
||||
"""
|
||||
registry = get_all_rpc_methods()
|
||||
hard_errors: List[str] = []
|
||||
warnings: List[str] = []
|
||||
|
||||
for wc in plugins:
|
||||
worker_type = wc.get("type", "")
|
||||
if not worker_type:
|
||||
continue
|
||||
|
||||
toml_routes = load_worker_routes_from_dir(worker_type)
|
||||
registered = registry.get(worker_type, {})
|
||||
toml_keys = {f"{r['subject']}#{r['method']}" for r in toml_routes}
|
||||
model_names = _load_plugin_models(worker_type)
|
||||
|
||||
for r in toml_routes:
|
||||
key = f"{r['subject']}#{r['method']}"
|
||||
route_desc = f"{r['method']} {r['path']} -> subject '{r['subject']}'"
|
||||
|
||||
if key in registered:
|
||||
continue
|
||||
|
||||
if r["method"] in HTTP_METHODS and not r.get("internal", False):
|
||||
# 硬失败：运行时必然 503
|
||||
hint = ""
|
||||
if not registered:
|
||||
hint = "（该插件注册表为空，请检查启动日志中的 '加载 Worker 插件失败' 错误）"
|
||||
hard_errors.append(
|
||||
f"[{worker_type}] {route_desc} 缺少对应的 @rpc_method 注册{hint}"
|
||||
)
|
||||
else:
|
||||
# 软警告：MQTT / internal 端点缺失注册
|
||||
warnings.append(
|
||||
f"[{worker_type}] {route_desc} 无对应的 @rpc_method(method='{r['method']}') 注册"
|
||||
)
|
||||
|
||||
# 已注册方法无 TOML 路由（反向漂移检测）
|
||||
for key in registered:
|
||||
if key not in toml_keys:
|
||||
warnings.append(
|
||||
f"[{worker_type}] 已注册 '{key}' 但无对应 TOML 路由定义"
|
||||
)
|
||||
|
||||
# 模型一致性（仅当插件存在生成模型时校验）
|
||||
if model_names is not None:
|
||||
referenced: Set[str] = set()
|
||||
for r in toml_routes:
|
||||
route_name = r.get("route_name", "")
|
||||
if not route_name:
|
||||
continue
|
||||
method = r["method"]
|
||||
if r.get("params"):
|
||||
expected = _model_name(route_name, method, "Request")
|
||||
referenced.add(expected)
|
||||
if expected not in model_names:
|
||||
warnings.append(
|
||||
f"[{worker_type}] '{r['subject']}' [{method}] 声明了 params，"
|
||||
f"但模型 {expected} 不存在（可能未运行 gen_models）"
|
||||
)
|
||||
if r.get("response_model"):
|
||||
expected = _model_name(route_name, method, "Response")
|
||||
referenced.add(expected)
|
||||
if expected not in model_names:
|
||||
warnings.append(
|
||||
f"[{worker_type}] '{r['subject']}' [{method}] 声明了 response_model，"
|
||||
f"但模型 {expected} 不存在（可能未运行 gen_models）"
|
||||
)
|
||||
|
||||
for name in sorted(model_names - referenced):
|
||||
warnings.append(
|
||||
f"[{worker_type}] 模型 {name} 未被任何 TOML 路由引用"
|
||||
f"（孤儿模型，可重新运行 gen_models 清理）"
|
||||
)
|
||||
|
||||
# 输出结果
|
||||
if warnings:
|
||||
logger.warning("契约校验发现 %d 个警告:", len(warnings))
|
||||
for w in warnings:
|
||||
logger.warning(" ⚠ %s", w)
|
||||
|
||||
if hard_errors:
|
||||
logger.error("契约校验失败，发现 %d 个硬性错误:", len(hard_errors))
|
||||
for e in hard_errors:
|
||||
logger.error(" ✗ %s", e)
|
||||
raise RuntimeError(
|
||||
"启动契约校验失败: "
|
||||
+ "; ".join(hard_errors)
|
||||
)
|
||||
|
||||
logger.info("契约校验通过（TOML 路由 ↔ @rpc_method 注册表一致）")
|
||||
@@ -21,6 +21,11 @@ from ..core.mqtt_bridge import MQTTBridge
|
||||
config = get_config()
|
||||
logger = get_logger("app")
|
||||
|
||||
# 挂载框架/插件父 logger 的 handler：logger 模块只为具名 logger 挂 handler，
|
||||
# 子 logger（chongming_worker.runtime、plugins.doubao 等）经 propagate 到父级才可见
|
||||
get_logger("chongming_worker")
|
||||
get_logger("plugins")
|
||||
|
||||
try:
|
||||
env = config["default"]["env"]
|
||||
except KeyError: # pragma: no cover
|
||||
@@ -377,6 +382,19 @@ 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 validate_contracts(app: FastAPI | None = None):
|
||||
"""
|
||||
契约校验（TOML 路由 ↔ @rpc_method 注册表 ↔ gen_models 生成模型）。
|
||||
|
||||
在 load_workers 之后、register_routes 之前调用：此时插件模块已导入
|
||||
（注册表就绪）。HTTP 路由缺失对应 @rpc_method 时抛 RuntimeError 终止启动，
|
||||
其余不一致以警告日志汇总。
|
||||
"""
|
||||
from ..core.contracts import validate_contracts as _validate
|
||||
|
||||
_validate(_discover_plugins())
|
||||
|
||||
def setup_mqtt_broker(app: FastAPI):
|
||||
"""启动 MQTT Broker 子进程（原子锁确保多 worker 仅启动一个）"""
|
||||
env_config = config.get(env, {})
|
||||
|
||||
@@ -97,7 +97,19 @@ class MQTTBridge:
|
||||
t: str = m.topic; p = m.data # type: ignore
|
||||
try: pl = json.loads(p.decode("utf-8"))
|
||||
except json.JSONDecodeError: return
|
||||
pl.setdefault("_method", "GET")
|
||||
|
||||
# 信封协议（严格）: {"meta": {...}, "data": {...}}
|
||||
# 非信封格式载荷直接丢弃并告警，请设备侧按新协议发布
|
||||
if not isinstance(pl, dict) or ("meta" not in pl and "data" not in pl):
|
||||
logger.warning(
|
||||
"丢弃非信封格式载荷 [%s]: 请使用 {'meta': {...}, 'data': {...}} 格式", t
|
||||
)
|
||||
return
|
||||
meta = pl.get("meta") or {}
|
||||
data = pl.get("data") or {}
|
||||
if not isinstance(data, dict):
|
||||
data = {}
|
||||
pl = {"meta": meta, "data": data}
|
||||
|
||||
# 1. 优先匹配显式配置的自定义订阅主题（顺序即优先级）
|
||||
for ct, compiled in self._compiled:
|
||||
@@ -136,9 +148,11 @@ class MQTTBridge:
|
||||
|
||||
async def _route(self, wt: str, subj: str, pl: dict, rt: str):
|
||||
logger.info(f"MQTT -> {subj}")
|
||||
meta = pl.get("meta") or {}
|
||||
try:
|
||||
r = await self._wm.call(worker_type=wt, subject=subj, payload=pl,
|
||||
timeout=float(pl.pop("_timeout",30)), cacheable=bool(pl.pop("_cacheable",False)))
|
||||
timeout=float(meta.get("timeout", 30)),
|
||||
cacheable=bool(meta.get("cacheable", False)))
|
||||
if r is not None: await self._pub(rt, r)
|
||||
except RuntimeError as e: await self._pub(rt, {"error":str(e)})
|
||||
except Exception as e: await self._pub(rt, {"error":str(e)})
|
||||
|
||||
@@ -186,6 +186,8 @@ def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]:
|
||||
if "subject" not in route:
|
||||
# 自动生成 subject: worker_type.route_name (如 "calc.add")
|
||||
route["subject"] = f"{worker_type}.{route_name}"
|
||||
# 记录来源文件名（契约校验用其推断 gen_models 模型命名）
|
||||
route["route_name"] = route_name
|
||||
|
||||
routes.append(route)
|
||||
logger.debug(
|
||||
@@ -268,11 +270,12 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t
|
||||
body: RequestModel, # type: ignore
|
||||
) -> Any:
|
||||
try:
|
||||
payload = body.model_dump()
|
||||
payload.setdefault("_method", http_method)
|
||||
# 信封协议: {"meta": {...}, "data": {...}}
|
||||
meta = {"method": http_method}
|
||||
user_info = getattr(request.state, "user", None)
|
||||
if user_info:
|
||||
payload["user_info"] = user_info
|
||||
meta["user_info"] = user_info
|
||||
payload = {"meta": meta, "data": body.model_dump()}
|
||||
result = await worker_manager.call(
|
||||
worker_type=worker_type,
|
||||
subject=subject,
|
||||
@@ -313,11 +316,12 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
|
||||
# 无参数情况
|
||||
if not param_spec_combined:
|
||||
async def _simple_endpoint(request: Request) -> Any:
|
||||
payload = {}
|
||||
payload.setdefault("_method", http_method)
|
||||
# 信封协议: {"meta": {...}, "data": {...}}
|
||||
meta = {"method": http_method}
|
||||
user_info = getattr(request.state, "user", None)
|
||||
if user_info:
|
||||
payload["user_info"] = user_info
|
||||
meta["user_info"] = user_info
|
||||
payload = {"meta": meta, "data": {}}
|
||||
try:
|
||||
result = await worker_manager.call(
|
||||
worker_type=worker_type,
|
||||
@@ -339,11 +343,12 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
|
||||
param_names = list(param_spec_combined.keys())
|
||||
|
||||
async def endpoint(request: Request, **kwargs: Any) -> Any:
|
||||
payload = {k: kwargs[k] for k in param_names}
|
||||
payload.setdefault("_method", http_method)
|
||||
# 信封协议: {"meta": {...}, "data": {...}}
|
||||
meta = {"method": http_method}
|
||||
user_info = getattr(request.state, "user", None)
|
||||
if user_info:
|
||||
payload["user_info"] = user_info
|
||||
meta["user_info"] = user_info
|
||||
payload = {"meta": meta, "data": {k: kwargs[k] for k in param_names}}
|
||||
try:
|
||||
result = await worker_manager.call(
|
||||
worker_type=worker_type,
|
||||
@@ -394,7 +399,7 @@ def _register_single_route(
|
||||
# MQTT 端点：不注册 HTTP 路由，仅通过 MQTT Bridge 调用
|
||||
if rc.method == "MQTT":
|
||||
logger.info(
|
||||
f"MQTT 端点: subject={worker_type}.{rc.subject} (仅 MQTT，不注册 HTTP 路由)"
|
||||
f"MQTT 端点: subject={rc.subject} (仅 MQTT，不注册 HTTP 路由)"
|
||||
)
|
||||
return
|
||||
|
||||
|
||||
+2
-1
@@ -20,7 +20,8 @@ def fake_create_workers(session, config, count=1, worker_manager=None):
|
||||
|
||||
|
||||
async def main():
|
||||
auth_plugin._create_workers = fake_create_workers
|
||||
# 替换 Plugin 基类的 worker 创建实现，避免依赖 zenoh 会话
|
||||
auth_plugin.plugin._create_workers_impl = fake_create_workers
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config = {"type": "auth", "data_dir": tmp}
|
||||
|
||||
+26
-11
@@ -99,20 +99,20 @@ async def test_custom_routing():
|
||||
)
|
||||
|
||||
# 1. 默认 subject = <worker>.<最后一段>；响应 = <实际主题>/response
|
||||
await bridge._handle(make_msg("devices/dev-01/telemetry", {"v": 1}))
|
||||
await bridge._handle(make_msg("devices/dev-01/telemetry", {"meta": {}, "data": {"v": 1}}))
|
||||
assert wm.calls[-1]["worker_type"] == "device"
|
||||
assert wm.calls[-1]["subject"] == "device.telemetry"
|
||||
assert published[-1][0] == "devices/dev-01/telemetry/response"
|
||||
|
||||
# 2. 带占位符 + 自定义响应主题
|
||||
await bridge._handle(make_msg("foo/a/b", {"v": 2}))
|
||||
await bridge._handle(make_msg("foo/a/b", {"meta": {}, "data": {"v": 2}}))
|
||||
assert wm.calls[-1]["worker_type"] == "foo"
|
||||
assert wm.calls[-1]["subject"] == "foo.handle.a/b"
|
||||
assert published[-1][0] == "out/a/b/resp"
|
||||
|
||||
# 3. 内置 chongming 兜底
|
||||
await bridge._handle(
|
||||
make_msg(f"{MQTT_TOPIC_PREFIX}/example/add", {"a": 1, "b": 2})
|
||||
make_msg(f"{MQTT_TOPIC_PREFIX}/example/add", {"meta": {}, "data": {"a": 1, "b": 2}})
|
||||
)
|
||||
assert wm.calls[-1]["worker_type"] == "example"
|
||||
assert wm.calls[-1]["subject"] == "example.add"
|
||||
@@ -120,7 +120,7 @@ async def test_custom_routing():
|
||||
print("✅ 自定义主题路由 + 内置兜底通过")
|
||||
|
||||
|
||||
async def test_custom_routing_control_fields():
|
||||
async def test_envelope_meta_fields():
|
||||
wm = FakeWM()
|
||||
bridge, published = make_bridge(
|
||||
wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
|
||||
@@ -128,14 +128,27 @@ async def test_custom_routing_control_fields():
|
||||
await bridge._handle(
|
||||
make_msg(
|
||||
"devices/d1/telemetry",
|
||||
{"_timeout": 5, "_cacheable": True, "_method": "POST"},
|
||||
{"meta": {"timeout": 5, "cacheable": True, "method": "POST"}, "data": {}},
|
||||
)
|
||||
)
|
||||
assert wm.calls[-1]["timeout"] == 5.0
|
||||
assert wm.calls[-1]["cacheable"] is True
|
||||
assert "_timeout" not in wm.calls[-1]["payload"]
|
||||
assert wm.calls[-1]["payload"].get("_method") == "POST"
|
||||
print("✅ 控制字段 (_timeout/_cacheable/_method) 透传通过")
|
||||
payload = wm.calls[-1]["payload"]
|
||||
assert payload["meta"]["method"] == "POST"
|
||||
assert payload["data"] == {}
|
||||
print("✅ 信封 meta 字段 (method/timeout/cacheable) 透传通过")
|
||||
|
||||
|
||||
async def test_flat_payload_dropped():
|
||||
"""严格信封协议：扁平载荷（无 meta/data）应被丢弃且不调用 worker"""
|
||||
wm = FakeWM()
|
||||
bridge, published = make_bridge(
|
||||
wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
|
||||
)
|
||||
await bridge._handle(make_msg("devices/d1/telemetry", {"_method": "POST", "v": 1}))
|
||||
assert wm.calls == []
|
||||
assert published == []
|
||||
print("✅ 非信封（扁平）载荷被丢弃通过")
|
||||
|
||||
|
||||
async def test_non_json_dropped():
|
||||
@@ -165,7 +178,7 @@ async def test_board_ping_topic():
|
||||
}
|
||||
],
|
||||
)
|
||||
await bridge._handle(make_msg("/board_ping", {"BoardId": "B001", "Status": 1}))
|
||||
await bridge._handle(make_msg("/board_ping", {"meta": {}, "data": {"BoardId": "B001", "Status": 1}}))
|
||||
assert wm.calls[-1]["worker_type"] == "aibox_sophon_adapter"
|
||||
assert wm.calls[-1]["subject"] == "aibox_sophon_adapter.board_ping"
|
||||
assert published[-1][0] == "/board_ping/response"
|
||||
@@ -182,7 +195,8 @@ def test_plugin_config_driven_collection():
|
||||
assert "aibox_sophon_adapter" in by_type, "未发现 aibox_sophon_adapter 插件"
|
||||
topics = by_type["aibox_sophon_adapter"]["mqtt_topics"]
|
||||
assert any(
|
||||
t["topic"] == "/board_ping" and t["worker"] == "aibox_sophon_adapter"
|
||||
t["topic"] == "box/aibox_sophon/board_ping"
|
||||
and t["worker"] == "aibox_sophon_adapter"
|
||||
and t["subject"] == "aibox_sophon_adapter.board_ping"
|
||||
for t in topics
|
||||
), topics
|
||||
@@ -194,7 +208,8 @@ def test_plugin_config_driven_collection():
|
||||
async def main():
|
||||
test_compile_topic()
|
||||
await test_custom_routing()
|
||||
await test_custom_routing_control_fields()
|
||||
await test_envelope_meta_fields()
|
||||
await test_flat_payload_dropped()
|
||||
await test_non_json_dropped()
|
||||
await test_board_ping_topic()
|
||||
test_plugin_config_driven_collection()
|
||||
|
||||
+5
-5
@@ -30,11 +30,11 @@ REQUEST_TOPIC = f"{PREFIX}/{WORKER}/{OPERATION}"
|
||||
RESPONSE_TOPIC = f"{PREFIX}/{WORKER}/{OPERATION}/response"
|
||||
BROKER_URL = "mqtt://localhost:1883/"
|
||||
|
||||
# 测试用例
|
||||
# 测试用例（信封协议: meta.method 控制分派，data 携带业务参数）
|
||||
TEST_CASES = [
|
||||
{"a": 3, "b": 5, "_method": "MQTT", "expected": 8},
|
||||
{"a": -1, "b": 10, "_method": "MQTT", "expected": 9},
|
||||
{"a": 3.14, "b": 2.86, "_method": "MQTT", "expected": 6.0},
|
||||
{"a": 3, "b": 5, "expected": 8},
|
||||
{"a": -1, "b": 10, "expected": 9},
|
||||
{"a": 3.14, "b": 2.86, "expected": 6.0},
|
||||
]
|
||||
|
||||
PASS = 0
|
||||
@@ -50,7 +50,7 @@ async def run_test():
|
||||
|
||||
for i, case in enumerate(TEST_CASES):
|
||||
a, b, expected = case["a"], case["b"], case["expected"]
|
||||
payload = {"a": a, "b": b, "_method": "MQTT"}
|
||||
payload = {"meta": {"method": "MQTT"}, "data": {"a": a, "b": b}}
|
||||
payload_bytes = json.dumps(payload).encode("utf-8")
|
||||
|
||||
await client.publish(REQUEST_TOPIC, payload_bytes, QOS_1)
|
||||
|
||||
@@ -319,7 +319,7 @@ source = { git = "http://124.71.68.6:3000/chongming/chongming-stream-manager.git
|
||||
[[package]]
|
||||
name = "chongming-worker"
|
||||
version = "0.1.0"
|
||||
source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#4b8985d5bd4fc45894bc3e27ecbd541277b83e2f" }
|
||||
source = { git = "http://124.71.68.6:3000/chongming/chongming-utils-worker.git#a58e92c21e5bf4b3ada55b2abc894e8ead7d17bc" }
|
||||
dependencies = [
|
||||
{ name = "eclipse-zenoh" },
|
||||
{ name = "tomli" },
|
||||
|
||||
@@ -44,18 +44,19 @@ check("set 后 require 返回同一对象", gv.require("db") is marker)
|
||||
check("模块属性 db 已被更新", gv.db is marker)
|
||||
|
||||
|
||||
# ========== 2. __init__.py 导入链（router + auto_bind_models）==========
|
||||
# ========== 2. __init__.py 导入链（router + Plugin 基类）==========
|
||||
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)
|
||||
check("plugin 实例存在", getattr(plug, "plugin", None) is not None)
|
||||
check("plugin._initialized 默认 False", plug.plugin._initialized is False)
|
||||
|
||||
# ========== 3. create_workers 幂等性（mock session/config）==========
|
||||
import asyncio
|
||||
|
||||
# 用桩替代 _create_workers，避免真实 zenoh 会话
|
||||
# 用桩替代 Plugin 基类的 worker 创建实现，避免真实 zenoh 会话
|
||||
calls = []
|
||||
|
||||
|
||||
@@ -64,7 +65,7 @@ def _stub_create_workers(session, config, count=1, worker_manager=None):
|
||||
return ["task-stub"]
|
||||
|
||||
|
||||
plug._create_workers = _stub_create_workers # type: ignore[attr-defined]
|
||||
plug.plugin._create_workers_impl = _stub_create_workers # type: ignore[method-assign]
|
||||
|
||||
|
||||
async def main():
|
||||
@@ -73,7 +74,7 @@ async def main():
|
||||
# 第一次调用：应初始化 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)
|
||||
check("plugin._initialized 置 True", plug.plugin._initialized is True)
|
||||
from plugins.example.global_var import db as gv_db
|
||||
check("global_var.db 已初始化", gv_db is not None)
|
||||
|
||||
@@ -86,7 +87,7 @@ async def main():
|
||||
|
||||
# cleanup 后复位
|
||||
await plug.cleanup("worker/example/worker-i-1")
|
||||
check("cleanup 后 _initialized 复位 False", plug._initialized is False)
|
||||
check("cleanup 后 plugin._initialized 复位 False", plug.plugin._initialized is False)
|
||||
from plugins.example.global_var import db as gv_db3
|
||||
check("cleanup 后 db 为 None", gv_db3 is None)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user