386 lines
12 KiB
Rust
386 lines
12 KiB
Rust
mod control;
|
|
mod config;
|
|
mod connection;
|
|
mod db;
|
|
mod event;
|
|
mod handler;
|
|
mod middleware;
|
|
mod model;
|
|
mod service;
|
|
mod telemetry;
|
|
mod util;
|
|
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();
|
|
util::log::init_logger();
|
|
|
|
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);
|
|
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/{unit_id}/ack-fault",
|
|
post(handler::control::ack_fault_unit),
|
|
)
|
|
.route(
|
|
"/api/unit/{unit_id}/runtime",
|
|
get(handler::control::get_unit_runtime),
|
|
)
|
|
.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));
|
|
|
|
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();
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|