diff --git a/README.md b/README.md index f62a280..a42d5c9 100644 --- a/README.md +++ b/README.md @@ -1,22 +1,23 @@ # Chongming-Zenoh -基于 **FastAPI + Zenoh + Vue 3** 的**配置驱动微服务网关平台**。 +基于 **FastAPI + Zenoh + MQTT + Vue 3** 的**配置驱动微服务网关平台**。 -核心理念：通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型和 FastAPI 路由， -底层通过 Zenoh 分布式消息协议进行 RPC 通信。Worker 插件可独立开发、热加载、水平扩展。 +核心理念：通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型，同时暴露 **HTTP** 和 **MQTT** 双通道。 +底层通过 Zenoh 分布式消息协议进行 RPC 通信，Worker 插件可独立开发、热加载、水平扩展。 --- ## 核心特性 - **FastAPI + Uvicorn**：高性能异步 Web 框架，自动生成 OpenAPI 文档 -- **Zenoh 服务发现**：基于 Zenoh liveliness 的 Worker 注册与 RPC 调用，支持负载均衡、超时重试 -- **动态路由生成**：根据 `config/workers` 下的 TOML 配置自动生成 API 端点，无需手写路由代码 -- **Worker 框架**：使用 `@rpc_method` 装饰器定义 RPC 方法，自动参数解析与 Pydantic 模型绑定 -- **自动化模型生成**：`gen_models` 脚本根据 TOML 配置自动生成请求/响应 Pydantic 模型 -- **JWT 鉴权**：Rust 实现的 JWT 编码/验证，支持路由级别的 `auth_required` 控制 +- **Zenoh 服务发现**：基于 Zenoh liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 超时重试 +- **MQTT 集成**：内嵌 amqtt Broker，MQTT 消息自动桥接到 Worker 插件，支持请求/响应与单向上报 +- **动态路由生成**：扫描 `config/workers` TOML 配置自动生成 HTTP 路由，`[MQTT]` 节单独处理 +- **Worker 框架**：`@rpc_method` 装饰器定义 RPC 方法，自动参数解析与 Pydantic 模型绑定 +- **自动化模型生成**：`gen_models` 脚本根据 TOML 配置生成请求/响应 Pydantic 模型（含 MQTT） +- **JWT 鉴权**：Rust 实现，路由级别 `auth_required` 控制 - **Argon2id 密码哈希**：Rust 实现的高性能密码安全方案 -- **对象存储代理**：集成 `nanio`（S3 兼容）文件上传/下载 +- **对象存储代理**：集成 nanio（S3 兼容）文件上传/下载 - **构建打包**：支持 PyInstaller / Nuitka 一键打包为独立可执行文件 - **Vue 3 前端**：Element Plus UI、Pinia 状态管理、自动生成类型安全的 API 客户端 @@ -26,41 +27,37 @@ ```text chongming-zenoh/ -├── config.toml # 应用主配置 +├── config.toml # 应用主配置（含 MQTT、Zenoh、DB 等） ├── pyproject.toml # 项目依赖与工具配置 ├── config/ # 配置文件目录 -│ ├── README.md # 配置说明 │ ├── zenoh_config.json5 # Zenoh 会话配置 -│ └── workers/ # Worker 路由定义 -│ ├── config.toml # Worker 插件注册列表 -│ ├── auth/ # 鉴权插件 TOML -│ ├── doubao/ # 豆包插件 TOML -│ └── example/ # 示例插件 TOML +│ └── workers/ # Worker 路由定义（TOML） +│ ├── config.toml # 插件注册列表 +│ ├── auth/ # 鉴权插件（login / register / me / …） +│ ├── doubao/ # 豆包视觉理解插件 +│ └── example/ # 示例插件（add / heartbeat / …） ├── src/ │ ├── chongming/ # 核心应用 │ │ ├── app/ -│ │ │ ├── core/ # WorkerManager、RouteGenerator、鉴权 +│ │ │ ├── core/ # WorkerManager, RouteGenerator, MQTTBridge, auth │ │ │ └── routes/ # 内置路由（存储上传等） -│ │ └── scripts/ # 工具脚本（build, gen_models） +│ │ └── scripts/ # build, gen_models │ ├── chongming-web/ # Vue 3 前端 -│ │ ├── src/ -│ │ │ ├── api/generated/ # 自动生成的 API 客户端（@hey-api/openapi-ts） -│ │ │ ├── views/ # 页面组件 -│ │ │ ├── router/ # 路由配置 -│ │ │ └── stores/ # Pinia 状态管理 -│ │ └── vite.config.ts +│ │ └── src/ +│ │ ├── api/generated/ # 自动生成 API 客户端 +│ │ ├── views/ # 页面组件 +│ │ └── router/ # Vue Router 配置 │ └── plugins/ # Worker 插件源码 -│ ├── auth/ # 鉴权（注册/登录/用户信息/改密） -│ ├── doubao/ # 豆包视觉理解 -│ └── example/ # 示例（四则运算） -├── utils/ # 子工具包（独立 Python 包） -│ ├── config/ # 配置加载 -│ ├── database/ # 异步 SQLite 封装 -│ ├── jwt/ # JWT 编码/验证（Rust） -│ ├── password/ # Argon2id 密码哈希（Rust） -│ ├── request/ # Zenoh 客户端封装 -│ └── worker/ # Worker 框架核心 -├── public/ # 公共启动文件 +│ ├── auth/ # JWT 鉴权 +│ ├── doubao/ # 豆包 AI +│ └── example/ # 示例（含 MQTT 端点） +├── utils/ # 7 个独立工具包 +│ ├── config/ database/ jwt/ lock/ +│ ├── password/ request/ worker/ +├── test_mqtt_e2e.py # MQTT 端到端测试脚本 +├── publisher.py # MQTT 发布示例 +├── subscriber.py # MQTT 订阅示例 +├── test_amqtt.py # amqtt Broker 独立测试 ├── data/storage/chongming/ # 本地数据存储 ├── logs/ # 日志目录 └── resources/ # 模块银行打包资源 @@ -74,32 +71,30 @@ chongming-zenoh/ - Python >= 3.12 - Rust 工具链（用于编译 `jwt` / `password` 扩展） -- Node.js >= 22（用于前端开发） +- Node.js >= 22（前端开发） - [uv](https://docs.astral.sh/uv/) 包管理器 ### 安装与启动 ```bash -# 1. 克隆项目 -git clone -cd chongming-zenoh - -# 2. 安装 Python 依赖（自动编译 Rust 扩展） -uv sync - -# 3. 启动服务（开发模式，含热重载） -uv run serve +uv sync # 安装依赖 + 编译 Rust 扩展 +uv run serve # 启动服务（开发模式，热重载） ``` -访问 API 文档：`http://localhost:8000/docs` +服务启动后可用： + +| 通道 | 地址 | 说明 | +| ------- | ---------------------------- | ------------------------ | +| HTTP | `http://localhost:8000` | REST API + Swagger 文档 | +| MQTT | `mqtt://localhost:1883` | 内嵌 Broker，Worker 桥接 | +| Nanio | `http://localhost:9000` | S3 兼容对象存储 | +| Swagger | `http://localhost:8000/docs` | OpenAPI 文档 | ### 前端开发 ```bash cd src/chongming-web -npm install -npm run dev # 开发服务器 http://localhost:5173 -npm run build # 生产构建 +npm install && npm run dev # http://localhost:5173 ``` --- @@ -107,42 +102,122 @@ npm run build # 生产构建 ## 架构与数据流 ```mermaid -graph LR - A[用户请求] --> B[FastAPI 路由
自动生成] - B --> C{鉴权检查} - C -->|通过| D[WorkerManager.call] - D --> E[Zenoh Query] - E --> F[Worker Queryable] - F --> G[插件业务逻辑] - G --> H[响应] +graph TB + HTTP[HTTP 客户端] -->|REST API| FastAPI[FastAPI 路由
自动生成] + MQTT[MQTT 客户端] -->|chongming/+/+| Bridge[MQTT Bridge
内嵌 Broker] + + FastAPI --> WM[WorkerManager] + Bridge --> WM + + WM -->|Zenoh Query| W1[Worker 副本 1] + WM -->|负载均衡| W2[Worker 副本 2] + WM -->|超时重试| W3[Worker 副本 N] + + W1 & W2 & W3 -->|插件业务逻辑| DB[(SQLite)] + W1 & W2 & W3 -->|AI 调用| AI[豆包 API] ``` 启动流程： ``` -serve() → FastAPI lifespan: +uv run serve → FastAPI lifespan: 1. setup_nanio() — 本地 S3 子进程 2. setup_s3() — boto3 S3 客户端 3. setup_zenoh() — Zenoh peer 会话 - 4. setup_worker_manager() — WorkerManager + liveliness 发现 - 5. load_workers() — 动态 import 插件 - 6. register_routes() — 扫描 config/workers/ 生成路由 + 4. setup_worker_manager() — WorkerManager + liveliness 刷新 + 5. load_workers() — 动态 import 插件 → 启动 Worker 协程 + 6. register_routes() — 扫描 config/workers/ 生成 HTTP 路由 + 7. setup_mqtt() — 内嵌 Broker + MQTT→Worker 桥接 +``` + +--- + +## MQTT 集成 + +### 协议约定 + +| 方向 | Topic | Payload | +| ---- | -------------------------------- | ------------------------------------------ | +| 请求 | `chongming//` | `{"_method":"MQTT","param1":"value1",...}` | +| 响应 | `chongming///response` | 由 Worker 返回（`None` 则跳过） | + +### 快速测试 + +```bash +# 终端 1: 启动服务 +uv run serve + +# 终端 2: 端到端测试 +python test_mqtt_e2e.py + +# 或手动测试 +python subscriber.py # 监听所有响应 +python publisher.py example add '{"a":3,"b":5,"_method":"MQTT"}' # 有响应 +python publisher.py example heartbeat '{"worker_id":"node-1","_method":"MQTT"}' # 无响应 +``` + +### 无响应模式（心跳上报等） + +Worker 返回 `None` 时，MQTT Bridge 自动跳过响应发布： + +```python +@rpc_method(subject="example.heartbeat", method="MQTT") +async def heartbeat(worker_id: str = "") -> None: + return None # Bridge 跳过发布 +``` + +### 配置 + +```toml +# config.toml — 按环境区分 +[development.mqtt] +enabled = true # false 禁用 MQTT +host = "localhost" +port = 1883 + +[development.mqtt.broker] +# 可选：amqtt Broker 高级参数 ``` --- ## 开发指南 +### 端点类型说明 + +TOML 文件中的节类型决定端点暴露方式： + +| 节类型 | HTTP 路由 | OpenAPI | MQTT | 请求模型命名 | 备注 | +| -------- | --------- | ------- | ---- | ------------------------- | ------------------- | +| `[GET]` | ✅ | ✅ | ✅ | `XxxGetRequest/Response` | | +| `[POST]` | ✅ | ✅ | ✅ | `XxxPostRequest/Response` | | +| `[MQTT]` | ❌ | ❌ | ✅ | `XxxMqttRequest` | 无 `response_model` | + +```toml +# config/workers/example/add.toml +[GET] +summary = "Add via HTTP GET" +params = ["a: float", "b: float"] + +[POST] +summary = "Add via HTTP POST" +params = ["a: float", "b: float"] + +[MQTT] +summary = "Add via MQTT" +params = ["a: float", "b: float"] +``` + ### 添加新 Worker 插件 -1. 创建 `src/plugins//` 目录结构 -2. 在 `config/workers/config.toml` 注册插件 +1. 创建 `src/plugins//` — `__init__.py`, `global_var.py`, `router/`, `models/` +2. 注册到 `config/workers/config.toml` 3. 创建 `config/workers//*.toml` 端点定义 -4. 运行 `uv run python -m chongming.scripts.gen_models` 生成模型 -5. 实现 `router/*.py` 业务逻辑 -6. 实现 `__init__.py` 工厂函数 +4. `uv run gen_models config/workers// src/plugins//` +5. 实现 `router/.py`，引用生成的模型 +6. 更新 `router/__init__.py` 导入 -详见 [插件开发指南](src/plugins/README.md) 和 [配置说明](config/README.md)。 +详见 [插件开发指南](src/plugins/README.md)。 ### 常用命令 @@ -156,27 +231,27 @@ uv pip install -e utils/ # 安装/更新工具包子包 ### TOML 配置字段速查 -| 字段 | 位置 | 说明 | -| -------------------- | --------------------- | --------------------------------------------------- | -| `count` | `workers/config.toml` | Worker 副本数 | -| `router_prefix` | `<插件>/config.toml` | API 路由前缀 | -| `defaults.timeout` | `<插件>/config.toml` | 默认超时（秒） | -| `defaults.cacheable` | `<插件>/config.toml` | 是否缓存 RPC 结果 | -| `params` | `<操作>.toml` `[GET]` | GET 参数列表（`"name: type"`） | -| `auth_required` | `<操作>.toml` | 是否需要 JWT 鉴权 | -| `internal` | `<操作>.toml` | 设为 `true` 则仅允许内部 RPC 调用，不注册 HTTP 路由 | -| `cacheable` | `<操作>.toml` | 端点是否可缓存 | -| `summary` | `<操作>.toml` | 端点描述（OpenAPI docs） | +| 字段 | 位置 | 说明 | +| -------------------- | --------------------- | --------------------------------------- | +| `count` | `workers/config.toml` | Worker 副本数 | +| `router_prefix` | `<插件>/config.toml` | API 路由前缀 | +| `defaults.timeout` | `<插件>/config.toml` | 默认超时（秒） | +| `defaults.cacheable` | `<插件>/config.toml` | 是否缓存 RPC 结果 | +| `params` | `<操作>.toml` | 参数列表（`"name: type"` 格式） | +| `auth_required` | `<操作>.toml` | 是否需要 JWT 鉴权 | +| `internal` | `<操作>.toml` | `true` 则仅限内部 RPC，不注册 HTTP 路由 | +| `cacheable` | `<操作>.toml` | 端点是否可缓存 | +| `summary` | `<操作>.toml` | 端点描述（用于 OpenAPI / 日志） | --- ## 内置插件 -| 插件 | 路由前缀 | 说明 | -| ----------- | --------- | -------------------------------------------- | -| **auth** | `/auth` | 用户注册、登录、JWT 鉴权、个人信息、密码修改 | -| **doubao** | `/doubao` | 火山引擎豆包视觉理解模型集成 | -| **example** | `/calc` | 四则运算示例（加/减/乘/除）+ 访问计数器 | +| 插件 | HTTP 前缀 | MQTT 主题前缀 | 说明 | +| ----------- | --------- | ------------- | ------------------------------------ | +| **auth** | `/auth` | `auth.*` | 注册、登录、JWT 鉴权、个人信息、改密 | +| **doubao** | `/doubao` | `doubao.*` | 火山引擎豆包视觉理解模型 | +| **example** | `/calc` | `example.*` | 四则运算 + 访问计数 + MQTT 心跳示例 | --- @@ -185,10 +260,10 @@ uv pip install -e utils/ # 安装/更新工具包子包 | 包名 | 说明 | 技术栈 | | -------------------- | --------------------------------------------- | ----------- | | `chongming-config` | TOML 配置加载、TypedDict 类型定义 | Python | -| `chongming-database` | 异步 SQLite 封装（aiosqlite + SQLModel） | Python | +| `chongming-database` | 异步 SQLite（aiosqlite + SQLModel） | Python | | `chongming-jwt` | JWT 编码/验证（HS256/384/512, RS256/384/512） | Rust + PyO3 | | `chongming-password` | Argon2id 密码哈希与验证 | Rust + PyO3 | -| `chongming-request` | Zenoh Worker 直连客户端（负载均衡、超时重试） | Python | +| `chongming-request` | Zenoh Worker 客户端（负载均衡、超时重试） | Python | | `chongming-worker` | Worker 框架核心（`@rpc_method`、运行时） | Python | --- diff --git a/config.toml b/config.toml index a1ea445..8e602df 100644 --- a/config.toml +++ b/config.toml @@ -50,6 +50,15 @@ allow_headers = ["*"] allow_methods = ["*"] allow_origins = ["*"] +[development.mqtt] +enabled = true +host = "localhost" +port = 1883 + +[development.mqtt.broker] +# amqtt Broker 可选配置，默认允许匿名连接 +# 详细参数参考: https://github.com/Yakifo/amqtt + [plugins_worker] config_path = "./config/workers/config.toml" diff --git a/publisher.py b/publisher.py new file mode 100644 index 0000000..fabf0a8 --- /dev/null +++ b/publisher.py @@ -0,0 +1,47 @@ +""" +MQTT Publisher — 向 chongming Worker 发送 MQTT 请求 + +用法: + python publisher.py [json_payload] + +示例: + python publisher.py example add '{"a": 3, "b": 5, "_method": "GET"}' + python publisher.py doubao analyze '{"text": "hello", "_method": "POST"}' +""" + +import asyncio +import sys +import json +import logging +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_1 + +logging.basicConfig(level=logging.INFO) + +# MQTT 主题前缀（与 mqtt_bridge.py 保持一致） +PREFIX = "chongming" + +# 默认参数 +DEFAULT_WORKER = "example" +DEFAULT_OP = "add" +DEFAULT_PAYLOAD = {"a": 3, "b": 5, "_method": "GET"} + +async def main(): + worker_type = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_WORKER + operation = sys.argv[2] if len(sys.argv) > 2 else DEFAULT_OP + payload_str = sys.argv[3] if len(sys.argv) > 3 else json.dumps(DEFAULT_PAYLOAD) + + topic = f"{PREFIX}/{worker_type}/{operation}" + payload = json.loads(payload_str) + + client = MQTTClient() + await client.connect("mqtt://localhost:1883/") + + await client.publish(topic, json.dumps(payload).encode("utf-8"), QOS_1) + print(f"[Publisher] Sent to '{topic}': {payload}") + + await client.disconnect() + print("[Publisher] Disconnected") + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py index 6b8759f..c28408d 100644 --- a/src/chongming/app/__init__.py +++ b/src/chongming/app/__init__.py @@ -52,7 +52,10 @@ async def lifespan(app: FastAPI): # 6. 注册路由 await lifespan_setup.register_routes(app, manager) - # 7. 短暂等待，确保 worker 状态同步 + # 7. 启动 MQTT Bridge（内嵌 Broker + Worker 桥接） + await lifespan_setup.setup_mqtt(app, manager) + + # 8. 短暂等待，确保 worker 状态同步 await asyncio.sleep(0.5) # 打印 worker 就绪状态 @@ -68,10 +71,13 @@ async def lifespan(app: FastAPI): # ---- 关闭阶段 ---- logger.info("正在关闭应用...") - # 1. 清理 workers（调用 cleanup） + # 1. 停止 MQTT Bridge + await lifespan_setup.shutdown_mqtt(app) + + # 2. 清理 workers（调用 cleanup） await lifespan_setup.cleanup_workers(app) - # 2. 取消刷新任务 + # 3. 取消刷新任务 refresh_task = getattr(app.state, "refresh_task", None) if refresh_task: refresh_task.cancel() @@ -80,17 +86,17 @@ async def lifespan(app: FastAPI): except asyncio.CancelledError: pass - # 3. 取消所有 worker 任务 + # 4. 取消所有 worker 任务 for task in worker_tasks: task.cancel() if worker_tasks: await asyncio.gather(*worker_tasks, return_exceptions=True) - # 4. 关闭 Zenoh 会话 + # 5. 关闭 Zenoh 会话 session.close() logger.info("Zenoh 会话已关闭") - # 5. 终止 nanio + # 6. 终止 nanio await lifespan_setup.shutdown_nanio(app) logger.info("应用已关闭") diff --git a/src/chongming/app/core/config_.py b/src/chongming/app/core/config_.py index 94eeee7..1b35692 100644 --- a/src/chongming/app/core/config_.py +++ b/src/chongming/app/core/config_.py @@ -102,6 +102,15 @@ class Cors(TypedDict, total=False): allow_headers: List[str] +class MQTTConfig(TypedDict, total=False): + """MQTT Bridge 配置（按环境区分）""" + + enabled: bool + host: str + port: int + broker: dict + + class ModuleSystem(TypedDict, total=False): """模块系统配置，定义模块加载类型和路径""" @@ -123,6 +132,7 @@ class Env(TypedDict, total=False): server: Required[Server] logging: Required[Logging] cors: Cors + mqtt: MQTTConfig module_system: ModuleSystem file_system: FileSystem diff --git a/src/chongming/app/core/lifespan_setup.py b/src/chongming/app/core/lifespan_setup.py index 8f30192..1942b43 100644 --- a/src/chongming/app/core/lifespan_setup.py +++ b/src/chongming/app/core/lifespan_setup.py @@ -17,6 +17,7 @@ from chongming_config import get_config, get_workers_config from ..core.logger import get_logger from ..core.route_generator import register_dynamic_routes from ..core.worker_manager import WorkerManager +from ..core.mqtt_bridge import MQTTBridge config = get_config() logger = get_logger("app") @@ -182,6 +183,46 @@ async def register_routes(app: FastAPI, manager: WorkerManager): register_dynamic_routes(app, manager, prefix=prefix) # 可选：等待路由注册生效（已在 register_dynamic_routes 内部完成日志） +async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge | None: + """ + 启动 MQTT Bridge（内嵌 Broker + 客户端桥接）。 + + 从 config.toml 的环境段（如 [development.mqtt]）读取配置， + 默认监听 localhost:1883。若 enabled 为 false，则跳过启动。 + """ + env_config = config.get(env, {}) + mqtt_config = env_config.get("mqtt", {}) + if not mqtt_config.get("enabled", True): + logger.info("MQTT Bridge 已禁用（mqtt.enabled = false）") + return None + + host = mqtt_config.get("host", "localhost") + port = mqtt_config.get("port", 1883) + broker_config = mqtt_config.get("broker", {}) + + bridge = MQTTBridge( + worker_manager=worker_manager, + host=host, + port=port, + broker_config=broker_config, + ) + + app.state.mqtt_bridge = bridge + await bridge.start() + logger.info(f"MQTT Bridge 已启动 (host={host}, port={port})") + return bridge + + +async def shutdown_mqtt(app: FastAPI): + """停止 MQTT Bridge。""" + bridge: MQTTBridge | None = getattr(app.state, "mqtt_bridge", None) + if bridge is None: + return + + if bridge.is_running: + await bridge.stop() + + async def shutdown_nanio(app: FastAPI): """终止 nanio 进程。""" proc = getattr(app.state, "nanio_process", None) diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py new file mode 100644 index 0000000..35113ca --- /dev/null +++ b/src/chongming/app/core/mqtt_bridge.py @@ -0,0 +1,292 @@ +""" +MQTT Bridge — 将 MQTT 消息桥接到 Worker 插件 + +架构: + MQTT 客户端订阅 chongming/+/+ 主题 + → 解析 topic: chongming// + → 调用 WorkerManager.call(worker_type, subject, payload) + → 将响应发布到 chongming///response + +同时内嵌一个 amqtt Broker，无需外部 MQTT 服务器。 +""" + +import asyncio +import json +from typing import Any, Dict, Optional + +from amqtt.broker import Broker +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_1 + +from .logger import get_logger + +logger = get_logger("mqtt") + +# MQTT 主题前缀 +MQTT_TOPIC_PREFIX = "chongming" +# 订阅通配符：匹配 chongming// +MQTT_SUBSCRIBE_TOPIC = f"{MQTT_TOPIC_PREFIX}/+/+" + + +class MQTTBridge: + """ + MQTT 桥接器 + + 内嵌 amqtt Broker + 客户端，将 MQTT 消息转发到 WorkerManager。 + + 用法: + bridge = MQTTBridge(worker_manager, host="localhost", port=1883) + await bridge.start() + # ... 应用运行 ... + await bridge.stop() + """ + + def __init__( + self, + worker_manager, + host: str = "localhost", + port: int = 1883, + broker_config: Optional[Dict[str, Any]] = None, + ): + self._worker_manager = worker_manager + self._host = host + self._port = port + self._broker_config = broker_config or {} + + self._broker: Optional[Broker] = None + self._client: Optional[MQTTClient] = None + self._broker_task: Optional[asyncio.Task] = None + self._listen_task: Optional[asyncio.Task] = None + self._running = False + + # ============================================================ + # 生命周期 + # ============================================================ + + async def start(self) -> None: + """启动 MQTT Broker 和客户端桥接""" + if self._running: + logger.warning("MQTT Bridge 已在运行中") + return + + # 1. 启动内嵌 Broker + await self._start_broker() + + # 2. 启动客户端桥接 + await self._start_client() + + self._running = True + logger.info(f"MQTT Bridge 已启动 (broker: {self._host}:{self._port})") + + async def stop(self) -> None: + """停止客户端和 Broker""" + if not self._running: + return + + logger.info("正在关闭 MQTT Bridge...") + + # 1. 停止客户端 + await self._stop_client() + + # 2. 停止 Broker + await self._stop_broker() + + self._running = False + logger.info("MQTT Bridge 已关闭") + + # ============================================================ + # Broker（内嵌 MQTT 服务器） + # ============================================================ + + async def _start_broker(self) -> None: + """启动内嵌 amqtt Broker""" + config = { + "listeners": { + "default": { + "type": "tcp", + "bind": f"{self._host}:{self._port}", + "max_connections": 500, + }, + }, + "timeout_disconnect_delay": 2, + "sys_interval": 0, # 禁用 sys 主题周期性发布，避免 None > 0 异常 + "auth": { + "allow-anonymous": True, + }, + } + # 合并用户自定义配置 + config.update(self._broker_config) + + self._broker = Broker(config) + + async def _run_broker(): + try: + await self._broker.start() # type: ignore + logger.info(f"amqtt Broker 已启动 @ {self._host}:{self._port}") + # 保持 broker 运行直到被取消 + while True: + await asyncio.sleep(1) + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Broker 运行异常: {e}") + + self._broker_task = asyncio.create_task(_run_broker()) + # 等待 broker 就绪 + await asyncio.sleep(0.5) + + async def _stop_broker(self) -> None: + """停止 Broker""" + if self._broker_task and not self._broker_task.done(): + self._broker_task.cancel() + try: + await self._broker_task + except asyncio.CancelledError: + pass + + if self._broker: + try: + await self._broker.shutdown() + except Exception as e: + logger.error(f"Broker 关闭异常: {e}") + self._broker = None + + logger.info("amqtt Broker 已停止") + + # ============================================================ + # 客户端桥接 + # ============================================================ + + async def _start_client(self) -> None: + """启动 MQTT 客户端并订阅 worker 主题""" + self._client = MQTTClient() + await self._client.connect(f"mqtt://{self._host}:{self._port}/") + await self._client.subscribe([(MQTT_SUBSCRIBE_TOPIC, QOS_1)]) + logger.info(f"MQTT 客户端已订阅: {MQTT_SUBSCRIBE_TOPIC}") + + self._listen_task = asyncio.create_task(self._message_loop()) + + async def _stop_client(self) -> None: + """断开 MQTT 客户端""" + if self._listen_task and not self._listen_task.done(): + self._listen_task.cancel() + try: + await self._listen_task + except asyncio.CancelledError: + pass + + if self._client: + try: + await self._client.unsubscribe([MQTT_SUBSCRIBE_TOPIC]) + await self._client.disconnect() + except Exception as e: + logger.error(f"MQTT 客户端断开异常: {e}") + self._client = None + + logger.info("MQTT 客户端已断开") + + # ============================================================ + # 消息处理 + # ============================================================ + + async def _message_loop(self) -> None: + """消息循环：接收 MQTT 消息并转发到 WorkerManager""" + while True: + try: + message = await self._client.deliver_message() # type: ignore + await self._handle_message(message) + except asyncio.CancelledError: + break + except Exception as e: + logger.error(f"MQTT 消息处理异常: {e}") + + async def _handle_message(self, message) -> None: + """ + 处理收到的 MQTT 消息 + + 解析 topic: chongming// + 转发到 WorkerManager.call() + 发布响应到 chongming///response + """ + topic: str = message.topic # type: ignore + payload_bytes: bytes = message.data # type: ignore + + # 解析主题 + parts = topic.split("/") + if len(parts) < 3: + logger.warning(f"无效的 MQTT 主题: {topic}") + return + + worker_type = parts[1] + operation = parts[2] + subject = f"{worker_type}.{operation}" + response_topic = f"{MQTT_TOPIC_PREFIX}/{worker_type}/{operation}/response" + + # 解析载荷 + try: + payload_str = payload_bytes.decode("utf-8") if payload_bytes else "{}" + payload = json.loads(payload_str) + except json.JSONDecodeError: + logger.warning(f"无效的 JSON 载荷: {payload_bytes}") + await self._publish_response( + response_topic, + {"error": "Invalid JSON payload"}, + ) + return + + # 确保 _method 字段存在 + if "_method" not in payload: + payload["_method"] = "GET" + + logger.info( + f"MQTT → Worker: {topic} [{payload.get('_method', 'GET')}] " + f"-> {worker_type}/{subject}" + ) + + # 调用 WorkerManager + try: + result = await self._worker_manager.call( + worker_type=worker_type, + subject=subject, + payload=payload, + timeout=float(payload.pop("_timeout", 30.0)), + cacheable=bool(payload.pop("_cacheable", False)), + ) + # result 为 None 时不发布响应（如心跳上报等无需返回的场景） + if result is not None: + await self._publish_response(response_topic, result) + logger.debug(f"Worker → MQTT: {response_topic} success") + else: + logger.debug(f"Worker → MQTT: {topic} (无响应，已跳过)") + except RuntimeError as e: + error_resp = {"error": str(e), "worker_type": worker_type, "subject": subject} + await self._publish_response(response_topic, error_resp) + logger.warning(f"Worker 调用失败: {e}") + except Exception as e: + error_resp = {"error": str(e), "worker_type": worker_type, "subject": subject} + await self._publish_response(response_topic, error_resp) + logger.error(f"MQTT 桥接异常: {e}") + + async def _publish_response(self, topic: str, data: Any) -> None: + """将响应发布到 MQTT 主题""" + try: + payload = json.dumps(data, ensure_ascii=False, default=str) + await self._client.publish(topic, payload.encode("utf-8"), QOS_1) # type: ignore + except Exception as e: + logger.error(f"MQTT 发布失败 [{topic}]: {e}") + + # ============================================================ + # 属性 + # ============================================================ + + @property + def is_running(self) -> bool: + return self._running + + @property + def broker_host(self) -> str: + return self._host + + @property + def broker_port(self) -> int: + return self._port diff --git a/src/chongming/app/core/route_generator.py b/src/chongming/app/core/route_generator.py index 65ddbf3..7929851 100644 --- a/src/chongming/app/core/route_generator.py +++ b/src/chongming/app/core/route_generator.py @@ -128,9 +128,11 @@ def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]: response_model = { ... } [POST] ... # 同一个接口支持多个请求方法 + [MQTT] + ... # MQTT 专用端点（不注册 HTTP 路由） 返回: List[Dict], 每个 dict 包含 flatten 后的路由配置, - 每个方法节(如 GET) 生成一个独立的路由条目。 + 每个方法节(如 GET/POST/MQTT) 生成一个独立的路由条目。 """ from pathlib import Path import tomllib @@ -386,6 +388,13 @@ def _register_single_route( """注册单个路由，生成正确的函数签名以便 OpenAPI 正确生成文档""" rc = RouteConfig(route_conf) + # MQTT 端点：不注册 HTTP 路由，仅通过 MQTT Bridge 调用 + if rc.method == "MQTT": + logger.info( + f"MQTT 端点: subject={worker_type}.{rc.subject} (仅 MQTT，不注册 HTTP 路由)" + ) + return + # internal 端点：不注册 HTTP 路由，仅保留 Zenoh RPC 内部调用能力 if rc.internal: logger.info( diff --git a/src/chongming/scripts/gen_models.py b/src/chongming/scripts/gen_models.py index 6ed8261..f639354 100644 --- a/src/chongming/scripts/gen_models.py +++ b/src/chongming/scripts/gen_models.py @@ -91,8 +91,9 @@ def parse_response_model(response_model: dict[str, Any]) -> list[dict[str, Any]] def _model_name(base: str, method: str, suffix: str) -> str: - """生成模型名: e.g. add + GET + Request -> AddGetRequest""" - return f"{base.title().replace('-', '').replace('_', '')}{method.title()}{suffix}" + """生成模型名: e.g. add + GET + Request -> AddGetRequest, add + MQTT + Request -> AddMqttRequest""" + method_title = "Mqtt" if method == "MQTT" else method.title() + return f"{base.title().replace('-', '').replace('_', '')}{method_title}{suffix}" def generate_pydantic_code( @@ -243,7 +244,7 @@ def scan_config_dir(config_dir: Path) -> list[dict[str, Any]]: route_name = toml_file.stem - for method in ("GET", "POST", "PUT", "DELETE", "PATCH"): + for method in ("GET", "POST", "PUT", "DELETE", "PATCH", "MQTT"): if method not in data: continue diff --git a/subscriber.py b/subscriber.py new file mode 100644 index 0000000..edb8aac --- /dev/null +++ b/subscriber.py @@ -0,0 +1,64 @@ +""" +MQTT Subscriber — 监听 chongming Worker 的响应 + +用法: + python subscriber.py [worker_type] [operation] + +示例: + python subscriber.py # 监听所有 chongming 响应 + python subscriber.py example add # 只监听 example.add 的响应 +""" + +import asyncio +import json +import sys +import logging +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_1 + +logging.basicConfig(level=logging.INFO) + +PREFIX = "chongming" + +# 默认订阅所有 chongming 响应 +DEFAULT_TOPIC = f"{PREFIX}/+/+/response" + +async def main(): + worker_type = sys.argv[1] if len(sys.argv) > 1 else None + operation = sys.argv[2] if len(sys.argv) > 2 else None + + if worker_type and operation: + topic = f"{PREFIX}/{worker_type}/{operation}/response" + elif worker_type: + topic = f"{PREFIX}/{worker_type}/+/response" + else: + topic = DEFAULT_TOPIC + + client = MQTTClient() + await client.connect("mqtt://localhost:1883/") + + await client.subscribe([(topic, QOS_1)]) + print(f"[Subscriber] Subscribed to '{topic}'. Waiting for messages... (Press Ctrl+C to stop)") + + try: + while True: + message = await client.deliver_message() + payload = message.data.decode("utf-8") # type: ignore + try: + data = json.loads(payload) + print(f"[Subscriber] Topic: {message.topic}") # type: ignore + print(f" Response: {json.dumps(data, indent=2, ensure_ascii=False)}") + except json.JSONDecodeError: + print(f"[Subscriber] Topic: {message.topic}, Raw: {payload}") # type: ignore + except asyncio.CancelledError: + print("[Subscriber] Stopping...") + finally: + await client.unsubscribe([topic]) + await client.disconnect() + print("[Subscriber] Disconnected") + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + print("[Subscriber] Interrupted by user") \ No newline at end of file diff --git a/test_amqtt.py b/test_amqtt.py new file mode 100644 index 0000000..301132b --- /dev/null +++ b/test_amqtt.py @@ -0,0 +1,30 @@ +import asyncio +from asyncio import CancelledError +import logging + +from amqtt.broker import Broker + +""" +This sample shows how run a broker +""" + +formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s" +logging.basicConfig(level=logging.INFO, format=formatter) + +async def run_server() -> None: + broker = Broker() + try: + await broker.start() + while True: + await asyncio.sleep(1) + except CancelledError: + await broker.shutdown() + +def __main__(): + try: + asyncio.run(run_server()) + except KeyboardInterrupt: + print("Server exiting...") + +if __name__ == "__main__": + __main__() \ No newline at end of file diff --git a/test_mqtt_e2e.py b/test_mqtt_e2e.py new file mode 100644 index 0000000..ff5196e --- /dev/null +++ b/test_mqtt_e2e.py @@ -0,0 +1,102 @@ +""" +MQTT 端到端测试脚本 + +测试流程: + 1. 启动内嵌 amqtt Broker + 2. 启动 subscriber（监听响应） + 3. 启动 publisher（发送 MQTT 请求 → chongming/example/add） + 4. 验证 subscriber 收到正确的响应 + +注意: 此脚本独立运行，不依赖 chongming 服务端。 + 使用时需确保 chongming 服务已通过 `uv run serve` 启动。 +""" + +import asyncio +import json +import sys +import logging +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_1 + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(name)s] %(message)s", +) + +PREFIX = "chongming" +WORKER = "example" +OPERATION = "add" +REQUEST_TOPIC = f"{PREFIX}/{WORKER}/{OPERATION}" +RESPONSE_TOPIC = f"{PREFIX}/{WORKER}/{OPERATION}/response" +BROKER_URL = "mqtt://localhost:1883/" + +# 测试用例 +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}, +] + +PASS = 0 +FAIL = 0 + + +async def run_test(): + global PASS, FAIL + client = MQTTClient() + await client.connect(BROKER_URL) + await client.subscribe([(RESPONSE_TOPIC, QOS_1)]) + print(f"📡 已订阅: {RESPONSE_TOPIC}\n") + + for i, case in enumerate(TEST_CASES): + a, b, expected = case["a"], case["b"], case["expected"] + payload = {"a": a, "b": b, "_method": "MQTT"} + payload_bytes = json.dumps(payload).encode("utf-8") + + await client.publish(REQUEST_TOPIC, payload_bytes, QOS_1) + print(f"📤 [{i+1}/{len(TEST_CASES)}] 发送: {REQUEST_TOPIC} → {payload}") + + # 等待响应（超时 5 秒） + try: + message = await asyncio.wait_for(client.deliver_message(), timeout=5.0) + data = json.loads(message.data.decode("utf-8")) # type: ignore + actual = data.get("result") + status = "✅ PASS" if abs(actual - expected) < 0.01 else "❌ FAIL" + if status == "✅ PASS": + PASS += 1 + else: + FAIL += 1 + print(f"📥 [{i+1}/{len(TEST_CASES)}] 响应: {json.dumps(data)} → {status} (expect={expected}, got={actual})") + except asyncio.TimeoutError: + FAIL += 1 + print(f"📥 [{i+1}/{len(TEST_CASES)}] ❌ FAIL: 超时无响应") + + await client.unsubscribe([RESPONSE_TOPIC]) + await client.disconnect() + + print(f"\n{'='*50}") + print(f"测试结果: ✅ {PASS} 通过 ❌ {FAIL} 失败") + print(f"{'='*50}") + return FAIL == 0 + + +async def main(): + # 先检查 Broker 是否可用 + test_client = MQTTClient() + try: + await asyncio.wait_for(test_client.connect(BROKER_URL), timeout=3.0) + await test_client.disconnect() + except Exception: + print("❌ 无法连接到 MQTT Broker (localhost:1883)") + print(" 请先启动 chongming 服务: uv run serve") + sys.exit(1) + + success = await run_test() + sys.exit(0 if success else 1) + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + print("\n⏹️ 测试被中断")