lan/CarrotPrint
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update dependencies and refactor application structure

Browse Source 
- Updated various dependencies in `Cargo.toml` to their latest versions, including `tonic`, `tokio`, and `url`.
- Refactored `ServerConfig` to include a new `grpc_target` field, replacing the previous bind and port fields.
- Removed the `routes.rs` file and integrated its functionality into the main application logic.
- Enhanced the `PrinterInfo` struct to include additional fields for printer details.
- Simplified the main application flow by directly using the `ControlService` for handling requests.
- Cleaned up unused code and improved overall organization of the project structure.
This commit is contained in:
兰一民
2025-12-14 18:25:26 +08:00
parent 2f1bdac921
commit 5789a4346a
12 changed files with 1510 additions and 742 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1007
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

14
Cargo.toml
View File

@@ -6,8 +6,6 @@ edition = "2024"
[dependencies]
anyhow = "1"
async-trait = "0.1"
axum = { version = "0.7", features = ["multipart", "json"] }
base64 = "0.21"
config = { version = "0.14", default-features = false, features = ["yaml"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
@@ -16,6 +14,16 @@ thiserror = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "fs", "signal", "net"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
tower-http = { version = "0.5", features = ["trace"] }
cups_rs = "0.3"
tempfile = "3"
tonic = { version = "0.11", features = ["transport"] }
prost = "0.12"
prost-types = "0.12"
tokio-stream = "0.1"
hostname = "0.3"
sha2 = "0.10"
url = "2"
[build-dependencies]
tonic-build = "0.11"
protoc-bin-vendored = "3"

12
build.rs Normal file
View File

@@ -0,0 +1,12 @@
fn main() {
let protoc = protoc_bin_vendored::protoc_bin_path().expect("failed to fetch protoc");
unsafe {
std::env::set_var("PROTOC", protoc);
}
tonic_build::configure()
.build_server(false)
.compile(&["proto/control.proto"], &["proto"])
.expect("failed to compile protobuf");
}

82
proto/control.proto Normal file
View File

@@ -0,0 +1,82 @@
syntax = "proto3";
package control.v1;
service Control {
rpc ControlStream(stream ClientMessage) returns (stream ServerMessage);
}
message ClientMessage {
oneof msg {
Hello hello = 1;
PrintResult result = 2;
Pong pong = 3;
PrinterUpdate printers = 4;
}
}
message ServerMessage {
oneof msg {
PrintInstruction print = 1;
Ping ping = 2;
}
}
message Hello {
string client_id = 1;
string version = 2;
string hostname = 3;
repeated PrinterInfo printers = 4;
}
message Ping {
string nonce = 1;
}
message Pong {
string nonce = 1;
}
message PrintInstruction {
string request_id = 1;
bytes pdf_data = 2;
PrintParams params = 3;
}
message PrintParams {
uint32 copies = 1;
string duplex = 2; // one_sided, long_edge, short_edge
string color = 3; // auto, color, monochrome
string media = 4;
string quality = 5; // draft, normal, high
string orientation = 6; // portrait, landscape
string job_name = 7;
string printer = 8;
}
message PrintResult {
string request_id = 1;
bool ok = 2;
string printer = 3;
string message = 4;
uint32 retries_used = 5;
}
message PrinterUpdate {
repeated PrinterInfo printers = 1;
}
message PrinterInfo {
string id = 1;
string name = 2;
string host = 3;
string uri = 4;
string state = 5;
bool accepting_jobs = 6;
bool color_supported = 7;
uint32 active_jobs = 8;
uint32 ppm = 9;
uint32 ppm_color = 10;
repeated string reasons = 11;
}

26
src/config.rs
View File

@@ -22,13 +22,9 @@ pub struct AppConfig {
#[derive(Debug, Clone, Deserialize)]
pub struct ServerConfig {
#[serde(default = "default_bind")]
pub bind: String,
#[serde(default = "default_port")]
pub port: u16,
/// 请求体大小上限（字节）
#[serde(default = "default_body_limit")]
pub body_limit_bytes: u64,
/// gRPC 控制端地址，例如 http://127.0.0.1:50051
#[serde(default = "default_grpc_target")]
pub grpc_target: String,
}
#[derive(Debug, Clone, Deserialize)]
@@ -93,9 +89,7 @@ impl RetryConfig {
impl Default for ServerConfig {
fn default() -> Self {
Self {
bind: default_bind(),
port: default_port(),
body_limit_bytes: default_body_limit(),
grpc_target: default_grpc_target(),
}
}
}
@@ -147,16 +141,8 @@ impl AppConfig {
}
}
fn default_bind() -> String {
"0.0.0.0".to_string()
}
fn default_port() -> u16 {
8080
}
fn default_body_limit() -> u64 {
20 * 1024 * 1024 // 20MB
fn default_grpc_target() -> String {
"http://127.0.0.1:50051".to_string()
}
fn default_attempts() -> usize {

370
src/control_client.rs Normal file
View File

@@ -0,0 +1,370 @@
use std::{collections::HashMap, path::PathBuf, sync::Arc, time::Duration};
use anyhow::{anyhow, Context, Result};
use tempfile::NamedTempFile;
use tokio::{
io::AsyncWriteExt,
task::JoinHandle,
time::{sleep, timeout},
};
use tokio_stream::wrappers::ReceiverStream;
use tracing::{info, warn};
use crate::{
config::AppConfig,
models::{ColorModeSetting, DuplexSetting, OrientationSetting, PrintParams, QualitySetting},
printer::PrinterInfo,
proto::control::{
client_message, control_client::ControlClient, server_message, ClientMessage, Hello, Ping,
Pong, PrintInstruction, PrintParams as ProtoPrintParams, PrintResult, PrinterInfo as ProtoPrinter,
PrinterUpdate,
},
scheduler::Scheduler,
};
use cups_rs::PrinterState;
const RECONNECT_BASE_MS: u64 = 1000;
const RECONNECT_MAX_MS: u64 = 10000;
pub struct ControlService {
scheduler: Arc<Scheduler>,
config: AppConfig,
}
impl ControlService {
pub fn new(scheduler: Arc<Scheduler>, config: AppConfig) -> Self {
Self { scheduler, config }
}
pub async fn run(&self) -> Result<()> {
let mut backoff = RECONNECT_BASE_MS;
loop {
match self.connect_and_run().await {
Ok(()) => return Ok(()),
Err(e) => {
warn!("控制流连接失败: {e}");
sleep(Duration::from_millis(backoff)).await;
backoff = (backoff * 2).min(RECONNECT_MAX_MS);
}
}
}
}
async fn connect_and_run(&self) -> Result<()> {
let mut client = ControlClient::connect(self.config.server.grpc_target.clone())
.await
.context("连接 gRPC 控制端失败")?;
let (tx, rx) = tokio::sync::mpsc::channel::<ClientMessage>(32);
let outbound = ReceiverStream::new(rx);
let response = client
.control_stream(outbound)
.await
.context("建立控制流失败")?;
let mut stream = response.into_inner();
let printers = collect_printers(&self.scheduler).await;
let printer_payload: Vec<ProtoPrinter> =
printers.iter().map(to_proto_printer).collect();
let hostname_value = hostname().unwrap_or_default();
let client_id_value = if hostname_value.is_empty() {
"unknown-client".to_string()
} else {
hostname_value.clone()
};
// send hello
let hello = ClientMessage {
msg: Some(client_message::Msg::Hello(Hello {
client_id: client_id_value,
version: env!("CARGO_PKG_VERSION").to_string(),
hostname: hostname_value,
printers: printer_payload.clone(),
})),
};
let _ = tx.send(hello).await;
let tx_for_print = tx.clone();
let scheduler = self.scheduler.clone();
let config = self.config.clone();
let watch_tx = tx.clone();
let watch_scheduler = self.scheduler.clone();
let watch_interval =
Duration::from_secs(self.config.scheduler.refresh_interval_secs.max(5));
tokio::spawn(async move {
if let Err(e) =
watch_printer_changes(watch_scheduler, watch_tx, printers, watch_interval).await
{
warn!("打印机变更上报失败: {e}");
}
});
while let Some(msg) = stream.message().await? {
match msg.msg {
Some(server_message::Msg::Print(print)) => {
let tx = tx_for_print.clone();
let scheduler = scheduler.clone();
let config = config.clone();
tokio::spawn(async move {
if let Err(e) = handle_print(print, scheduler, config, tx).await {
warn!("处理打印指令失败: {e}");
}
});
}
Some(server_message::Msg::Ping(Ping { nonce })) => {
let _ = tx_for_print
.send(ClientMessage {
msg: Some(client_message::Msg::Pong(Pong { nonce })),
})
.await;
}
None => warn!("收到空指令"),
}
}
Err(anyhow!("控制流结束"))
}
}
async fn handle_print(
instr: PrintInstruction,
scheduler: Arc<Scheduler>,
config: AppConfig,
tx: tokio::sync::mpsc::Sender<ClientMessage>,
) -> Result<()> {
let params_proto = instr
.params
.as_ref()
.ok_or_else(|| anyhow!("缺少打印参数"))?;
let params = map_params(params_proto)?;
let temp = TempPdf::new(instr.pdf_data).await?;
let result = scheduler
.dispatch(
temp.path(),
&params,
&config.defaults,
&config.retry,
)
.await;
match result {
Ok((job, retries_used)) => {
info!(
printer = %job.printer,
job_id = job.job_id,
retries = retries_used,
"打印任务完成"
);
send_result(
tx,
PrintResult {
request_id: instr.request_id,
ok: true,
printer: job.printer,
message: "".to_string(),
retries_used: retries_used as u32,
},
)
.await;
}
Err(e) => {
warn!(error = %e, "打印任务失败");
send_result(
tx,
PrintResult {
request_id: instr.request_id,
ok: false,
printer: "".to_string(),
message: e.to_string(),
retries_used: 0,
},
)
.await;
}
}
Ok(())
}
async fn send_result(
tx: tokio::sync::mpsc::Sender<ClientMessage>,
result: PrintResult,
) {
if let Err(e) = tx
.send(ClientMessage {
msg: Some(client_message::Msg::Result(result)),
})
.await
{
warn!("发送打印结果失败: {e}");
}
}
async fn collect_printers(scheduler: &Scheduler) -> Vec<PrinterInfo> {
match scheduler.refresh_snapshot().await {
Ok(list) => list,
Err(e) => {
warn!("刷新打印机列表失败: {e}");
scheduler.snapshot().await
}
}
}
async fn watch_printer_changes(
scheduler: Arc<Scheduler>,
tx: tokio::sync::mpsc::Sender<ClientMessage>,
mut last: Vec<PrinterInfo>,
interval: Duration,
) -> Result<()> {
loop {
sleep(interval).await;
let current = collect_printers(&scheduler).await;
if printers_changed(&last, &current) {
let payload: Vec<ProtoPrinter> = current.iter().map(to_proto_printer).collect();
tx.send(ClientMessage {
msg: Some(client_message::Msg::Printers(PrinterUpdate {
printers: payload,
})),
})
.await
.map_err(|e| anyhow!("发送打印机更新失败: {e}"))?;
last = current;
}
}
}
fn printers_changed(prev: &[PrinterInfo], curr: &[PrinterInfo]) -> bool {
if prev.len() != curr.len() {
return true;
}
let prev_map: HashMap<&str, &PrinterInfo> =
prev.iter().map(|p| (p.id.as_str(), p)).collect();
for printer in curr {
match prev_map.get(printer.id.as_str()) {
Some(existing) if *existing == printer => continue,
_ => return true,
}
}
false
}
fn to_proto_printer(p: &PrinterInfo) -> ProtoPrinter {
ProtoPrinter {
id: p.id.clone(),
name: p.dest.name.clone(),
host: p.host.clone().unwrap_or_default(),
uri: p.device_uri.clone().unwrap_or_default(),
state: printer_state_label(&p.state),
accepting_jobs: p.accepting_jobs,
color_supported: p.color_supported,
active_jobs: p.active_jobs as u32,
ppm: p.ppm.unwrap_or_default(),
ppm_color: p.ppm_color.unwrap_or_default(),
reasons: p.reasons.clone(),
}
}
fn printer_state_label(state: &PrinterState) -> String {
match state {
PrinterState::Idle => "idle",
PrinterState::Processing => "processing",
PrinterState::Stopped => "stopped",
PrinterState::Unknown => "unknown",
}
.to_string()
}
fn map_params(params: &ProtoPrintParams) -> Result<PrintParams> {
Ok(PrintParams {
copies: Some(params.copies.max(1)),
duplex: Some(match params.duplex.as_str() {
"one_sided" | "" => DuplexSetting::OneSided,
"long_edge" => DuplexSetting::TwoSidedLongEdge,
"short_edge" => DuplexSetting::TwoSidedShortEdge,
other => return Err(anyhow!("未知双面选项: {other}")),
}),
color: Some(match params.color.as_str() {
"auto" | "" => ColorModeSetting::Auto,
"color" => ColorModeSetting::Color,
"monochrome" => ColorModeSetting::Monochrome,
other => return Err(anyhow!("未知色彩选项: {other}")),
}),
media: if params.media.is_empty() {
None
} else {
Some(params.media.clone())
},
quality: Some(match params.quality.as_str() {
"draft" => QualitySetting::Draft,
"high" => QualitySetting::High,
"" | "normal" => QualitySetting::Normal,
other => return Err(anyhow!("未知质量选项: {other}")),
}),
orientation: match params.orientation.as_str() {
"" => None,
"portrait" => Some(OrientationSetting::Portrait),
"landscape" => Some(OrientationSetting::Landscape),
other => return Err(anyhow!("未知方向选项: {other}")),
},
job_name: if params.job_name.is_empty() {
None
} else {
Some(params.job_name.clone())
},
printer: if params.printer.is_empty() {
None
} else {
Some(params.printer.clone())
},
})
}
struct TempPdf {
path: PathBuf,
_handle: JoinHandle<Result<()>>,
}
impl TempPdf {
async fn new(data: Vec<u8>) -> Result<Self> {
let (tx, rx) = tokio::sync::oneshot::channel();
let handle = tokio::spawn(async move {
let file = NamedTempFile::new().context("创建临时文件失败")?;
let mut writer = tokio::fs::File::from_std(file.reopen().context("打开临时文件失败")?);
writer
.write_all(&data)
.await
.context("写入临时文件失败")?;
writer.flush().await.context("刷新临时文件失败")?;
let path = file.path().to_path_buf();
// Hold file until drop to keep on disk
tx.send(path).ok();
// keep file alive
sleep(Duration::from_secs(300)).await;
Ok(())
});
let path = timeout(Duration::from_secs(5), rx)
.await
.context("等待临时文件路径超时")?
.context("获取临时文件路径失败")?;
Ok(Self {
path,
_handle: handle,
})
}
fn path(&self) -> &std::path::Path {
&self.path
}
}
fn hostname() -> Result<String> {
hostname::get()
.map(|s| s.to_string_lossy().to_string())
.context("读取主机名失败")
}

43
src/main.rs
View File

@@ -1,18 +1,16 @@
mod config;
mod error;
mod control_client;
mod models;
mod printer;
mod routes;
mod proto;
mod scheduler;
use std::{net::SocketAddr, sync::Arc};
use std::sync::Arc;
use anyhow::Context;
use axum::serve;
use control_client::ControlService;
use printer::{CupsClient, SharedPrinterAdapter};
use routes::{build_router, AppState};
use scheduler::Scheduler;
use tower_http::{limit::RequestBodyLimitLayer, trace::TraceLayer};
use tracing_subscriber::{fmt, EnvFilter};
#[tokio::main]
@@ -54,33 +52,14 @@ async fn main() -> Result<(), anyhow::Error> {
config.scheduler.printer_speeds.clone(),
));
let state = AppState {
scheduler,
config: config.clone(),
};
let service = ControlService::new(scheduler, config.clone());
let app = build_router(state)
.layer(TraceLayer::new_for_http())
.layer(RequestBodyLimitLayer::new(
config.server
.body_limit_bytes
.try_into()
.unwrap_or(usize::MAX),
));
let addr: SocketAddr = format!("{}:{}", config.server.bind, config.server.port)
.parse()
.context("解析监听地址失败")?;
let listener = tokio::net::TcpListener::bind(addr)
.await
.context("监听端口失败")?;
tracing::info!("服务启动于 http://{}", listener.local_addr()?);
serve(listener, app)
.with_graceful_shutdown(shutdown_signal())
.await
.context("HTTP 服务异常")?;
tokio::select! {
res = service.run() => res.context("控制客户端运行失败")?,
_ = shutdown_signal() => {
tracing::info!("收到退出信号，退出");
}
}
Ok(())
}

14
src/models.rs
View File

@@ -175,17 +175,3 @@ impl PrintParams {
}
}
#[derive(Debug, Deserialize)]
pub struct JsonPrintRequest {
pub pdf_base64: String,
pub params: PrintParams,
}
#[derive(Debug, Serialize)]
pub struct PrintJobResponse {
pub job_id: i32,
pub printer: String,
pub strategy: String,
pub retries_used: usize,
}

110
src/printer.rs
View File

@@ -7,22 +7,31 @@ use cups_rs::{
job::{cancel_job, create_job_with_options, FORMAT_PDF},
Destination, PrinterState,
};
use sha2::{Digest, Sha256};
use tokio::task::spawn_blocking;
use tracing::{debug, warn};
use tracing::warn;
use url::Url;
use crate::models::{PrintDefaults, PrintParams};
#[derive(Debug, Clone)]
pub struct PrinterInfo {
pub id: String,
pub dest: Destination,
pub state: PrinterState,
pub reasons: Vec<String>,
pub active_jobs: usize,
pub color_supported: bool,
pub ppm: Option<u32>,
pub ppm_color: Option<u32>,
pub device_uri: Option<String>,
pub host: Option<String>,
pub accepting_jobs: bool,
}
impl PrinterInfo {
pub fn healthy(&self) -> bool {
self.dest.is_accepting_jobs() && self.state.is_available() && !self.has_error_reason()
self.accepting_jobs && self.state.is_available() && !self.has_error_reason()
}
fn has_error_reason(&self) -> bool {
@@ -53,7 +62,6 @@ pub trait PrintAdapter: Send + Sync {
params: &PrintParams,
defaults: &PrintDefaults,
) -> Result<PrintJobResult>;
async fn cancel(&self, job_id: i32) -> Result<()>;
}
#[derive(Debug, Default)]
@@ -69,15 +77,48 @@ impl PrintAdapter for CupsClient {
for dest in destinations {
let state = dest.state();
let reasons = dest.state_reasons();
let options = dest.get_options();
let color_supported = options
.get("print-color-mode-supported")
.map(|modes| {
modes
.split(',')
.any(|m| m.trim().eq_ignore_ascii_case("color"))
})
.unwrap_or(true);
let ppm = parse_speed_option(options.get("printer-speed"));
let ppm_color = parse_speed_option(
options
.get("printer-speed-color-supported")
.or_else(|| options.get("printer-color-ppm-supported")),
);
let active_jobs = get_active_jobs(Some(dest.name.as_str()))
.map(|jobs| jobs.len())
.unwrap_or(0);
let device_uri = options
.get("device-uri")
.cloned()
.or_else(|| options.get("printer-uri-supported").cloned());
let host = device_uri
.as_deref()
.and_then(extract_host_from_uri)
.or_else(|| options.get("printer-host").map(|s| s.to_string()));
let id = derive_printer_id(&dest, device_uri.as_deref(), host.as_deref());
let accepting_jobs = dest.is_accepting_jobs();
infos.push(PrinterInfo {
id,
dest,
state,
reasons,
active_jobs,
color_supported,
ppm,
ppm_color,
device_uri,
host,
accepting_jobs,
});
}
Ok::<_, anyhow::Error>(infos)
@@ -116,13 +157,64 @@ impl PrintAdapter for CupsClient {
.await
.context("提交打印任务失败")?
}
async fn cancel(&self, job_id: i32) -> Result<()> {
spawn_blocking(move || cancel_job(job_id).context("取消打印作业失败"))
.await
.context("取消任务失败")?
}
}
pub type SharedPrinterAdapter = Arc<dyn PrintAdapter>;
fn parse_speed_option(input: Option<&String>) -> Option<u32> {
input.and_then(|s| s.trim().parse::<u32>().ok())
}
fn derive_printer_id(
dest: &Destination,
device_uri: Option<&str>,
host: Option<&str>,
) -> String {
let mut hasher = Sha256::new();
hasher.update(dest.name.as_bytes());
if let Some(uri) = device_uri {
hasher.update(uri.as_bytes());
}
if let Some(host) = host {
hasher.update(host.as_bytes());
}
let hash = hasher.finalize();
let mut hex = format!("{:x}", hash);
hex.truncate(12);
format!("prn-{hex}")
}
fn extract_host_from_uri(uri: &str) -> Option<String> {
Url::parse(uri)
.ok()
.and_then(|u| u.host_str().map(|h| h.to_string()))
.or_else(|| {
uri.split("//")
.nth(1)
.and_then(|rest| {
rest.split(['/', ':', '?'])
.filter(|s| !s.is_empty())
.next()
})
.map(|s| s.to_string())
})
}
impl PartialEq for PrinterInfo {
fn eq(&self, other: &Self) -> bool {
self.id == other.id
&& self.state == other.state
&& self.reasons == other.reasons
&& self.active_jobs == other.active_jobs
&& self.color_supported == other.color_supported
&& self.ppm == other.ppm
&& self.ppm_color == other.ppm_color
&& self.device_uri == other.device_uri
&& self.host == other.host
&& self.accepting_jobs == other.accepting_jobs
}
}
impl Eq for PrinterInfo {}

4
src/proto.rs Normal file
View File

@@ -0,0 +1,4 @@
pub mod control {
tonic::include_proto!("control.v1");
}

502
src/routes.rs
View File

@@ -1,502 +0,0 @@
use std::{io::Write, sync::Arc};
use axum::{
extract::{Multipart, Path, Query, State},
http::StatusCode,
routing::{get, post},
Json, Router,
};
use anyhow::anyhow;
use base64::engine::general_purpose::STANDARD as BASE64_STD;
use base64::Engine;
use serde::Serialize;
use tempfile::NamedTempFile;
use tracing::info;
use crate::{
config::AppConfig,
error::AppError,
models::{JsonPrintRequest, PrintJobResponse, PrintParams},
scheduler::Scheduler,
};
#[derive(Clone)]
pub struct AppState {
pub scheduler: Arc<Scheduler>,
pub config: AppConfig,
}
pub fn build_router(state: AppState) -> Router {
Router::new()
.route("/health", get(health))
.route("/printers", get(printers))
.route("/printers/:name/detail", get(printer_detail))
.route("/print", post(print_base64))
.route("/print/upload", post(print_multipart))
.with_state(state)
}
async fn health(State(state): State<AppState>) -> Result<Json<HealthResp>, AppError> {
let printers = state.scheduler.healthy_snapshot().await?;
let details = printers
.into_iter()
.map(|p| {
let healthy = p.healthy();
PrinterStatus {
name: p.dest.name.clone(),
state: format!("{}", p.state),
accepting_jobs: p.dest.is_accepting_jobs(),
active_jobs: p.active_jobs,
reasons: p.reasons,
healthy,
}
})
.collect::<Vec<_>>();
Ok(Json(HealthResp {
status: "ok".to_string(),
available_printers: details.len(),
printers: details,
}))
}
async fn printers(State(state): State<AppState>) -> Result<Json<PrintersResp>, AppError> {
let printers = state.scheduler.healthy_snapshot().await?;
let details = printers
.into_iter()
.map(|p| {
let healthy = p.healthy();
PrinterStatus {
name: p.dest.name.clone(),
state: format!("{}", p.state),
accepting_jobs: p.dest.is_accepting_jobs(),
active_jobs: p.active_jobs,
reasons: p.reasons,
healthy,
}
})
.collect::<Vec<_>>();
Ok(Json(PrintersResp { printers: details }))
}
async fn print_base64(
State(state): State<AppState>,
Json(payload): Json<JsonPrintRequest>,
) -> Result<(StatusCode, Json<PrintJobResponse>), AppError> {
let data = BASE64_STD
.decode(payload.pdf_base64)
.map_err(|e| AppError::BadRequest(format!("PDF base64 解码失败: {}", e)))?;
let response = process_print(&state, data, payload.params).await?;
Ok((StatusCode::ACCEPTED, Json(response)))
}
async fn print_multipart(
State(state): State<AppState>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<PrintJobResponse>), AppError> {
let mut pdf: Option<Vec<u8>> = None;
let mut params: Option<PrintParams> = None;
while let Some(field) = multipart
.next_field()
.await
.map_err(|e| AppError::BadRequest(format!("解析表单失败: {}", e)))?
{
match field.name() {
Some("file") => {
pdf = Some(
field
.bytes()
.await
.map_err(|e| AppError::BadRequest(format!("读取文件失败: {}", e)))?
.to_vec(),
);
}
Some("params") => {
let text = field
.text()
.await
.map_err(|e| AppError::BadRequest(format!("读取参数失败: {}", e)))?;
params = Some(
serde_json::from_str(&text)
.map_err(|e| AppError::BadRequest(format!("参数 JSON 无效: {}", e)))?,
);
}
_ => {}
}
}
let pdf = pdf.ok_or_else(|| AppError::BadRequest("缺少 file 字段".into()))?;
let params = params.ok_or_else(|| AppError::BadRequest("缺少 params 字段".into()))?;
let response = process_print(&state, pdf, params).await?;
Ok((StatusCode::ACCEPTED, Json(response)))
}
async fn process_print(
state: &AppState,
pdf_data: Vec<u8>,
params: PrintParams,
) -> Result<PrintJobResponse, AppError> {
if !is_pdf(&pdf_data) {
return Err(AppError::BadRequest("仅支持 PDF 文件".into()));
}
let temp = TempPdf::new(pdf_data).await?;
let (job, retries_used) = state
.scheduler
.dispatch(
temp.path(),
&params,
&state.config.defaults,
&state.config.retry,
)
.await?;
info!(
printer = %job.printer,
job_id = job.job_id,
retries = retries_used,
"打印任务已提交"
);
Ok(PrintJobResponse {
job_id: job.job_id,
printer: job.printer,
strategy: format!("{:?}", state.config.scheduler.strategy),
retries_used,
})
}
struct TempPdf {
file: NamedTempFile,
}
impl TempPdf {
async fn new(data: Vec<u8>) -> Result<Self, AppError> {
let file = tokio::task::spawn_blocking(move || -> Result<NamedTempFile, AppError> {
let mut file = NamedTempFile::new()
.map_err(|e| AppError::Internal(anyhow!("创建临时文件失败: {}", e)))?;
file.write_all(&data)
.map_err(|e| AppError::Internal(anyhow!(e)))?;
file.flush()
.map_err(|e| AppError::Internal(anyhow!(e)))?;
Ok(file)
})
.await
.map_err(|e| AppError::Internal(anyhow!(e)))??;
Ok(Self { file })
}
fn path(&self) -> &std::path::Path {
self.file.path()
}
}
#[derive(Serialize)]
struct HealthResp {
status: String,
available_printers: usize,
printers: Vec<PrinterStatus>,
}
#[derive(Serialize)]
struct PrinterStatus {
name: String,
state: String,
accepting_jobs: bool,
active_jobs: usize,
reasons: Vec<String>,
healthy: bool,
}
fn is_pdf(data: &[u8]) -> bool {
data.starts_with(b"%PDF")
}
#[derive(Serialize)]
struct PrintersResp {
printers: Vec<PrinterStatus>,
}
#[derive(Serialize)]
struct PrinterDetailResp {
name: String,
state: String,
accepting_jobs: bool,
active_jobs: usize,
reasons: Vec<String>,
healthy: bool,
info: Option<String>,
location: Option<String>,
make_and_model: Option<String>,
uri: Option<String>,
device_uri: Option<String>,
options: std::collections::HashMap<String, String>,
media: MediaInfo,
alerts: Vec<String>,
}
async fn printer_detail(
State(state): State<AppState>,
Path(name): Path<String>,
Query(query): Query<std::collections::HashMap<String, String>>,
) -> Result<Json<PrinterDetailResp>, AppError> {
let refresh = query.get("refresh").map(|s| {
let lower = s.to_ascii_lowercase();
lower == "1" || lower == "true" || lower == "yes" || lower == "y"
}).unwrap_or(false);
let p = state
.scheduler
.printer_detail(&name, refresh)
.await
.map_err(|e| AppError::Printer(e.to_string()))?;
let healthy = p.healthy();
let (media_opt, alerts) = parse_media_and_alerts(p.dest.get_options());
let media = if let Some(m) = media_opt {
m
} else {
let from_ipptool = fetch_media_via_ipptool(
&state.config,
&p.dest.uri().cloned().unwrap_or_default(),
)
.await;
match from_ipptool {
Some(m) => m,
None => fetch_media_via_lpoptions(&state.config, &p.dest.name)
.await
.unwrap_or_default(),
}
};
let resp = PrinterDetailResp {
name: p.dest.name.clone(),
state: format!("{}", p.state),
accepting_jobs: p.dest.is_accepting_jobs(),
active_jobs: p.active_jobs,
reasons: p.reasons,
healthy,
info: p.dest.info().cloned(),
location: p.dest.location().cloned(),
make_and_model: p.dest.make_and_model().cloned(),
uri: p.dest.uri().cloned(),
device_uri: p.dest.device_uri().cloned(),
options: p.dest.get_options().clone(),
media,
alerts,
};
Ok(Json(resp))
}
#[derive(Serialize, Default, Clone)]
struct MediaInfo {
ready: Vec<String>,
supported: Vec<String>,
default: Option<String>,
sources: Vec<String>,
types: Vec<String>,
raw_ready_col: Option<String>,
}
fn parse_media_and_alerts(
options: &std::collections::HashMap<String, String>,
) -> (Option<MediaInfo>, Vec<String>) {
let mut media = None;
let mut m = MediaInfo::default();
if let Some(r) = options.get("media-ready") {
m.ready = split_list(r);
media = Some(m.clone());
}
if let Some(r) = options.get("media-supported") {
m.supported = split_list(r);
media = Some(m.clone());
}
if let Some(r) = options.get("media-default") {
m.default = Some(r.clone());
media = Some(m.clone());
}
if let Some(r) = options.get("media-source-supported") {
m.sources = split_list(r);
media = Some(m.clone());
}
if let Some(r) = options.get("media-type-supported") {
m.types = split_list(r);
media = Some(m.clone());
}
if let Some(r) = options.get("media-col-ready") {
m.raw_ready_col = Some(r.clone());
media = Some(m.clone());
}
let mut alerts = Vec::new();
if let Some(a) = options.get("printer-alert") {
alerts.extend(split_list(a));
}
if let Some(a) = options.get("printer-alert-description") {
alerts.extend(split_list(a));
}
(media, alerts)
}
fn split_list(s: &str) -> Vec<String> {
s.split(',')
.map(|v| v.trim().to_string())
.filter(|v| !v.is_empty())
.collect()
}
async fn fetch_media_via_lpoptions(config: &AppConfig, printer: &str) -> Option<MediaInfo> {
let server_env = std::env::var("CUPS_SERVER").ok();
let server = config
.cups
.server
.as_ref()
.map(|s| s.as_str())
.or_else(|| server_env.as_deref())
.unwrap_or("localhost:631");
let server_arg = format!("-h{}", server);
let printer_arg = format!("-p{}", printer);
let output_detail = tokio::task::spawn_blocking({
let server_arg = server_arg.clone();
let printer_arg = printer_arg.clone();
move || {
std::process::Command::new("lpoptions")
.args([server_arg.as_str(), printer_arg.as_str(), "-l"])
.output()
}
})
.await;
let output_defaults = tokio::task::spawn_blocking(move || {
std::process::Command::new("lpoptions")
.args([server_arg.as_str(), printer_arg.as_str()])
.output()
})
.await;
let detail = match output_detail {
Ok(Ok(o)) => o,
_ => return None,
};
if !detail.status.success() {
return None;
}
let defaults_opt = match output_defaults {
Ok(Ok(o)) if o.status.success() => Some(o),
_ => None,
};
let text = String::from_utf8_lossy(&detail.stdout);
let mut media = MediaInfo::default();
for line in text.lines() {
if line.starts_with("PageSize/") {
let tokens: Vec<&str> = line.split(':').nth(1).unwrap_or("").split_whitespace().collect();
let mut supported = Vec::new();
let mut default = None;
for t in tokens {
if t.starts_with('*') {
default = Some(t.trim_start_matches('*').to_string());
supported.push(default.as_ref().unwrap().clone());
} else {
supported.push(t.to_string());
}
}
media.supported = supported;
media.default = default;
} else if line.starts_with("InputSlot/") {
let tokens: Vec<&str> = line.split(':').nth(1).unwrap_or("").split_whitespace().collect();
let mut sources = Vec::new();
for t in tokens {
sources.push(t.trim_start_matches('*').to_string());
}
media.sources = sources;
} else if line.starts_with("MediaType/") {
let tokens: Vec<&str> = line.split(':').nth(1).unwrap_or("").split_whitespace().collect();
let mut types = Vec::new();
for t in tokens {
types.push(t.trim_start_matches('*').to_string());
}
media.types = types;
}
}
if let Some(o) = defaults_opt {
let defaults_line = String::from_utf8_lossy(&o.stdout);
for token in defaults_line.split_whitespace() {
if let Some(rest) = token.strip_prefix("InputSlot=") {
media.ready = vec![rest.to_string()];
}
}
}
Some(media)
}
async fn fetch_media_via_ipptool(_config: &AppConfig, uri: &str) -> Option<MediaInfo> {
if uri.is_empty() {
return None;
}
let uri = uri.to_string();
let output = tokio::task::spawn_blocking(move || {
std::process::Command::new("ipptool")
.args(["-v", "-t", uri.as_str(), "get-printer-attributes.test"])
.output()
})
.await
.ok()?;
let output = match output {
Ok(o) => o,
Err(_) => return None,
};
if !output.status.success() {
return None;
}
let mut text = String::new();
text.push_str(&String::from_utf8_lossy(&output.stdout));
text.push_str(&String::from_utf8_lossy(&output.stderr));
let mut media = MediaInfo::default();
for line in text.lines() {
let trimmed = line.trim();
if let Some(rest) = trimmed.strip_prefix("media-default (keyword) = ") {
media.default = Some(rest.to_string());
} else if let Some(rest) = trimmed.strip_prefix("media-ready (1setOf keyword) = ") {
media.ready = split_list(rest);
} else if let Some(rest) = trimmed.strip_prefix("media-supported (1setOf keyword) = ") {
media.supported = split_list(rest);
} else if let Some(rest) = trimmed.strip_prefix("media-source-supported (1setOf keyword) = ") {
media.sources = split_list(rest);
} else if let Some(rest) = trimmed.strip_prefix("media-type-supported (1setOf keyword) = ") {
media.types = split_list(rest);
} else if let Some(rest) = trimmed.strip_prefix("media-col-ready (1setOf collection) = ") {
media.raw_ready_col = Some(rest.to_string());
}
}
if media.ready.is_empty()
&& media.supported.is_empty()
&& media.default.is_none()
&& media.sources.is_empty()
&& media.types.is_empty()
&& media.raw_ready_col.is_none()
{
None
} else {
Some(media)
}
}

68
src/scheduler.rs
View File

@@ -13,14 +13,13 @@ use tracing::{info, warn};
use crate::{
config::{LoadBalanceStrategy, RetryConfig},
models::{PrintDefaults, PrintParams},
printer::{PrintAdapter, PrintJobResult, PrinterInfo, SharedPrinterAdapter},
printer::{PrintJobResult, PrinterInfo, SharedPrinterAdapter},
};
pub struct Scheduler {
adapter: SharedPrinterAdapter,
strategy: LoadBalanceStrategy,
rr_cursor: AtomicUsize,
refresh_interval: Duration,
cache: std::sync::Arc<RwLock<Vec<PrinterInfo>>>,
default_ppm_bw: u32,
default_ppm_color: u32,
@@ -44,7 +43,6 @@ impl Scheduler {
adapter: adapter.clone(),
strategy,
rr_cursor: AtomicUsize::new(0),
refresh_interval,
cache: cache.clone(),
default_ppm_bw,
default_ppm_color,
@@ -75,6 +73,23 @@ impl Scheduler {
this
}
pub async fn snapshot(&self) -> Vec<PrinterInfo> {
self.cache.read().await.clone()
}
pub async fn refresh_snapshot(&self) -> Result<Vec<PrinterInfo>> {
match self.adapter.list_printers().await {
Ok(list) => {
*self.cache.write().await = list.clone();
Ok(list)
}
Err(e) => {
warn!("刷新打印机列表失败: {}", e);
Ok(self.cache.read().await.clone())
}
}
}
async fn healthy_printers(&self) -> Result<Vec<PrinterInfo>> {
let mut printers = self.cache.read().await.clone();
@@ -121,7 +136,7 @@ impl Scheduler {
sorted
}
}
}
}
pub async fn dispatch(
&self,
@@ -220,32 +235,6 @@ impl Scheduler {
))
}
pub async fn healthy_snapshot(&self) -> Result<Vec<PrinterInfo>> {
let printers = self.cache.read().await.clone();
Ok(printers)
}
pub async fn printer_detail(&self, name: &str, refresh: bool) -> Result<PrinterInfo> {
if refresh {
if let Ok(list) = self.adapter.list_printers().await {
*self.cache.write().await = list;
}
}
{
let cache = self.cache.read().await;
if let Some(p) = cache.iter().find(|p| p.dest.name == name).cloned() {
return Ok(p);
}
}
let list = self.adapter.list_printers().await?;
*self.cache.write().await = list.clone();
list.into_iter()
.find(|p| p.dest.name == name)
.ok_or_else(|| anyhow!("未找到打印机: {}", name))
}
fn state_rank(state: &PrinterState) -> u8 {
match state {
PrinterState::Idle => 0,
@@ -304,7 +293,14 @@ impl Scheduler {
fn cost_estimate(&self, printer: &PrinterInfo, color: crate::models::ColorModeSetting) -> f64 {
// 简化成本：活跃作业数 / 速度；假设每个作业页数相近
let ppm = self.effective_ppm(printer, color);
(printer.active_jobs as f64 + 1.0) / ppm
let estimated_wait = (printer.active_jobs as f64 + 1.0) / ppm;
if let Some(target) = self.target_wait_minutes {
let normalized_target = target.max(0.1);
return estimated_wait / normalized_target;
}
estimated_wait
}
}
@@ -314,6 +310,7 @@ mod tests {
use anyhow::Result;
use async_trait::async_trait;
use cups_rs::{Destination, PrinterState};
use crate::printer::PrintAdapter;
use std::sync::Arc;
use tokio::runtime::Runtime;
@@ -335,14 +332,11 @@ mod tests {
) -> Result<PrintJobResult> {
unreachable!()
}
async fn cancel(&self, _job_id: i32) -> Result<()> {
Ok(())
}
}
fn mock_printer(name: &str, active_jobs: usize) -> PrinterInfo {
PrinterInfo {
id: format!("id-{name}"),
dest: Destination {
name: name.to_string(),
instance: None,
@@ -353,9 +347,11 @@ mod tests {
reasons: vec![],
active_jobs,
color_supported: true,
color_modes_supported: vec!["color".to_string(), "monochrome".to_string()],
ppm: Some(30),
ppm_color: Some(20),
device_uri: None,
host: None,
accepting_jobs: true,
}
}