Add initial project structure with configuration, printing, and routing functionality

- Created a new Rust project with Cargo configuration.
- Added `.gitignore` to exclude target directory.
- Implemented `Cargo.lock` for dependency management.
- Defined application configuration in `config/app.yaml` and `src/config.rs`.
- Developed printing logic in `src/printer.rs` using `cups_rs` for printer interactions.
- Established routing and request handling in `src/routes.rs` for health checks and print requests.
- Introduced a scheduler in `src/scheduler.rs` to manage print job distribution.
- Created models for print parameters and job responses in `src/models.rs`.
- Set up the main application entry point in `src/main.rs` with server initialization and configuration loading.
- Included necessary dependencies in `Cargo.toml` for async handling, tracing, and configuration management.
This commit is contained in:
2025-12-11 19:50:03 +08:00
commit 2f1bdac921
10 changed files with 2980 additions and 0 deletions

1
.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
/target

1377
Cargo.lock generated Normal file

File diff suppressed because it is too large Load Diff

21
Cargo.toml Normal file
View File

@@ -0,0 +1,21 @@
[package]
name = "print-backend"
version = "0.1.0"
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"
serde_yaml = "0.9"
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"

28
config/app.yaml Normal file
View File

@@ -0,0 +1,28 @@
server:
bind: "0.0.0.0"
port: 8080
cups:
server: "http://v4.hitwh.games:631"
user: "root"
password: "lanyimin123"
# 如需指定 CUPS 服务器/用户,示例:
# cups:
# server: "http://cups.local:631"
# user: "printsvc"
scheduler:
strategy: "least_queued" # round_robin | least_queued
refresh_interval_secs: 10
retry:
attempts: 3
backoff_ms: 1500
defaults:
copies: 1
duplex: "two_sided_long_edge" # one_sided | two_sided_long_edge | two_sided_short_edge
color: "auto" # auto | color | monochrome
media: "A4"
quality: "normal" # draft | normal | high

210
src/config.rs Normal file
View File

@@ -0,0 +1,210 @@
use std::time::Duration;
use anyhow::Context;
use config as cfg;
use serde::Deserialize;
use crate::models::{ColorModeSetting, DuplexSetting, PrintDefaults, QualitySetting};
#[derive(Debug, Clone, Deserialize)]
pub struct AppConfig {
#[serde(default)]
pub server: ServerConfig,
#[serde(default)]
pub cups: CupsConfig,
#[serde(default)]
pub scheduler: SchedulerConfig,
#[serde(default)]
pub retry: RetryConfig,
#[serde(default)]
pub defaults: PrintDefaults,
}
#[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,
}
#[derive(Debug, Clone, Deserialize)]
pub struct CupsConfig {
/// 例如 http://cups.local:631 或 /run/cups/cups.sock
#[serde(default)]
pub server: Option<String>,
#[serde(default)]
pub user: Option<String>,
#[serde(default)]
pub password: Option<String>,
}
#[derive(Debug, Clone, Deserialize)]
pub struct SchedulerConfig {
#[serde(default = "default_strategy")]
pub strategy: LoadBalanceStrategy,
/// 后台刷新打印机状态的最小间隔,避免频繁访问 CUPS
#[serde(default = "default_refresh_secs")]
pub refresh_interval_secs: u64,
/// 黑白默认 PPM当打印机未上报时
#[serde(default = "default_ppm_bw")]
pub default_ppm_bw: u32,
/// 彩色默认 PPM当打印机未上报时
#[serde(default = "default_ppm_color")]
pub default_ppm_color: u32,
/// 期望的最大等待时间(分钟),仅用于排序估计
#[serde(default)]
pub target_wait_minutes: Option<f64>,
/// 打印机速度覆盖
#[serde(default)]
pub printer_speeds: std::collections::HashMap<String, PrinterSpeedOverride>,
}
#[derive(Debug, Clone, Deserialize, Copy)]
#[serde(rename_all = "snake_case")]
pub enum LoadBalanceStrategy {
RoundRobin,
LeastQueued,
}
impl Default for LoadBalanceStrategy {
fn default() -> Self {
LoadBalanceStrategy::LeastQueued
}
}
#[derive(Debug, Clone, Deserialize)]
pub struct RetryConfig {
#[serde(default = "default_attempts")]
pub attempts: usize,
#[serde(default = "default_backoff_ms")]
pub backoff_ms: u64,
}
impl RetryConfig {
pub fn backoff(&self) -> Duration {
Duration::from_millis(self.backoff_ms)
}
}
impl Default for ServerConfig {
fn default() -> Self {
Self {
bind: default_bind(),
port: default_port(),
body_limit_bytes: default_body_limit(),
}
}
}
impl Default for CupsConfig {
fn default() -> Self {
Self {
server: None,
user: None,
password: None,
}
}
}
impl Default for SchedulerConfig {
fn default() -> Self {
Self {
strategy: default_strategy(),
refresh_interval_secs: default_refresh_secs(),
default_ppm_bw: default_ppm_bw(),
default_ppm_color: default_ppm_color(),
target_wait_minutes: None,
printer_speeds: Default::default(),
}
}
}
impl Default for RetryConfig {
fn default() -> Self {
Self {
attempts: default_attempts(),
backoff_ms: default_backoff_ms(),
}
}
}
impl AppConfig {
pub fn load(path: Option<&str>) -> Result<Self, anyhow::Error> {
let path = path.unwrap_or("config/app.yaml");
let builder = cfg::Config::builder()
.add_source(cfg::File::with_name(path))
.add_source(cfg::Environment::with_prefix("PRINT").separator("__"));
builder
.build()
.context("读取配置失败")?
.try_deserialize::<AppConfig>()
.context("配置格式错误")
}
}
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_attempts() -> usize {
3
}
fn default_backoff_ms() -> u64 {
1500
}
fn default_strategy() -> LoadBalanceStrategy {
LoadBalanceStrategy::LeastQueued
}
fn default_refresh_secs() -> u64 {
10
}
fn default_ppm_bw() -> u32 {
30
}
fn default_ppm_color() -> u32 {
20
}
// Defaults reused by PrintDefaults
pub(crate) fn default_copies() -> u32 {
1
}
pub(crate) fn default_duplex() -> DuplexSetting {
DuplexSetting::OneSided
}
pub(crate) fn default_color() -> ColorModeSetting {
ColorModeSetting::Auto
}
pub(crate) fn default_quality() -> QualitySetting {
QualitySetting::Normal
}
#[derive(Debug, Clone, Deserialize)]
pub struct PrinterSpeedOverride {
#[serde(default)]
pub ppm: Option<u32>,
#[serde(default)]
pub ppm_color: Option<u32>,
}

109
src/main.rs Normal file
View File

@@ -0,0 +1,109 @@
mod config;
mod error;
mod models;
mod printer;
mod routes;
mod scheduler;
use std::{net::SocketAddr, sync::Arc};
use anyhow::Context;
use axum::serve;
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]
async fn main() -> Result<(), anyhow::Error> {
init_tracing();
let config_path = std::env::var("CONFIG_PATH").ok();
let config = config::AppConfig::load(config_path.as_deref())?;
if let Some(server) = &config.cups.server {
let normalized = normalize_server(server);
unsafe {
std::env::set_var("CUPS_SERVER", &normalized);
}
cups_rs::config::set_server(Some(&normalized)).context("设置 CUPS 服务器失败")?;
tracing::info!("已配置远程 CUPS 服务器: {}", normalized);
}
if let Some(user) = &config.cups.user {
unsafe {
std::env::set_var("CUPS_USER", user);
}
cups_rs::config::set_user(Some(user)).context("设置 CUPS 用户失败")?;
}
if let Some(password) = &config.cups.password {
// 设置 CUPS_PASSWORD 供远程认证使用
unsafe {
std::env::set_var("CUPS_PASSWORD", password);
}
}
let adapter: SharedPrinterAdapter = Arc::new(CupsClient::default());
let scheduler = Arc::new(Scheduler::new(
config.scheduler.strategy,
config.scheduler.refresh_interval_secs,
adapter,
config.scheduler.default_ppm_bw,
config.scheduler.default_ppm_color,
config.scheduler.target_wait_minutes,
config.scheduler.printer_speeds.clone(),
));
let state = AppState {
scheduler,
config: 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 服务异常")?;
Ok(())
}
fn init_tracing() {
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
fmt().with_env_filter(filter).init();
}
async fn shutdown_signal() {
let _ = tokio::signal::ctrl_c().await;
tracing::info!("收到退出信号,准备退出");
}
fn normalize_server(input: &str) -> String {
let without_scheme = if let Some((_, rest)) = input.split_once("://") {
rest
} else {
input
};
without_scheme
.split('/')
.next()
.unwrap_or(without_scheme)
.to_string()
}

191
src/models.rs Normal file
View File

@@ -0,0 +1,191 @@
use std::time::{SystemTime, UNIX_EPOCH};
use cups_rs::job::{ColorMode, DuplexMode, Orientation, PrintOptions, PrintQuality};
use serde::{Deserialize, Serialize};
use crate::config;
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ColorModeSetting {
Auto,
Color,
Monochrome,
}
impl Default for ColorModeSetting {
fn default() -> Self {
ColorModeSetting::Auto
}
}
impl From<ColorModeSetting> for ColorMode {
fn from(value: ColorModeSetting) -> Self {
match value {
ColorModeSetting::Auto => ColorMode::Auto,
ColorModeSetting::Color => ColorMode::Color,
ColorModeSetting::Monochrome => ColorMode::Monochrome,
}
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DuplexSetting {
OneSided,
TwoSidedLongEdge,
TwoSidedShortEdge,
}
impl Default for DuplexSetting {
fn default() -> Self {
DuplexSetting::OneSided
}
}
impl From<DuplexSetting> for DuplexMode {
fn from(value: DuplexSetting) -> Self {
match value {
DuplexSetting::OneSided => DuplexMode::OneSided,
DuplexSetting::TwoSidedLongEdge => DuplexMode::TwoSidedPortrait,
DuplexSetting::TwoSidedShortEdge => DuplexMode::TwoSidedLandscape,
}
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum QualitySetting {
Draft,
Normal,
High,
}
impl Default for QualitySetting {
fn default() -> Self {
QualitySetting::Normal
}
}
impl From<QualitySetting> for PrintQuality {
fn from(value: QualitySetting) -> Self {
match value {
QualitySetting::Draft => PrintQuality::Draft,
QualitySetting::Normal => PrintQuality::Normal,
QualitySetting::High => PrintQuality::High,
}
}
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OrientationSetting {
Portrait,
Landscape,
}
impl From<OrientationSetting> for Orientation {
fn from(value: OrientationSetting) -> Self {
match value {
OrientationSetting::Portrait => Orientation::Portrait,
OrientationSetting::Landscape => Orientation::Landscape,
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PrintParams {
#[serde(default)]
pub copies: Option<u32>,
#[serde(default)]
pub duplex: Option<DuplexSetting>,
#[serde(default)]
pub color: Option<ColorModeSetting>,
#[serde(default)]
pub media: Option<String>,
#[serde(default)]
pub quality: Option<QualitySetting>,
#[serde(default)]
pub orientation: Option<OrientationSetting>,
#[serde(default)]
pub job_name: Option<String>,
/// 可选,指定打印机名称;为空则自动选择
#[serde(default)]
pub printer: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PrintDefaults {
#[serde(default = "config::default_copies")]
pub copies: u32,
#[serde(default = "config::default_duplex")]
pub duplex: DuplexSetting,
#[serde(default = "config::default_color")]
pub color: ColorModeSetting,
#[serde(default)]
pub media: Option<String>,
#[serde(default = "config::default_quality")]
pub quality: QualitySetting,
#[serde(default)]
pub orientation: Option<OrientationSetting>,
}
impl Default for PrintDefaults {
fn default() -> Self {
Self {
copies: config::default_copies(),
duplex: config::default_duplex(),
color: config::default_color(),
media: None,
quality: config::default_quality(),
orientation: None,
}
}
}
impl PrintParams {
pub fn to_print_options(&self, defaults: &PrintDefaults) -> PrintOptions {
let mut opts = PrintOptions::new().copies(self.copies.unwrap_or(defaults.copies));
opts = opts.duplex(self.duplex.unwrap_or(defaults.duplex).into());
opts = opts.color_mode(self.color.clone().unwrap_or(defaults.color).into());
opts = opts.quality(self.quality.unwrap_or(defaults.quality).into());
if let Some(media) = self.media.as_ref().or(defaults.media.as_ref()) {
opts = opts.media(media);
}
if let Some(orientation) = self.orientation.as_ref().or(defaults.orientation.as_ref()) {
opts = opts.orientation(orientation.clone().into());
}
opts
}
pub fn effective_job_name(&self) -> String {
if let Some(name) = &self.job_name {
return name.clone();
}
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
format!("print-job-{}", now)
}
}
#[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,
}

128
src/printer.rs Normal file
View File

@@ -0,0 +1,128 @@
use std::{path::Path, sync::Arc};
use anyhow::{Context, Result};
use async_trait::async_trait;
use cups_rs::{
get_active_jobs, get_all_destinations,
job::{cancel_job, create_job_with_options, FORMAT_PDF},
Destination, PrinterState,
};
use tokio::task::spawn_blocking;
use tracing::{debug, warn};
use crate::models::{PrintDefaults, PrintParams};
#[derive(Debug, Clone)]
pub struct PrinterInfo {
pub dest: Destination,
pub state: PrinterState,
pub reasons: Vec<String>,
pub active_jobs: usize,
}
impl PrinterInfo {
pub fn healthy(&self) -> bool {
self.dest.is_accepting_jobs() && self.state.is_available() && !self.has_error_reason()
}
fn has_error_reason(&self) -> bool {
self.reasons.iter().any(|r| {
let lower = r.to_ascii_lowercase();
lower.contains("error")
|| lower.contains("offline")
|| lower.contains("jam")
|| lower.contains("paper-out")
|| lower.contains("paused")
})
}
}
#[derive(Debug, Clone)]
pub struct PrintJobResult {
pub job_id: i32,
pub printer: String,
}
#[async_trait]
pub trait PrintAdapter: Send + Sync {
async fn list_printers(&self) -> Result<Vec<PrinterInfo>>;
async fn submit_job(
&self,
printer: &Destination,
file_path: &Path,
params: &PrintParams,
defaults: &PrintDefaults,
) -> Result<PrintJobResult>;
async fn cancel(&self, job_id: i32) -> Result<()>;
}
#[derive(Debug, Default)]
pub struct CupsClient;
#[async_trait]
impl PrintAdapter for CupsClient {
async fn list_printers(&self) -> Result<Vec<PrinterInfo>> {
spawn_blocking(|| {
let destinations = get_all_destinations().context("获取打印机列表失败")?;
let mut infos = Vec::with_capacity(destinations.len());
for dest in destinations {
let state = dest.state();
let reasons = dest.state_reasons();
let active_jobs = get_active_jobs(Some(dest.name.as_str()))
.map(|jobs| jobs.len())
.unwrap_or(0);
infos.push(PrinterInfo {
dest,
state,
reasons,
active_jobs,
});
}
Ok::<_, anyhow::Error>(infos)
})
.await
.context("获取打印机失败")?
}
async fn submit_job(
&self,
printer: &Destination,
file_path: &Path,
params: &PrintParams,
defaults: &crate::models::PrintDefaults,
) -> Result<PrintJobResult> {
let dest = printer.clone();
let path = file_path.to_path_buf();
let options = params.to_print_options(defaults);
let job_name = params.effective_job_name();
spawn_blocking(move || {
let job = create_job_with_options(&dest, &job_name, &options)
.context("创建打印作业失败")?;
match job.submit_file(path.as_path(), FORMAT_PDF) {
Ok(_) => Ok(PrintJobResult {
job_id: job.id,
printer: dest.name.clone(),
}),
Err(e) => {
warn!("提交到打印机 {} 失败,尝试取消: {}", dest.name, e);
let _ = cancel_job(job.id);
Err(e).context("提交打印文件失败")
}
}
})
.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>;

502
src/routes.rs Normal file
View File

@@ -0,0 +1,502 @@
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)
}
}

413
src/scheduler.rs Normal file
View File

@@ -0,0 +1,413 @@
use std::{
path::Path,
sync::atomic::{AtomicUsize, Ordering},
time::Duration,
};
use anyhow::{anyhow, Result};
use cups_rs::PrinterState;
use tokio::sync::RwLock;
use tokio::time::sleep;
use tracing::{info, warn};
use crate::{
config::{LoadBalanceStrategy, RetryConfig},
models::{PrintDefaults, PrintParams},
printer::{PrintAdapter, 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,
target_wait_minutes: Option<f64>,
printer_speeds: std::collections::HashMap<String, crate::config::PrinterSpeedOverride>,
}
impl Scheduler {
pub fn new(
strategy: LoadBalanceStrategy,
refresh_interval_secs: u64,
adapter: SharedPrinterAdapter,
default_ppm_bw: u32,
default_ppm_color: u32,
target_wait_minutes: Option<f64>,
printer_speeds: std::collections::HashMap<String, crate::config::PrinterSpeedOverride>,
) -> Self {
let cache = std::sync::Arc::new(RwLock::new(Vec::new()));
let refresh_interval = Duration::from_secs(refresh_interval_secs.max(1));
let this = Self {
adapter: adapter.clone(),
strategy,
rr_cursor: AtomicUsize::new(0),
refresh_interval,
cache: cache.clone(),
default_ppm_bw,
default_ppm_color,
target_wait_minutes,
printer_speeds,
};
// 启动后台刷新循环
tokio::spawn(Self::refresh_loop(
adapter,
cache,
refresh_interval,
));
// 启动时立即拉取一次,填充缓存
let init_adapter = this.adapter.clone();
let init_cache = this.cache.clone();
tokio::spawn(async move {
match init_adapter.list_printers().await {
Ok(list) => {
*init_cache.write().await = list;
info!("启动时已获取打印机列表并写入缓存");
}
Err(e) => warn!("启动时获取打印机列表失败: {}", e),
}
});
this
}
async fn healthy_printers(&self) -> Result<Vec<PrinterInfo>> {
let mut printers = self.cache.read().await.clone();
if printers.is_empty() {
match self.adapter.list_printers().await {
Ok(list) => {
*self.cache.write().await = list.clone();
printers = list;
}
Err(e) => {
warn!("获取打印机失败: {}", e);
}
}
}
Ok(printers
.into_iter()
.filter(|p| p.healthy())
.collect::<Vec<_>>())
}
fn order_printers(&self, printers: &[PrinterInfo]) -> Vec<PrinterInfo> {
match self.strategy {
LoadBalanceStrategy::RoundRobin => {
if printers.is_empty() {
return Vec::new();
}
let start = self.rr_cursor.fetch_add(1, Ordering::Relaxed) % printers.len();
printers
.iter()
.cycle()
.skip(start)
.take(printers.len())
.cloned()
.collect()
}
LoadBalanceStrategy::LeastQueued => {
let mut sorted = printers.to_vec();
sorted.sort_by(|a, b| {
a.active_jobs
.cmp(&b.active_jobs)
.then(Self::state_rank(&a.state).cmp(&Self::state_rank(&b.state)))
});
sorted
}
}
}
pub async fn dispatch(
&self,
pdf_path: &Path,
params: &PrintParams,
defaults: &PrintDefaults,
retry: &RetryConfig,
) -> Result<(PrintJobResult, usize)> {
self.dispatch_with_target(pdf_path, params, defaults, retry, None)
.await
}
pub async fn dispatch_with_target(
&self,
pdf_path: &Path,
params: &PrintParams,
defaults: &PrintDefaults,
retry: &RetryConfig,
target_printer: Option<&str>,
) -> Result<(PrintJobResult, usize)> {
let printers = self.healthy_printers().await?;
if printers.is_empty() {
return Err(anyhow!("没有可用的打印机"));
}
let requested_color = params.color.unwrap_or(defaults.color);
let ordered = if let Some(target) = target_printer {
let filtered: Vec<PrinterInfo> = printers
.into_iter()
.filter(|p| p.dest.name == target)
.collect();
if filtered.is_empty() {
return Err(anyhow!("指定的打印机不可用或不存在: {}", target));
}
filtered
} else {
self.order_printers(&printers)
};
let ordered: Vec<PrinterInfo> = ordered
.into_iter()
.filter(|p| match requested_color {
crate::models::ColorModeSetting::Color => p.color_supported,
_ => true,
})
.collect();
if ordered.is_empty() {
return Err(anyhow!("没有满足色彩需求的打印机"));
};
let mut ordered = ordered;
ordered.sort_by(|a, b| {
let ca = self.cost_estimate(a, requested_color);
let cb = self.cost_estimate(b, requested_color);
ca.partial_cmp(&cb).unwrap_or(std::cmp::Ordering::Equal)
});
let max_attempts = retry.attempts.max(1).min(ordered.len());
let mut last_err = None;
for (idx, printer) in ordered.into_iter().enumerate().take(max_attempts) {
info!(
printer = %printer.dest.name,
strategy = ?self.strategy,
attempt = idx + 1,
"提交打印任务"
);
match self
.adapter
.submit_job(&printer.dest, pdf_path, params, defaults)
.await
{
Ok(job) => return Ok((job, idx)),
Err(e) => {
warn!(
printer = %printer.dest.name,
error = %e,
"提交失败,尝试下一个打印机"
);
last_err = Some(e);
if idx + 1 < max_attempts {
sleep(retry.backoff()).await;
}
}
}
}
Err(anyhow!(
"所有打印机提交失败: {}",
last_err
.map(|e| e.to_string())
.unwrap_or_else(|| "未知原因".to_string())
))
}
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,
PrinterState::Processing => 1,
PrinterState::Stopped => 2,
PrinterState::Unknown => 3,
}
}
async fn refresh_loop(
adapter: SharedPrinterAdapter,
cache: std::sync::Arc<RwLock<Vec<PrinterInfo>>>,
interval: Duration,
) {
loop {
match adapter.list_printers().await {
Ok(list) => {
*cache.write().await = list;
}
Err(e) => warn!("刷新打印机列表失败: {}", e),
}
sleep(interval).await;
}
}
fn effective_ppm(&self, printer: &PrinterInfo, color: crate::models::ColorModeSetting) -> f64 {
// override by name
if let Some(override_speed) = self.printer_speeds.get(&printer.dest.name) {
if matches!(color, crate::models::ColorModeSetting::Color) {
if let Some(ppm) = override_speed.ppm_color {
return ppm.max(1) as f64;
}
}
if let Some(ppm) = override_speed.ppm {
return ppm.max(1) as f64;
}
}
if matches!(color, crate::models::ColorModeSetting::Color) {
if let Some(ppm) = printer.ppm_color {
return ppm.max(1) as f64;
}
}
if let Some(ppm) = printer.ppm {
return ppm.max(1) as f64;
}
// fall back to defaults
if matches!(color, crate::models::ColorModeSetting::Color) {
return self.default_ppm_color.max(1) as f64;
}
self.default_ppm_bw.max(1) as f64
}
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
}
}
#[cfg(test)]
mod tests {
use super::*;
use anyhow::Result;
use async_trait::async_trait;
use cups_rs::{Destination, PrinterState};
use std::sync::Arc;
use tokio::runtime::Runtime;
#[derive(Clone)]
struct FakeAdapter;
#[async_trait]
impl PrintAdapter for FakeAdapter {
async fn list_printers(&self) -> Result<Vec<PrinterInfo>> {
Ok(vec![])
}
async fn submit_job(
&self,
_printer: &Destination,
_file_path: &Path,
_params: &PrintParams,
_defaults: &PrintDefaults,
) -> Result<PrintJobResult> {
unreachable!()
}
async fn cancel(&self, _job_id: i32) -> Result<()> {
Ok(())
}
}
fn mock_printer(name: &str, active_jobs: usize) -> PrinterInfo {
PrinterInfo {
dest: Destination {
name: name.to_string(),
instance: None,
is_default: false,
options: Default::default(),
},
state: PrinterState::Idle,
reasons: vec![],
active_jobs,
color_supported: true,
color_modes_supported: vec!["color".to_string(), "monochrome".to_string()],
ppm: Some(30),
ppm_color: Some(20),
}
}
#[test]
fn test_least_queue_sorting() {
let rt = Runtime::new().unwrap();
let ordered = rt.block_on(async {
let adapter: SharedPrinterAdapter = Arc::new(FakeAdapter);
let scheduler = Scheduler::new(
LoadBalanceStrategy::LeastQueued,
60,
adapter,
30,
20,
None,
std::collections::HashMap::new(),
);
scheduler.order_printers(&[
mock_printer("b", 3),
mock_printer("a", 1),
mock_printer("c", 2),
])
});
let names: Vec<String> = ordered.into_iter().map(|p| p.dest.name).collect();
assert_eq!(names, vec!["a", "c", "b"]);
}
#[test]
fn test_round_robin_rotation() {
let rt = Runtime::new().unwrap();
let printers = vec![mock_printer("p1", 0), mock_printer("p2", 0)];
let (first, second) = rt.block_on(async {
let adapter: SharedPrinterAdapter = Arc::new(FakeAdapter);
let scheduler = Scheduler::new(
LoadBalanceStrategy::RoundRobin,
60,
adapter,
30,
20,
None,
std::collections::HashMap::new(),
);
let first = scheduler.order_printers(&printers);
let second = scheduler.order_printers(&printers);
(first, second)
});
assert_eq!(first[0].dest.name, "p1");
assert_eq!(second[0].dest.name, "p2");
}
}