modify: rust 插件

This commit is contained in:
2026-08-31 16:30:15 +08:00
parent a7f4b1d58b
commit f645c9df65
3 changed files with 791 additions and 13 deletions
+1 -1
View File
@@ -4,7 +4,7 @@ requires = ["hatchling"]
[project] [project]
name = "chongming-cli" name = "chongming-cli"
version = "0.2.2" version = "0.2.4"
description = "chongming 命令脚本集合:构建打包 / 进程插件打包 / 插件脚手架 / 模型生成" description = "chongming 命令脚本集合:构建打包 / 进程插件打包 / 插件脚手架 / 模型生成"
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
+9 -5
View File
@@ -20,7 +20,8 @@
└── start.sh 容器入口: uv sync + uv run chongming-worker --type <name> └── start.sh 容器入口: uv sync + uv run chongming-worker --type <name>
说明: 说明:
- 打包自动排除 __pycache__ / *.pyc / .venv / *.egg-info / .git - 打包自动排除运行时/构建产物: __pycache__ / *.pyc / .venv / *.egg-info / .git /
cache_data(diskcache) / data(worker data_dir) / dist(构建输出) / logs / *.log
- start.sh 权限强制为 0o755(防止 git 丢执行位导致容器无法启动) - start.sh 权限强制为 0o755(防止 git 丢执行位导致容器无法启动)
- 新项目骨架用 `uv run gen_plugin <name> --process` 生成 - 新项目骨架用 `uv run gen_plugin <name> --process` 生成
""" """
@@ -36,12 +37,14 @@ from ._root import find_project_root
NAME_RE = re.compile(r"^[a-z][a-z0-9_]*$") NAME_RE = re.compile(r"^[a-z][a-z0-9_]*$")
# 打包排除的路径模式(目录名 / glob) # 打包排除的路径模式(目录名 / glob)
EXCLUDE_DIRS = {"__pycache__", ".venv", ".git"} # - 运行时/构建产物:cache_data(diskcache)/ data(worker data_dir)/ dist(构建输出)/ logs
EXCLUDE_SUFFIXES = (".pyc", ".pyo", ".egg-info") EXCLUDE_DIRS = {"__pycache__", ".venv", ".git", "cache_data", "data", "dist", "logs", ".pytest_cache"}
EXCLUDE_SUFFIXES = (".pyc", ".pyo", ".egg-info", ".log")
# 项目根目录(CHONGMING_ROOT 环境变量 > CWD 向上查找,见 _root.py) # 项目根目录(CHONGMING_ROOT 环境变量 > CWD 向上查找,见 _root.py)
ROOT = find_project_root() ROOT = find_project_root()
WORKERS_DIR = ROOT / "workers" # WORKERS_DIR = ROOT / "workers"
WORKERS_DIR = ROOT
def _rel(path: Path) -> Path | str: def _rel(path: Path) -> Path | str:
@@ -119,7 +122,8 @@ def build_plugin_package(name: str, output_dir: str | Path = "dist") -> Path:
if not NAME_RE.match(name): if not NAME_RE.match(name):
raise ValueError(f"插件名不合法: '{name}'(需小写字母/数字/下划线,且以字母开头)") raise ValueError(f"插件名不合法: '{name}'(需小写字母/数字/下划线,且以字母开头)")
project_dir = WORKERS_DIR / name # project_dir = WORKERS_DIR / name
project_dir = WORKERS_DIR
if not project_dir.is_dir(): if not project_dir.is_dir():
raise FileNotFoundError(f"未找到插件项目目录: {_rel(project_dir)}/") raise FileNotFoundError(f"未找到插件项目目录: {_rel(project_dir)}/")
+781 -7
View File
@@ -2,7 +2,7 @@
插件脚手架生成器 — 生成风格统一的 Worker 插件模板 插件脚手架生成器 — 生成风格统一的 Worker 插件模板
用法: 用法:
uv run gen_plugin <name> [operation ...] [--process] uv run gen_plugin <name> [operation ...] [--process] [--rust]
示例: 示例:
# 生成 inline 插件骨架(默认,代码在 src/plugins/,配置在 config/workers/): # 生成 inline 插件骨架(默认,代码在 src/plugins/,配置在 config/workers/):
@@ -11,6 +11,9 @@
# 生成进程插件完整项目(自包含,代码 + 配置 + 依赖清单 + 入口脚本): # 生成进程插件完整项目(自包含,代码 + 配置 + 依赖清单 + 入口脚本):
uv run gen_plugin myplugin --process uv run gen_plugin myplugin --process
# 生成 Rust 进程插件完整项目(自包含,配置在插件根,build.sh 打包):
uv run gen_plugin myplugin --rust
# 指定初始操作: # 指定初始操作:
uv run gen_plugin myplugin analyze count --process uv run gen_plugin myplugin analyze count --process
@@ -27,6 +30,23 @@
workers/<name>/pyproject.toml uv 依赖清单(chongming-worker + orjson) workers/<name>/pyproject.toml uv 依赖清单(chongming-worker + orjson)
workers/<name>/start.sh 入口脚本(uv sync + chongming-worker) workers/<name>/start.sh 入口脚本(uv sync + chongming-worker)
生成内容(--rust): workers/<name>/ 自包含 Rust 项目根(= 可打包 zip 的镜像,工程化分层,
结构对位 workers/rsdemo):
workers/<name>/Cargo.toml crate(path 依赖 chongming-worker-rs;可选 redis-cache / database feature)
workers/<name>/src/lib.rs 库层:声明各模块(args/locator/util/app/handlers/generated_models)
workers/<name>/src/main.rs 入口(薄):CLI 解析 + 日志 + runtime + App::init + register + serve + dispose
workers/<name>/src/args.rs clap 命令行参数(--type/--count/--config/--config_dir/--instance)
workers/<name>/src/locator.rs 插件根自定位 plugin_root() + 配置路径解析(Locator)
workers/<name>/src/util.rs 通用小工具 now_ts / deser(含单测)
workers/<name>/src/app.rs 装配 App:数据库生命周期 App::init/dispose + 端点注册 + 模型指纹校验
workers/<name>/src/generated_models.rs 模型(gen_models 生成,指纹对齐端点 TOML)
workers/<name>/src/handlers/<op>.rs 每个端点的业务实现(每端点一文件)
workers/<name>/config/workers/<name>/*.toml 插件注册 + 端点定义(事实来源)
workers/<name>/config/zenoh_config.json5 Zenoh 会话配置(自仓库根复制)
workers/<name>/config.toml 最小根配置(仅 [zenoh] 节)
workers/<name>/build.sh 一键打包(构建 release + 打包插件根含配置)
workers/<name>/README.md / .gitignore
生成内容(默认 inline): 生成内容(默认 inline):
config/workers/<name>/config.toml Worker 注册 + 运行时配置 config/workers/<name>/config.toml Worker 注册 + 运行时配置
config/workers/<name>/<op>.toml 每个操作的 RPC 端点定义 config/workers/<name>/<op>.toml 每个操作的 RPC 端点定义
@@ -39,8 +59,10 @@
import argparse import argparse
import re import re
import shutil import shutil
import subprocess
import sys import sys
from pathlib import Path from pathlib import Path
from string import Template
from ._root import find_project_root from ._root import find_project_root
@@ -81,6 +103,22 @@ count = 1
# 经 Zenoh 注册路由,网关动态挂载;与 src/plugins/ 下的 inline 插件可同时存在 # 经 Zenoh 注册路由,网关动态挂载;与 src/plugins/ 下的 inline 插件可同时存在
load_mode = "external" load_mode = "external"
[workers.cache]
backend = "diskcache" # "diskcache"(默认) | "redis"
cache_store_path = "./cache_data"
# 启动时该类 worker 的全局副本数优先来源(指 config/workers/<type>/config.toml 的 count):
# "toml" (默认) = 经典模式:启动时用 toml 的 count 覆盖 cache 并写回为新的全局总量
# "cache" = 全局模式:chongming-cache 为单一事实来源,toml 仅在 cache 缺失时引导
launch_worker_count = "toml"
# redis 后端专用(backend = "redis" 时生效,均有默认值)
# redis_host = "127.0.0.1"
# redis_port = 6379
# redis_db = 0
# redis_password = ""
# redis_prefix = "chongming:"
[worker] [worker]
name = "{name}" name = "{name}"
version = "0.1.0" version = "0.1.0"
@@ -279,10 +317,13 @@ def _gen_pyproject_toml(name: str) -> str:
uv sync 会跳过其入口点;worker 进程入口由 SDK 的 chongming-worker 脚本提供。 uv sync 会跳过其入口点;worker 进程入口由 SDK 的 chongming-worker 脚本提供。
""" """
deps = [ deps = [
"chongming-cache",
"chongming-worker", "chongming-worker",
] ]
deps_dev = [
"chongming-cli",
]
deps_lines = "\n".join(f' "{d}",' for d in deps) deps_lines = "\n".join(f' "{d}",' for d in deps)
deps_dev_lines = "\n".join(f' "{d}",' for d in deps_dev)
return f"""# {name} 进程插件依赖清单(扩展依赖直接在此追加,如 "minio") return f"""# {name} 进程插件依赖清单(扩展依赖直接在此追加,如 "minio")
[project] [project]
name = "{name}-worker" name = "{name}-worker"
@@ -301,7 +342,7 @@ url = "{PYPI_MIRROR}"
[dependency-groups] [dependency-groups]
dev = [ dev = [
"chongming-cli", {deps_dev_lines}
] ]
""" """
@@ -350,9 +391,10 @@ uv sync # 首次:安装依赖(项目独立 venv,与容
uv run chongming-worker --type {name} uv run chongming-worker --type {name}
``` ```
## 打包(在仓库根执行) ## 打包(在插件根目录执行)
```bash ```bash
cd workers/{name}
uv run build_plugin {name} # → dist/{name}.zip(= 本目录的直接压缩) uv run build_plugin {name} # → dist/{name}.zip(= 本目录的直接压缩)
``` ```
@@ -360,6 +402,648 @@ uv run build_plugin {name} # → dist/{name}.zip(= 本目录的直接压缩
""" """
# ============================================================
# Rust 进程插件模板(--rust,基于 utils/worker-rs SDK + workers/rsdemo 参考实现)
# ============================================================
def _gen_rust_config_toml(name: str) -> str:
"""插件注册 + [workers] 配置([workers.cache] 默认注释禁用)。"""
return f'''# ====== 插件注册信息(Rust 进程插件:独立进程运行) ======
[workers]
type = "{name}"
module = "{name}"
factory = "create_workers"
cleanup = "cleanup"
count = 1
# 进程插件:由 Docker/K8s 编排拉起独立进程(容器入口 ./start.sh / 二进制 --type {name}),
# 经 Zenoh 注册路由;load_mode = "external" 必须保留,网关跳过进程内加载
load_mode = "external"
[worker]
name = "{name}"
version = "0.1.0"
description = "{name} 进程插件(Rust)— 独立进程运行,经 Zenoh 注册路由"
config_version = "1.0"
[worker.defaults]
timeout = 30.0
cacheable = false
auth_required = false
internal = false
method = "GET"
[worker.registration]
router_prefix = "/{name}"
tags = ["{name}", "process-plugin", "rust"]
# ====== Worker 运行时配置(传给 create_workers) ======
[workers.config]
greeting_prefix = "{name}"
# ====== 数据库(可选,对齐 Python chongming_database) ======
# 取消注释并编译开启 database feature 后可用(见 build.sh / Cargo.toml [features]):
# SQLite(type = "sqlite",database_path)或 PostgreSQL(type = "postgresql",url)。
# 未配置 database 段时插件不建库(对齐 Python `db_name = None`)。
# [workers.config.database]
# type = "sqlite"
# database_path = "./data/{name}.db"
# # postgresql 用 url: type = "postgresql"; url = "postgres://user:pwd@127.0.0.1:5432/db"
# ====== 共享缓存(多实例收敛,可选) ======
# 未配置 / backend 缺失时退化为"无缓存"(单实例语义):副本数与 [workers.config]
# 直接读写本地 config.toml。多实例部署**必须**启用 Redis 共享缓存,使副本数与配置
# 跨实例单一事实来源收敛。启用后需以 --features redis-cache 编译(见 build.sh)。
# [workers.cache]
# backend = "redis" # 仅 "redis" 支持跨实例
# url = "redis://127.0.0.1:6379/0"
# # 也可用分段配置(与 url 二选一):
# # host = "127.0.0.1"
# # port = 6379
# # db = 0
# launch_worker_count = "cache"
'''
def _gen_rust_operation_toml(name: str, operation: str) -> str:
"""端点 TOML:声明请求参数与响应模型(含 worker_id/timestamp,模板 handler 依此填充)。"""
return f'''# {name}.{operation} — {operation} 操作
[GET]
summary = "{operation} 操作描述"
params = ["value: str"]
[GET.response_model]
result = {{ type = "str", required = true }}
worker_id = {{ type = "str", required = false, default = "" }}
timestamp = {{ type = "float", required = true }}
'''
def _gen_rust_handler_op(operation: str, camel_op: str, name: str) -> str:
"""单个端点的 handler 模块(src/handlers/<op>.rs):反序列化 → 组装 → Reply::Json。"""
return f'''//! `{operation}` 端点:GET。
use chongming_worker_rs::prelude::*;
use crate::generated_models::{name} as models;
use crate::util::{{deser, now_ts}};
pub async fn {operation}(request: Request) -> Result<Reply, String> {{
let req: models::{camel_op}GetRequest = deser(&request.data)?;
let resp = models::{camel_op}GetResponse {{
result: format!("hello: {{}}", req.value),
worker_id: Some(request.worker_id),
timestamp: now_ts(),
}};
Ok(Reply::Json(serde_json::to_value(&resp).map_err(|e| e.to_string())?))
}}
'''
def _gen_rust_handlers_mod(operations: list[str]) -> str:
mods = "\n".join(f"pub mod {op};" for op in operations)
return f"""//! 业务端点实现:每端点一文件,纯函数风格(依赖经参数注入,便于复用/单测)。
{mods}
"""
def _gen_rust_lib_rs() -> str:
return '''//! # chongming 进程插件(Rust)— 库层
//!
//! 按 Rust 工程惯例组织:`lib.rs` 暴露各模块并承载可测逻辑,`main.rs` 仅做装配
//! (CLI 解析 + 日志 + runtime + serve)。路由描述符/线协议见 docs/plugin-wire-protocol.md。
//!
//! 模块职责:
//! - `args`:命令行参数(clap derive)
//! - `locator`:插件根定位 + 配置路径解析(配置自包含在插件根)
//! - `util`:通用小工具(时间戳 / 参数解析)
//! - `app`:插件装配(端点注册 + 模型指纹校验)
//! - `handlers`:业务端点实现(每端点一文件)
//! - `generated_models`:由端点 TOML 生成的请求/响应模型(勿手动修改)
//!
//! 模型一致性:端点 TOML 是唯一事实来源,改 TOML 后须重新运行 gen_models——启动时
//! [`app::check_models_fingerprint`] 会校验指纹并告警。
#![allow(dead_code)]
pub mod app;
pub mod args;
pub mod generated_models;
pub mod handlers;
pub mod locator;
pub mod util;
'''
def _gen_rust_args_rs(name: str) -> str:
return f'''//! 命令行参数(clap derive)。
use clap::Parser;
/// chongming 进程插件 Rust 实现
#[derive(Parser, Debug)]
#[command(name = "{name}", about = "chongming 进程插件 Rust 实现")]
pub struct Args {{
/// 插件类型(与 config/workers/<type>/ 目录对应)
#[arg(long, default_value = "{name}")]
pub r#type: String,
/// 副本数(覆盖插件配置中的 count)
#[arg(long)]
pub count: Option<usize>,
/// 全局配置文件路径(读取 [zenoh] 节;缺省 = 插件根/config.toml)
#[arg(long)]
pub config: Option<String>,
/// 插件配置目录(config/workers 的上层;缺省 = 插件根/config/workers)
#[arg(long)]
pub config_dir: Option<String>,
/// 实例 id(多实例全局分摊用;缺省 {{hostname}}:{{pid}})
#[arg(long)]
pub instance: Option<String>,
}}
'''
_RUST_LOCATOR_RS = '''//! 插件根定位与配置文件路径解析。
//!
//! 配置**自包含在插件根**(与 Python 进程插件一致):`plugin_root()` 从可执行文件
//! 所在目录向上找第一个含 `config.toml` 的目录。因此从任意 CWD(含容器部署)运行,
//! worker 都能命中自身 `config.toml` + `config/`,打包 zip 解压即运行。
use std::path::PathBuf;
/// 定位插件根目录:从可执行文件所在目录向上,找第一个含 `config.toml` 的目录。
/// 这样开发(target/{profile}/{type})与部署(binary 与 config.toml 同目录)都能命中插件根;
/// 找不到时回退到可执行文件所在目录。
pub fn plugin_root() -> PathBuf {
let exe_dir = std::env::current_exe()
.ok()
.and_then(|p| p.parent().map(|q| q.to_path_buf()))
.unwrap_or_else(|| PathBuf::from("."));
let mut dir = exe_dir.clone();
loop {
if dir.join("config.toml").is_file() {
return dir;
}
if !dir.pop() {
break;
}
}
exe_dir
}
/// 解析出的插件路径(调用方据此构建 `ServeOptions`)。
pub struct Locator {
/// 插件根目录
pub root: PathBuf,
/// 全局配置 `config.toml` 路径(读取 [zenoh] 节)
pub config_path: String,
/// 插件配置目录(config/workers 的上层)
pub config_dir: String,
}
impl Locator {
/// 用 CLI 中可选的 `--config` / `--config_dir` 覆盖插件根默认路径。
pub fn from_cli(config: Option<String>, config_dir: Option<String>) -> Self {
let root = plugin_root();
let config_path =
config.unwrap_or_else(|| root.join("config.toml").to_string_lossy().into_owned());
let config_dir = config_dir.unwrap_or_else(|| {
root.join("config").join("workers").to_string_lossy().into_owned()
});
Self {
root,
config_path,
config_dir,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn default_paths_point_inside_plugin_root() {
let loc = Locator::from_cli(None, None);
assert!(loc.config_path.ends_with("config.toml"));
assert!(loc.config_dir.ends_with("config/workers"));
}
}
'''
_RUST_UTIL_RS = '''//! 通用小工具:时间戳、请求参数解析。
use std::time::{SystemTime, UNIX_EPOCH};
use serde_json::Value;
/// 当前 Unix 时间戳(秒,浮点),对齐网关 timestamp 字段语义。
pub fn now_ts() -> f64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs_f64())
.unwrap_or(0.0)
}
/// 从请求端信封 `data` 反序列化生成模型(强类型请求)。失败返回清晰错误。
pub fn deser<T: serde::de::DeserializeOwned>(data: &Value) -> Result<T, String> {
serde_json::from_value(data.clone()).map_err(|e| format!("参数解析失败: {e}"))
}
#[cfg(test)]
mod tests {
use super::*;
use serde::Deserialize;
#[derive(Deserialize, PartialEq, Debug)]
struct Req {
name: String,
count: i64,
}
#[test]
fn deser_maps_json_to_typed_request() {
let v = serde_json::json!({"name": "chongming", "count": 2});
assert_eq!(
deser::<Req>(&v).ok().unwrap(),
Req { name: "chongming".into(), count: 2 }
);
}
#[test]
fn deser_reports_missing_field() {
let v = serde_json::json!({"name": "chongming"}); // 缺 count
assert!(deser::<Req>(&v).is_err());
}
#[test]
fn now_ts_is_positive() {
assert!(now_ts() > 1_700_000_000.0);
}
}
'''
_RUST_APP_TPL = Template('''//! 插件装配:数据库生命周期、端点注册、模型指纹校验。
//!
//! [`App`] 持有插件级共享资源(如数据库连接池),把各 handler 注册进
//! `HandlerRegistry`,并在退出时释放资源。`main.rs` 只负责 CLI 解析、日志、runtime
//! 与调用本模块。
#[cfg(feature = "database")]
use std::sync::Arc;
#[cfg(feature = "database")]
use chongming_worker_rs::database::{init_database, Database};
use chongming_worker_rs::gen;
use chongming_worker_rs::HandlerRegistry;
use crate::generated_models::${type} as models;
use crate::handlers;
/// 建表 SQL(SQLite 方言;database feature + config 声明 [workers.config].database 时生效)。
#[cfg(feature = "database")]
pub const CREATE_TABLE_SQL: &str =
"CREATE TABLE IF NOT EXISTS kv (k TEXT PRIMARY KEY, v TEXT NOT NULL)";
/// 插件运行时上下文:持有可共享资源,供 handler 与装配使用。
pub struct App {
worker_type: String,
#[cfg(feature = "database")]
db: Option<Arc<Database>>,
}
impl App {
/// 按插件类型与配置目录初始化(含可选数据库)。未编译 database feature 或
/// `[workers.config].database` 未配置时 `db = None`,插件降级为不建库
/// (对齐 Python `db_name = None`)。
pub async fn init(worker_type: &str, config_dir: &str) -> Result<Self, String> {
#[cfg(feature = "database")]
{
let db = init_database(worker_type, config_dir, Some(CREATE_TABLE_SQL))
.await?
.map(Arc::new);
return Ok(Self {
worker_type: worker_type.into(),
db,
});
}
#[cfg(not(feature = "database"))]
{
let _ = config_dir;
Ok(Self { worker_type: worker_type.into() })
}
}
/// 把全部端点注册进 registry。subject 按 `<type>.<op>` 与端点 TOML 对齐。
/// 依赖共享资源(如数据库连接池)的端点用闭包捕获 `Arc<T>` 注入。
pub fn register(&self, registry: &mut HandlerRegistry) {
let subject = |op: &str| format!("{}.{}", self.worker_type, op);
${registrations}
}
/// 优雅关闭持有的共享资源(数据库连接池),对齐 Python cleanup 里的 dispose。
pub async fn dispose(&self) {
#[cfg(feature = "database")]
if let Some(db) = &self.db {
db.dispose().await;
}
}
}
/// 模型一致性校验:端点 TOML 是唯一事实来源,指纹不匹配时告警并提示重新生成。
pub fn check_models_fingerprint(worker_type: &str, config_dir: &str) {
let fp = match gen::fingerprint(worker_type, config_dir) {
Ok(f) => f,
Err(e) => {
tracing::warn!("[{worker_type}] 计算模型指纹失败(跳过一致性校验): {e}");
return;
}
};
if fp != models::FINGERPRINT {
tracing::warn!(
"[{worker_type}] 模型指纹不匹配:端点 TOML 已变更但未重新生成模型。\\
运行: cargo run --manifest-path utils/worker-rs/Cargo.toml --bin gen_models -- \\
--types {worker_type} --config-dir workers/${type}/config/workers \\
--out workers/${type}/src/generated_models.rs"
);
} else {
tracing::info!("[{worker_type}] 模型一致性校验通过(TOML ↔ 生成模型)");
}
}
''')
_RUST_MAIN_TPL = Template('''//! ${type} 入口:只做装配——CLI 解析、日志初始化、构建 tokio runtime、调用库层
//! (`App::init` / `register` / `serve` / `dispose`)。业务与结构逻辑见各子模块(src/lib.rs)。
//!
//! 运行(任意 CWD 均可,self-locating config):
//! cargo run --manifest-path workers/${type}/Cargo.toml -- --type ${type}
use std::sync::Arc;
use chongming_worker_rs::{HandlerRegistry, ServeOptions};
use clap::Parser;
use tracing_subscriber::EnvFilter;
use ${type}::app;
use ${type}::args::Args;
use ${type}::locator::Locator;
fn main() -> Result<(), String> {
let args = Args::parse();
tracing_subscriber::fmt()
.with_env_filter(
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
)
.init();
// 解析插件根与配置文件(支持 --config / --config_dir 覆盖)
let loc = Locator::from_cli(args.config, args.config_dir);
// 模型一致性校验:端点 TOML 变更后须重新运行 gen_models
app::check_models_fingerprint(&args.r#type, &loc.config_dir);
let opts = ServeOptions {
config_path: loc.config_path,
config_dir: loc.config_dir.clone(),
count: args.count,
instance: args.instance,
};
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.map_err(|e| format!("创建 tokio runtime 失败: {e}"))?
.block_on(async {
// 装配:初始化共享资源(数据库连接池等,可选)
let app = app::App::init(&args.r#type, &loc.config_dir).await?;
// 注册全部端点,随后进入 serve——阻塞直至 SIGTERM/SIGINT
let mut registry = HandlerRegistry::new();
app.register(&mut registry);
let result =
chongming_worker_rs::serve::serve(&args.r#type, Arc::new(registry), opts).await;
// 优雅退出后释放共享资源(对齐 Python cleanup 里的 dispose)
app.dispose().await;
result
})
}
''')
def _gen_rust_main_rs(name: str, operations: list[str]) -> str:
return _RUST_MAIN_TPL.substitute(type=name)
def _gen_rust_app_rs(name: str, operations: list[str]) -> str:
registrations = "\n".join(
f' registry.register(&subject("{op}"), "GET", handlers::{op}::{op});'
for op in operations
)
return _RUST_APP_TPL.substitute(type=name, registrations=registrations)
def _gen_rust_cargo_toml(name: str) -> str:
return '''[package]
name = "''' + name + '''"
version = "0.1.0"
edition = "2021"
description = "chongming 进程插件 Rust 实现(由 gen_plugin --rust 生成)"
[dependencies]
chongming-worker-rs = { path = "../../utils/worker-rs" }
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
clap = { version = "4", features = ["derive"] }
tracing = "0.1"
tracing-subscriber = "0.3"
# 生成流式端点时取消下方注释(SSE 逐块多回复需要 async-stream)
# async-stream = "0.3"
# 数据库后端(sqlite/postgres,sqlx;由 database feature 开启,与 SDK 同版本同 feature)
sqlx = { version = "0.8", default-features = false, features = [
"runtime-tokio",
"sqlite",
"postgres",
"any",
], optional = true }
[features]
# 启用共享缓存(Redis 后端,多实例全局收敛配置/副本数)。
# 打包时: bash build.sh dist "redis-cache"
redis-cache = ["chongming-worker-rs/redis-cache"]
# 启用数据库(sqlx,对齐 Python chongming_database)。
# 同时配置 config.toml 的 [workers.config.database] 段,并开启则生效。
# 打包时: bash build.sh dist "database"
database = ["chongming-worker-rs/database", "sqlx"]
'''
def _gen_rust_readme(name: str) -> str:
return f"""# {name} 进程插件(Rust)
独立进程运行的 Worker 插件(`load_mode = "external"`),与网关零导入关系、仅经 Zenoh 通讯。
配置自包含在插件根(与 Python 进程插件一致),打包 zip 解压即运行。
## 目录结构(配置自包含,工程化分层)
```text
workers/{name}/
├── config.toml 最小根配置(仅 [zenoh].json5_path)
├── config/
│ ├── zenoh_config.json5 Zenoh 会话配置
│ └── workers/{name}/ 插件注册 + 端点配置(config.toml + <op>.toml)
├── Cargo.toml 独立 crate(path 依赖 chongming-worker-rs;可选 redis-cache / database feature)
├── src/
│ ├── lib.rs 库层:声明各模块(args/locator/util/app/handlers/generated_models)
│ ├── main.rs 入口(薄):CLI 解析 + 日志 + runtime + App::init + register + serve + dispose
│ ├── args.rs clap 命令行参数(--type/--count/--config/--config_dir/--instance)
│ ├── locator.rs 插件根自定位 `plugin_root()` + 配置路径解析(Locator)
│ ├── util.rs 通用小工具:`now_ts` / `deser`(含单测)
│ ├── app.rs 装配 `App`:数据库生命周期(App::init/dispose)+ 端点注册 + 模型指纹校验
│ ├── generated_models.rs 模型(gen_models 生成,勿手动修改)
│ └── handlers/
│ └── <op>.rs 每个端点的业务实现(每端点一文件)
├── build.sh 一键打包(构建 release + 打包插件根,配置一并带上)
└── README.md
```
## 开发
```bash
# 在插件根目录
cargo run --bin {name} -- --type {name} # 本地调试(任意 CWD,self-locating config)
# 端点 TOML 变更后重新生成模型(在仓库根执行)
cargo run --manifest-path utils/worker-rs/Cargo.toml --bin gen_models -- \\
--types {name} --config-dir workers/{name}/config/workers \\
--out workers/{name}/src/generated_models.rs
```
新增端点流程:`config/workers/{name}/<op>.toml`(事件来源)→ `src/handlers/<op>.rs`
(实现)→ `src/app.rs` 的 `App::register` 注册 → 重新运行上方 gen_models。
## 数据库(可选,对齐 Python `chongming_database`)
需开启 `database` feature 编译(`bash build.sh dist "database"`)并在
`config.toml` 的 `[workers.config.database]` 段配置后端;未配置时插件不建库
(对齐 Python `db_name = None`)。`App::init` 已包含数据库生命周期
(`init_database` + 建表 + `dispose`),handler 用闭包捕获 `Arc<Database>` 注入即可。
## 打包(在插件根目录执行)
```bash
cd workers/{name}
bash build.sh # → dist/{name}.zip(= 本目录的直接压缩,含 config)
bash build.sh dist "redis-cache" # 需要多实例共享缓存时
bash build.sh dist "database" # 需要数据库(sqlx)时(配合 config.toml 的 [workers.config.database])
```
产物 zip 根即容器工作目录:`{name}`(release 二进制)+ `config.toml` + `config/` +
`start.sh`,无需 Rust / Python 运行时,解压即运行(`./start.sh`)。
详见 docs/rust-plugin.md。
"""
def _gen_rust_gitignore() -> str:
return """/target/
/dist/
*.zip
cache_data/
data/
logs/
"""
_RUST_BUILD_SH = """#!/usr/bin/env bash
# Rust 进程插件打包器(自包含于插件根,与 Python `build_plugin` 对位)
#
# 用法(在插件根目录执行):
# bash build.sh # 构建 release + 打包 -> dist/<name>.zip
# bash build.sh /tmp/pkg # 指定输出目录
# bash build.sh dist "redis-cache" # 启用共享缓存(Redis 后端)编译
#
# 产物 zip 根即容器工作目录:含 release 二进制 + config.toml + config/
# (Zenoh 会话 + 端点 TOML)+ start.sh,配置随包一并带上,解压即运行(入口 ./start.sh)。
# 排除 target(构建产物)与本地缓存目录。
set -euo pipefail
NAME="$(basename "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)")"
OUTDIR="${1:-dist}"
CARGO_FEATURES="${2:-}"
PLUGIN_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# ---- 1. 构建 release 二进制 -------------------------------------------
if [ -n "$CARGO_FEATURES" ]; then
echo "▶ cargo build --release --features $CARGO_FEATURES"
cargo build --release --features "$CARGO_FEATURES" --manifest-path "$PLUGIN_DIR/Cargo.toml"
else
echo "▶ cargo build --release"
cargo build --release --manifest-path "$PLUGIN_DIR/Cargo.toml"
fi
BIN="$PLUGIN_DIR/target/release/$NAME"
test -x "$BIN" || { echo "❌ release 二进制不存在: $BIN"; exit 1; }
# ---- 2. 校验插件根「配置自包含」----------------------------------------
for f in "config.toml" "config/zenoh_config.json5" "config/workers/$NAME/config.toml"; do
test -f "$PLUGIN_DIR/$f" || { echo "❌ 插件根缺少 $f(配置需自包含在插件根)"; exit 1; }
done
# ---- 3. 组装打包根(复用容器工作目录布局)-------------------------------
STAGE="$PLUGIN_DIR/target/package/${NAME}_stage"
rm -rf "$STAGE"
mkdir -p "$STAGE/config/workers/$NAME"
cp "$BIN" "$STAGE/$NAME"
cp "$PLUGIN_DIR/config.toml" "$STAGE/config.toml"
cp "$PLUGIN_DIR/config/zenoh_config.json5" "$STAGE/config/zenoh_config.json5"
cp "$PLUGIN_DIR"/config/workers/"$NAME"/*.toml "$STAGE/config/workers/$NAME/"
# start.sh:容器入口(与运行目录无关,二进制自身 self-locating config)
cat > "$STAGE/start.sh" <<EOF
#!/usr/bin/env bash
# $NAME Rust 进程插件容器入口
set -e
cd "\$(dirname "\$0")"
exec ./$NAME --type $NAME "\$@"
EOF
chmod 0755 "$STAGE/start.sh"
# ---- 4. 打包 zip ------------------------------------------------------
OUT="$PLUGIN_DIR/$OUTDIR"
mkdir -p "$OUT"
ZIP="$OUT/$NAME.zip"
rm -f "$ZIP"
# 优先系统 zip,缺失时回退 python3 zipfile(把 stage 根下的内容压进 zip 根)
if command -v zip >/dev/null 2>&1; then
(cd "$STAGE" && zip -rq --symlinks "$ZIP" .)
else
python3 - "$STAGE" "$ZIP" <<'PY'
import os, sys, zipfile
stage, out = sys.argv[1], sys.argv[2]
with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zf:
for root, _dirs, files in os.walk(stage):
for f in files:
src = os.path.join(root, f)
arc = os.path.relpath(src, stage)
zf.write(src, arc)
PY
fi
SIZE=$(du -h "$ZIP" | cut -f1)
rm -rf "$STAGE"
echo "✅ Rust 插件打包完成: $ZIP ($SIZE)"
echo "📦 容器内部署: unzip $NAME.zip -d /app && cd /app && ./start.sh"
"""
# ============================================================ # ============================================================
# 主逻辑 # 主逻辑
# ============================================================ # ============================================================
@@ -391,12 +1075,97 @@ def _gen_models_cmd(config_dir: Path, code_dir: Path) -> str:
return f"{config_dir.relative_to(ROOT)} {code_dir.relative_to(ROOT)}" return f"{config_dir.relative_to(ROOT)} {code_dir.relative_to(ROOT)}"
def generate_plugin(name: str, operations: list[str], process: bool = False) -> int: def _gen_rust_models(name: str, config_dir: Path, out: Path) -> bool:
"""用 SDK 的 gen_models bin 从端点 TOML 生成 Rust 模型(generated_models.rs)。
失败时返回 False(调用方写入空指纹占位并提示)。
"""
cmd = [
"cargo", "run", "--quiet",
"--manifest-path", str(ROOT / "utils" / "worker-rs" / "Cargo.toml"),
"--bin", "gen_models",
"--", "--types", name,
"--config-dir", str(config_dir),
"--out", str(out),
]
r = subprocess.run(cmd, cwd=ROOT, capture_output=True, text=True)
if r.returncode == 0 and out.exists():
print(f"✅ 生成: {out.relative_to(ROOT)}(endpoints → Rust 模型,指纹已对齐)")
return True
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(
f'pub mod {name} {{\n #![allow(dead_code)]\n pub const FINGERPRINT: &str = "";\n}}\n',
encoding="utf-8",
)
print(f"⚠️ gen_models 未执行成功,已生成空指纹占位({out.relative_to(ROOT)})。\n"
f" 请运行: {' '.join(cmd)}\n stderr: {r.stderr.strip()[:200]}")
return False
def _generate_rust_plugin(name: str, operations: list[str]) -> int:
"""生成 Rust 进程插件完整项目(workers/<name>/,自包含 + 可打包,外部插件语义)。"""
project_dir = PROCESS_DIR / name
config_dir = project_dir / "config" / "workers"
src_dir = project_dir / "src"
if project_dir.exists() and src_dir.exists():
print(f"⏭️ 插件已存在: {project_dir.relative_to(ROOT)}/(Rust 暂不支持增量覆盖)")
print(f" • 新增端点: 在 {config_dir.relative_to(ROOT)}/{name}/ 加 <op>.toml,")
print(" 在 src/handlers/ 加 <op>.rs 并在 src/app.rs 的 App::register 注册,")
print(" 然后重新运行下方 gen_models 命令。")
return 0
for op in operations:
_write(config_dir / name / f"{op}.toml", _gen_rust_operation_toml(name, op))
_write(config_dir / name / "config.toml", _gen_rust_config_toml(name))
_write(project_dir / "config.toml", _gen_root_config_toml())
_write(project_dir / "Cargo.toml", _gen_rust_cargo_toml(name))
# 库 + 薄 bin 分层(工程化结构,与 workers/rsdemo 对位)
_write(src_dir / "lib.rs", _gen_rust_lib_rs())
_write(src_dir / "main.rs", _gen_rust_main_rs(name, operations))
_write(src_dir / "args.rs", _gen_rust_args_rs(name))
_write(src_dir / "locator.rs", _RUST_LOCATOR_RS)
_write(src_dir / "util.rs", _RUST_UTIL_RS)
_write(src_dir / "app.rs", _gen_rust_app_rs(name, operations))
_write(src_dir / "handlers" / "mod.rs", _gen_rust_handlers_mod(operations))
for op in operations:
_write(src_dir / "handlers" / f"{op}.rs",
_gen_rust_handler_op(op, snake_to_camel(op), name))
_write(project_dir / "build.sh", _RUST_BUILD_SH)
(project_dir / "build.sh").chmod(0o755)
_write(project_dir / "README.md", _gen_rust_readme(name))
_write(project_dir / ".gitignore", _gen_rust_gitignore())
# Zenoh 会话配置自仓库根复制(后续可自行改为插件专用配置)
if ZENOH_CONFIG.exists():
dest = project_dir / "config" / "zenoh_config.json5"
if not dest.exists():
shutil.copy(ZENOH_CONFIG, dest)
print(f"✅ 生成: {dest.relative_to(ROOT)}(自 {_rel(ZENOH_CONFIG)} 复制)")
else:
print(f"⚠️ 未找到 Zenoh 配置: {_rel(ZENOH_CONFIG)},请自行放入 config/zenoh_config.json5")
# 生成模型 + 指纹(与端点 TOML 对齐,保证启动指纹校验通过)
# 注:gen_models --config_dir 是 config/workers 父目录(内含 <type>/ 子目录)
_gen_rust_models(name, config_dir, src_dir / "generated_models.rs")
print(f"\n📋 Rust 进程插件 '{name}' 已就绪(外部插件,配置自包含在插件根),下一步:")
print(f" 1. cargo build --manifest-path {project_dir.relative_to(ROOT)}/Cargo.toml")
print(f" 2. cd {project_dir.relative_to(ROOT)} && cargo run --bin {name} -- --type {name}")
print(f" 3. 端点 TOML 变更后重新生成模型(见 README)")
print(f" 4. cd {project_dir.relative_to(ROOT)} && bash build.sh # 在插件根目录打包")
return 0
def generate_plugin(name: str, operations: list[str], process: bool = False, rust: bool = False) -> int:
"""生成插件骨架。返回 0 表示成功,1 表示失败。""" """生成插件骨架。返回 0 表示成功,1 表示失败。"""
if not NAME_RE.match(name): if not NAME_RE.match(name):
print(f"❌ 插件名不合法: '{name}'(需小写字母/数字/下划线,且以字母开头)") print(f"❌ 插件名不合法: '{name}'(需小写字母/数字/下划线,且以字母开头)")
return 1 return 1
if rust:
return _generate_rust_plugin(name, operations)
if process: if process:
project_dir = PROCESS_DIR / name project_dir = PROCESS_DIR / name
config_dir = project_dir / "config" / "workers" / name config_dir = project_dir / "config" / "workers" / name
@@ -465,7 +1234,7 @@ def generate_plugin(name: str, operations: list[str], process: bool = False) ->
print(f" 2. 编辑 {code_dir.relative_to(ROOT)}/router/ 实现业务逻辑") print(f" 2. 编辑 {code_dir.relative_to(ROOT)}/router/ 实现业务逻辑")
if process: if process:
print(f" 3. cd {PROCESS_DIR.relative_to(ROOT)}/{name} && ./start.sh 或 uv run chongming-worker --type {name}") print(f" 3. cd {PROCESS_DIR.relative_to(ROOT)}/{name} && ./start.sh 或 uv run chongming-worker --type {name}")
print(f" 4. uv run build_plugin {name}") print(f" 4. cd {PROCESS_DIR.relative_to(ROOT)}/{name} && uv run build_plugin {name} # 在插件根目录打包")
else: else:
print(" 3. uv run serve") print(" 3. uv run serve")
return 0 return 0
@@ -488,10 +1257,15 @@ def main() -> None:
action="store_true", action="store_true",
help="生成进程插件完整项目(workers/<name>/,自包含,可打包部署)", help="生成进程插件完整项目(workers/<name>/,自包含,可打包部署)",
) )
parser.add_argument(
"--rust",
action="store_true",
help="生成 Rust 进程插件完整项目(workers/<name>/,配置自包含,build.sh 打包)",
)
args = parser.parse_args() args = parser.parse_args()
operations = args.operations or ["ping"] operations = args.operations or ["ping"]
sys.exit(generate_plugin(args.name, operations, process=args.process)) sys.exit(generate_plugin(args.name, operations, process=args.process, rust=args.rust))
def gen_plugin(): def gen_plugin():