zcdsj
/
plc_control
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test(workspace): verify dual-app release builds

This commit is contained in:
caoqianming 2026-04-16 13:17:41 +08:00
parent 3cc13ccf1e
commit 60452f9065
2 changed files with 13 additions and 413 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
crates/plc_platform_core/tests/event_namespace.rs
View File

@ -14,3 +14,16 @@ fn namespaced_event_types_keep_their_prefix() {
"00000000-0000-0000-0000-000000000000"
);
}
#[test]
fn event_namespaces_match_the_supported_apps() {
let supported = [
"platform.source_connected",
"feeder.auto_control_started",
"ops.unit_started",
];
for name in supported {
assert!(name.contains('.'));
}
}

413
src/main.rs
View File

@ -1,413 +0,0 @@
#![cfg_attr(all(windows, not(debug_assertions)), windows_subsystem = "windows")]
mod control;
mod config;
mod connection;
mod db;
mod event;
mod handler;
mod middleware;
mod service;
mod telemetry;
mod websocket;
use axum::{
routing::{get, post, put},
Router,
};
use config::AppConfig;
use connection::ConnectionManager;
use db::init_database;
use event::EventManager;
use middleware::simple_logger;
use std::sync::Arc;
use tokio::sync::mpsc;
use axum::{extract::Request, middleware::Next, response::Response};
use tower_http::cors::{Any, CorsLayer};
use tower_http::services::ServeDir;
async fn no_cache(req: Request, next: Next) -> Response {
let mut res = next.run(req).await;
res.headers_mut().insert(
axum::http::header::CACHE_CONTROL,
axum::http::HeaderValue::from_static("no-store"),
);
res
}
#[derive(Clone)]
pub struct AppState {
pub config: AppConfig,
pub pool: sqlx::PgPool,
pub connection_manager: Arc<ConnectionManager>,
pub event_manager: Arc<EventManager>,
pub ws_manager: Arc<websocket::WebSocketManager>,
pub control_runtime: Arc<control::runtime::ControlRuntimeStore>,
}
#[tokio::main]
async fn main() {
dotenv::dotenv().ok();
plc_platform_core::util::log::init_logger();
let _single_instance =
match plc_platform_core::util::single_instance::try_acquire("PLCControl.Gateway") {
Ok(guard) => guard,
Err(err) if err.kind() == std::io::ErrorKind::AlreadyExists => {
tracing::warn!("Another PLC Control instance is already running");
return;
}
Err(err) => {
tracing::error!("Failed to initialize single instance guard: {}", err);
return;
}
};
let config = AppConfig::from_env().expect("Failed to load configuration");
let pool = init_database(&config.database_url)
.await
.expect("Failed to initialize database");
let mut connection_manager = ConnectionManager::new();
let ws_manager = Arc::new(websocket::WebSocketManager::new());
let event_manager = Arc::new(EventManager::new(
pool.clone(),
Arc::new(connection_manager.clone()),
Some(ws_manager.clone()),
));
connection_manager.set_event_manager(event_manager.clone());
connection_manager.set_pool_and_start_reconnect_task(Arc::new(pool.clone()));
let connection_manager = Arc::new(connection_manager);
let control_runtime = Arc::new(control::runtime::ControlRuntimeStore::new());
// Connect to all enabled sources concurrently
let sources = service::get_all_enabled_sources(&pool)
.await
.expect("Failed to fetch sources");
// Spawn a task for each source to connect and subscribe concurrently
let mut tasks = Vec::new();
for source in sources {
let cm = connection_manager.clone();
let p = pool.clone();
let source_name = source.name.clone();
let source_id = source.id;
let task = tokio::spawn(async move {
if let Err(e) = cm.connect_from_source(&p, source_id).await {
tracing::error!("Failed to connect to source {}: {}", source_name, e);
}
});
tasks.push(task);
}
// Wait for all connection tasks to complete
for task in tasks {
if let Err(e) = task.await {
tracing::error!("Source connection task failed: {:?}", e);
}
}
let state = AppState {
config: config.clone(),
pool,
connection_manager: connection_manager.clone(),
event_manager,
ws_manager,
control_runtime: control_runtime.clone(),
};
control::engine::start(state.clone(), control_runtime);
if config.simulate_plc {
control::simulate::start(state.clone());
}
let app = build_router(state.clone());
let addr = format!("{}:{}", config.server_host, config.server_port);
tracing::info!("Starting server at http://{}", addr);
let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
let ui_url = format!("http://{}:{}/ui", "localhost", config.server_port);
let (shutdown_tx, mut shutdown_rx) = mpsc::channel::<()>(1);
let shutdown_tx_ctrl = shutdown_tx.clone();
let rt_handle = tokio::runtime::Handle::current();
init_tray(ui_url, shutdown_tx.clone(), rt_handle);
let connection_manager_for_shutdown = connection_manager.clone();
tokio::spawn(async move {
tokio::signal::ctrl_c()
.await
.expect("Failed to install Ctrl+C handler");
let _ = shutdown_tx_ctrl.send(()).await;
});
let shutdown_signal = async move {
let _ = shutdown_rx.recv().await;
tracing::info!("Received shutdown signal, closing all connections...");
connection_manager_for_shutdown.disconnect_all().await;
tracing::info!("All connections closed");
};
axum::serve(listener, app)
.with_graceful_shutdown(shutdown_signal)
.await
.unwrap();
}
fn build_router(state: AppState) -> Router {
let all_route = Router::new()
.route(
"/api/source",
get(handler::source::get_source_list).post(handler::source::create_source),
)
.route(
"/api/source/{source_id}",
axum::routing::delete(handler::source::delete_source)
.put(handler::source::update_source),
)
.route(
"/api/source/{source_id}/reconnect",
axum::routing::post(handler::source::reconnect_source),
)
.route(
"/api/source/{source_id}/browse",
axum::routing::post(handler::source::browse_and_save_nodes),
)
.route(
"/api/source/{source_id}/node-tree",
get(handler::source::get_node_tree),
)
.route("/api/point", get(handler::point::get_point_list))
.route(
"/api/point/value/batch",
axum::routing::post(handler::point::batch_set_point_value),
)
.route(
"/api/point/batch",
axum::routing::post(handler::point::batch_create_points)
.delete(handler::point::batch_delete_points),
)
.route(
"/api/point/{point_id}/history",
get(handler::point::get_point_history),
)
.route(
"/api/point/{point_id}",
get(handler::point::get_point)
.put(handler::point::update_point)
.delete(handler::point::delete_point),
)
.route(
"/api/point/batch/set-tags",
put(handler::point::batch_set_point_tags),
)
.route(
"/api/point/batch/set-equipment",
put(handler::point::batch_set_point_equipment),
)
.route(
"/api/equipment",
get(handler::equipment::get_equipment_list).post(handler::equipment::create_equipment),
)
.route(
"/api/equipment/{equipment_id}",
get(handler::equipment::get_equipment)
.put(handler::equipment::update_equipment)
.delete(handler::equipment::delete_equipment),
)
.route(
"/api/equipment/batch/set-unit",
put(handler::equipment::batch_set_equipment_unit),
)
.route(
"/api/equipment/{equipment_id}/points",
get(handler::equipment::get_equipment_points),
)
.route(
"/api/unit",
get(handler::control::get_unit_list).post(handler::control::create_unit),
)
.route(
"/api/unit/{unit_id}",
get(handler::control::get_unit)
.put(handler::control::update_unit)
.delete(handler::control::delete_unit),
)
.route(
"/api/event",
get(handler::control::get_event_list),
)
.route(
"/api/control/equipment/{equipment_id}/start",
post(handler::control::start_equipment),
)
.route(
"/api/control/equipment/{equipment_id}/stop",
post(handler::control::stop_equipment),
)
.route(
"/api/control/unit/{unit_id}/start-auto",
post(handler::control::start_auto_unit),
)
.route(
"/api/control/unit/{unit_id}/stop-auto",
post(handler::control::stop_auto_unit),
)
.route(
"/api/control/unit/batch-start-auto",
post(handler::control::batch_start_auto),
)
.route(
"/api/control/unit/batch-stop-auto",
post(handler::control::batch_stop_auto),
)
.route(
"/api/control/unit/{unit_id}/ack-fault",
post(handler::control::ack_fault_unit),
)
.route(
"/api/unit/{unit_id}/runtime",
get(handler::control::get_unit_runtime),
)
.route(
"/api/unit/{unit_id}/detail",
get(handler::control::get_unit_detail),
)
.route(
"/api/tag",
get(handler::tag::get_tag_list).post(handler::tag::create_tag),
)
.route(
"/api/tag/{tag_id}",
get(handler::tag::get_tag_points)
.put(handler::tag::update_tag)
.delete(handler::tag::delete_tag),
)
.route(
"/api/page",
get(handler::page::get_page_list).post(handler::page::create_page),
)
.route(
"/api/page/{page_id}",
get(handler::page::get_page)
.put(handler::page::update_page)
.delete(handler::page::delete_page),
)
.route("/api/logs", get(handler::log::get_logs))
.route("/api/logs/stream", get(handler::log::stream_logs))
.route("/api/docs/api-md", get(handler::doc::get_api_md))
.route("/api/docs/readme-md", get(handler::doc::get_readme_md));
Router::new()
.merge(all_route)
.nest(
"/ui",
Router::new()
.fallback_service(ServeDir::new("web").append_index_html_on_directories(true))
.layer(axum::middleware::from_fn(no_cache)),
)
.route("/ws/public", get(websocket::public_websocket_handler))
.route(
"/ws/client/{client_id}",
get(websocket::client_websocket_handler),
)
.layer(axum::middleware::from_fn(simple_logger))
.layer(
CorsLayer::new()
.allow_origin(Any)
.allow_methods(Any)
.allow_headers(Any),
)
.with_state(state)
}
#[cfg(windows)]
fn init_tray(ui_url: String, shutdown_tx: mpsc::Sender<()>, rt_handle: tokio::runtime::Handle) {
std::thread::spawn(move || {
if let Err(e) = tray::run_tray(ui_url, shutdown_tx, rt_handle) {
tracing::warn!("Tray init failed: {}", e);
}
});
}
#[cfg(not(windows))]
fn init_tray(_ui_url: String, _shutdown_tx: mpsc::Sender<()>, _rt_handle: tokio::runtime::Handle) {}
#[cfg(windows)]
mod tray {
use std::error::Error;
use tokio::sync::mpsc;
use tray_icon::{
menu::{Menu, MenuEvent, MenuItem},
Icon, TrayIconBuilder,
};
use winit::application::ApplicationHandler;
use winit::event_loop::{ActiveEventLoop, ControlFlow, EventLoop};
use winit::platform::windows::EventLoopBuilderExtWindows;
pub fn run_tray(
ui_url: String,
shutdown_tx: mpsc::Sender<()>,
rt_handle: tokio::runtime::Handle,
) -> Result<(), Box<dyn Error>> {
let mut builder = EventLoop::builder();
builder.with_any_thread(true);
let event_loop = builder.build()?;
let menu = Menu::new();
let open_item = MenuItem::new("Open UI", true, None);
let exit_item = MenuItem::new("Exit", true, None);
menu.append(&open_item)?;
menu.append(&exit_item)?;
let icon = Icon::from_rgba(vec![0, 120, 212, 255], 1, 1)?;
let _tray = TrayIconBuilder::new()
.with_tooltip("PLC Control")
.with_menu(Box::new(menu))
.with_icon(icon)
.build()?;
let menu_rx = MenuEvent::receiver();
let mut app = TrayApp {
menu_rx,
open_id: open_item.id().clone(),
exit_id: exit_item.id().clone(),
ui_url,
shutdown_tx,
rt_handle,
};
event_loop.run_app(&mut app).map_err(|e| e.into())
}
struct TrayApp {
menu_rx: &'static tray_icon::menu::MenuEventReceiver,
open_id: tray_icon::menu::MenuId,
exit_id: tray_icon::menu::MenuId,
ui_url: String,
shutdown_tx: mpsc::Sender<()>,
rt_handle: tokio::runtime::Handle,
}
impl ApplicationHandler for TrayApp {
fn resumed(&mut self, _event_loop: &ActiveEventLoop) {}
fn window_event(
&mut self,
_event_loop: &ActiveEventLoop,
_window_id: winit::window::WindowId,
_event: winit::event::WindowEvent,
) {
}
fn about_to_wait(&mut self, event_loop: &ActiveEventLoop) {
event_loop.set_control_flow(ControlFlow::Wait);
while let Ok(menu_event) = self.menu_rx.try_recv() {
if menu_event.id == self.open_id {
let _ = webbrowser::open(&self.ui_url);
}
if menu_event.id == self.exit_id {
let _ = self.rt_handle.block_on(self.shutdown_tx.send(()));
event_loop.exit();
}
}
}
}
}