agent_can/daemon/

server.rs

use crate::can::{CanSocket, dbc};
use crate::daemon::config::DaemonConfig;
use crate::ipc::{self, BoxedLocalStream};
use crate::protocol::{
    MailboxMessageData, MailboxSignalData, Request, RequestAction, Response, ResponseData,
};
use crate::sim::types::SimCanFrame;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, split};
use tokio::sync::{mpsc, oneshot, watch};
use tokio::task::yield_now;
use tokio::time::{Duration, sleep};

const MAX_ACTIONS_PER_TURN: usize = 16;
const POLL_INTERVAL: Duration = Duration::from_millis(10);

struct DaemonState {
    config: DaemonConfig,
    socket: CanSocket,
    dbc: dbc::DbcBusOverlay,
    mailboxes: std::collections::BTreeMap<String, MailboxMessageData>,
    rx_seen: u64,
    rx_decoded: u64,
    rx_dropped: u64,
    update_seq: u64,
    shutdown: bool,
}

struct ActionMessage {
    request: Request,
    response_tx: oneshot::Sender<Response>,
}

impl DaemonState {
    fn status(&self) -> ResponseData {
        ResponseData::Status {
            bus: self.config.bus.clone(),
            adapter: self.config.adapter.clone(),
            dbc_path: self.config.dbc_path.clone(),
            bitrate: self.config.bitrate,
            bitrate_data: self.config.bitrate_data,
            fd_capable: self.config.fd_capable,
            mailbox_count: self.mailboxes.len(),
            rx_seen: self.rx_seen,
            rx_decoded: self.rx_decoded,
            rx_dropped: self.rx_dropped,
        }
    }
}

pub async fn run_listener(
    socket_path: std::path::PathBuf,
    config: DaemonConfig,
) -> Result<(), std::io::Error> {
    if let Some(parent) = socket_path.parent() {
        std::fs::create_dir_all(parent)?;
    }

    let socket = CanSocket::open(
        &config.adapter,
        config.bitrate,
        config.bitrate_data,
        config.fd_capable,
    )
    .map_err(std::io::Error::other)?;
    let dbc = dbc::DbcBusOverlay::load(std::path::Path::new(&config.dbc_path))
        .map_err(std::io::Error::other)?;
    let mut listener = ipc::bind_listener(&socket_path).await?;
    ipc::create_endpoint_marker(&socket_path)?;
    std::fs::write(
        crate::daemon::lifecycle::pid_path(&config.bus),
        std::process::id().to_string(),
    )?;

    let state = DaemonState {
        config,
        socket,
        dbc,
        mailboxes: std::collections::BTreeMap::new(),
        rx_seen: 0,
        rx_decoded: 0,
        rx_dropped: 0,
        update_seq: 0,
        shutdown: false,
    };

    let (action_tx, action_rx) = mpsc::channel::<ActionMessage>(256);
    let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
    let actor_task = tokio::spawn(run_actor_task(state, action_rx, shutdown_tx));

    let mut listener_error = None;
    loop {
        tokio::select! {
            changed = shutdown_rx.changed() => {
                match changed {
                    Ok(()) if *shutdown_rx.borrow() => break,
                    Ok(()) => {}
                    Err(_) => break,
                }
            }
            accepted = listener.accept() => {
                match accepted {
                    Ok(stream) => {
                        let action_tx = action_tx.clone();
                        tokio::spawn(async move {
                            let _ = handle_connection(stream, action_tx).await;
                        });
                    }
                    Err(err) => {
                        listener_error = Some(err);
                        break;
                    }
                }
            }
        }
    }

    drop(action_tx);
    let _ = actor_task.await;
    ipc::cleanup_endpoint(&socket_path);
    let pid_path = crate::daemon::lifecycle::pid_path(
        socket_path
            .file_stem()
            .and_then(|value| value.to_str())
            .unwrap_or("default"),
    );
    if pid_path.exists() {
        let _ = std::fs::remove_file(pid_path);
    }

    if let Some(err) = listener_error {
        return Err(err);
    }
    Ok(())
}

async fn handle_connection(
    stream: BoxedLocalStream,
    action_tx: mpsc::Sender<ActionMessage>,
) -> Result<(), std::io::Error> {
    let (read_half, mut write_half) = split(stream);
    let mut reader = BufReader::new(read_half);
    let mut line = String::new();

    loop {
        line.clear();
        let read = reader.read_line(&mut line).await?;
        if read == 0 {
            return Ok(());
        }
        let response = match serde_json::from_str::<Request>(line.trim_end()) {
            Ok(request) => {
                let request_id = request.id;
                let (response_tx, response_rx) = oneshot::channel();
                if action_tx
                    .send(ActionMessage {
                        request,
                        response_tx,
                    })
                    .await
                    .is_err()
                {
                    Response::err(request_id, "daemon unavailable")
                } else {
                    match response_rx.await {
                        Ok(response) => response,
                        Err(_) => Response::err(request_id, "daemon unavailable"),
                    }
                }
            }
            Err(err) => Response::err(uuid::Uuid::new_v4(), format!("invalid request json: {err}")),
        };
        let mut payload = serde_json::to_string(&response).unwrap_or_else(|err| {
            format!("{{\"success\":false,\"error\":\"response serialization failed: {err}\"}}")
        });
        payload.push('\n');
        write_half.write_all(payload.as_bytes()).await?;
    }
}

async fn run_actor_task(
    mut state: DaemonState,
    mut action_rx: mpsc::Receiver<ActionMessage>,
    shutdown_tx: watch::Sender<bool>,
) {
    loop {
        process_action_batch(&mut action_rx, &mut state, MAX_ACTIONS_PER_TURN).await;

        if state.shutdown {
            let _ = shutdown_tx.send(true);
            break;
        }

        if let Err(err) = poll_can(&mut state) {
            tracing::error!("CAN poll failed for bus '{}': {err}", state.config.bus);
            state.shutdown = true;
            let _ = shutdown_tx.send(true);
            break;
        }

        yield_now().await;
        sleep(POLL_INTERVAL).await;
    }
}

async fn process_action_batch(
    action_rx: &mut mpsc::Receiver<ActionMessage>,
    state: &mut DaemonState,
    limit: usize,
) {
    for _ in 0..limit {
        let Some(message) = action_rx.try_recv().ok() else {
            break;
        };
        let response = handle_action(message.request, state);
        let _ = message.response_tx.send(response);
    }
}

fn handle_action(request: Request, state: &mut DaemonState) -> Response {
    let id = request.id;
    let result = match request.action {
        RequestAction::Status => Ok(state.status()),
        RequestAction::Mailboxes => Ok(ResponseData::Mailboxes {
            messages: state.mailboxes.values().cloned().collect(),
        }),
        RequestAction::Mailbox { message } => state
            .mailboxes
            .get(&message)
            .cloned()
            .map(|message| ResponseData::Mailbox { message })
            .ok_or_else(|| format!("message '{message}' has no decoded mailbox value")),
        RequestAction::SendRaw {
            arb_id,
            data_hex,
            flags,
        } => {
            let payload = crate::can::parse_data_hex(&data_hex);
            payload.and_then(|payload| {
                let mut data = [0_u8; 64];
                data[..payload.len()].copy_from_slice(&payload);
                let frame = SimCanFrame {
                    arb_id,
                    len: payload.len() as u8,
                    flags: flags.unwrap_or(0),
                    data,
                };
                crate::can::validate_frame(&state.config.bus, state.config.fd_capable, &frame)?;
                state.socket.send(&frame)?;
                Ok(ResponseData::SentRaw {
                    arb_id,
                    len: frame.len,
                })
            })
        }
        RequestAction::SendMessage { message, signals } => state
            .dbc
            .encode_message(&message, &signals)
            .and_then(|frame| {
                crate::can::validate_frame(&state.config.bus, state.config.fd_capable, &frame)?;
                state.socket.send(&frame)?;
                Ok(ResponseData::SentMessage {
                    message,
                    arb_id: frame.arb_id,
                    len: frame.len,
                })
            }),
        RequestAction::Close => {
            state.shutdown = true;
            Ok(ResponseData::Ack)
        }
    };

    match result {
        Ok(data) => Response::ok(id, data),
        Err(err) => Response::err(id, err),
    }
}

fn poll_can(state: &mut DaemonState) -> Result<(), String> {
    let frames = state.socket.recv_all()?;
    for frame in frames {
        state.rx_seen += 1;
        crate::can::validate_frame(&state.config.bus, state.config.fd_capable, &frame)?;
        let Some(decoded) = state.dbc.decode_message(&frame)? else {
            state.rx_dropped += 1;
            continue;
        };
        state.rx_decoded += 1;
        state.update_seq += 1;
        let message = MailboxMessageData {
            message: decoded.name.clone(),
            arb_id: frame.arb_id,
            extended: decoded.extended,
            len: frame.len,
            update_seq: state.update_seq,
            signals: decoded
                .signals
                .into_iter()
                .map(|signal| MailboxSignalData {
                    name: signal.name,
                    value: signal.value,
                    unit: signal.unit,
                })
                .collect(),
        };
        state.mailboxes.insert(decoded.name, message);
    }
    Ok(())
}