asyn-rs 0.24.3

//! PortRuntime: promoted PortActor with event emission and graceful shutdown.

use std::sync::Arc;

use tokio::sync::{broadcast, mpsc};

use crate::interrupt::InterruptManager;
use crate::port::PortDriver;
use crate::port_actor::PortActor;
use crate::port_handle::PortHandle;
use crate::transport::InProcessClient;

use super::config::RuntimeConfig;
use super::event::RuntimeEvent;

/// Handle to a running PortRuntime. Provides shutdown and event subscription.
///
/// **Dropping this handle does not stop the port.** A port stops only when it
/// is explicitly shut down ([`Self::shutdown`], [`Self::shutdown_and_wait`]) or
/// when nothing can reach it any more — that is, when its last [`PortHandle`]
/// is dropped. So publishing a port's `PortHandle` (to the
/// [`crate::asyn_record`] registry, a [`crate::manager::PortManager`], a
/// driver) is by itself enough to keep the port alive for as long as that
/// publication lives; no caller has to park the runtime handle in a static to
/// stop the actor thread from dying underneath it.
#[derive(Clone)]
pub struct PortRuntimeHandle {
    port_handle: PortHandle,
    client: InProcessClient,
    event_tx: broadcast::Sender<RuntimeEvent>,
    /// Carries an explicit shutdown *request* to the actor. The actor stops on
    /// a `()` **sent** here — never on this channel closing, which merely means
    /// the last `PortRuntimeHandle` went away (see
    /// `PortActor::run_with_shutdown`).
    shutdown_tx: Arc<std::sync::Mutex<Option<mpsc::Sender<()>>>>,
    /// Receives a single () when the actor thread exits. Used by shutdown_and_wait().
    completion_rx: Arc<std::sync::Mutex<Option<std::sync::mpsc::Receiver<()>>>>,
    port_name: String,
}

impl PortRuntimeHandle {
    /// Get the underlying PortHandle for I/O operations.
    pub fn port_handle(&self) -> &PortHandle {
        &self.port_handle
    }

    /// Get an InProcessClient for protocol-based communication.
    pub fn client(&self) -> &InProcessClient {
        &self.client
    }

    /// Subscribe to runtime events.
    pub fn subscribe_events(&self) -> broadcast::Receiver<RuntimeEvent> {
        self.event_tx.subscribe()
    }

    /// Signal the runtime to shut down (non-blocking).
    ///
    /// Sends an explicit shutdown request; the actor thread exits after
    /// completing any in-progress request. Does not wait for the thread to
    /// stop. This outranks reachability: the port stops even while other
    /// `PortHandle`s (a registry entry, a device-support binding) could still
    /// submit to it, and those submissions then fail — which is the point of
    /// asking for a shutdown.
    ///
    /// Repeated calls are harmless: the request is already queued (or the actor
    /// has already gone), and both are a no-op.
    pub fn shutdown(&self) {
        // Poison-tolerant: shutdown must stay infallible even if a thread
        // panicked while holding the lock.
        let guard = self.shutdown_tx.lock().unwrap_or_else(|e| e.into_inner());
        if let Some(tx) = guard.as_ref() {
            // Capacity-1 channel: `Full` means a shutdown is already queued,
            // `Closed` means the actor has already stopped. Neither is an error.
            let _ = tx.try_send(());
        }
    }

    /// Signal shutdown and wait for the actor thread to exit.
    pub fn shutdown_and_wait(&self) {
        self.shutdown();
        let rx = self
            .completion_rx
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .take();
        if let Some(rx) = rx {
            let _ = rx.recv();
        }
    }

    /// Port name.
    pub fn port_name(&self) -> &str {
        &self.port_name
    }
}

/// The one way to wait for a port's connect — C `waitConnect`
/// (asynManager.c:3292-3336), which arms an exception handler and only then
/// blocks on the event.
///
/// Arming first is the whole point: a connect that lands between "am I
/// connected?" and "start waiting" would be missed by any caller that checked
/// the flag first. So [`Self::arm`] registers the callback, the caller may then
/// short-circuit on an already-connected port (C :3308-3311), and
/// [`Self::wait`] blocks for the rest. Both waiters in the crate — port
/// registration and the iocsh `asynWaitConnect` — go through it, so neither can
/// grow its own race.
pub struct ConnectWaiter {
    rx: std::sync::mpsc::Receiver<()>,
    services: crate::services::PortServices,
    callback: crate::exception::ExceptionCallbackId,
}

impl ConnectWaiter {
    /// Register for `port_name`'s connect exception. From here on the connect
    /// cannot be missed, only waited for.
    pub fn arm(services: &crate::services::PortServices, port_name: &str) -> Self {
        let (tx, rx) = std::sync::mpsc::channel::<()>();
        let waited_on = port_name.to_string();
        let callback = services.exceptions().add_callback(move |event| {
            if event.exception == crate::exception::AsynException::Connect
                && event.port_name == waited_on
            {
                let _ = tx.send(());
            }
        });
        Self {
            rx,
            services: services.clone(),
            callback,
        }
    }

    /// Block until the connect exception arrives or `timeout` elapses.
    /// `true` = connected.
    pub fn wait(self, timeout: std::time::Duration) -> bool {
        self.rx.recv_timeout(timeout).is_ok()
    }
}

impl Drop for ConnectWaiter {
    fn drop(&mut self) {
        self.services.exceptions().remove_callback(self.callback);
    }
}

/// Create a port runtime from a driver.
///
/// Returns:
/// - A `PortRuntimeHandle` for interacting with the runtime
/// - A `std::thread::JoinHandle` for the actor thread
///
/// The driver is moved into the actor thread (exclusive ownership).
pub fn create_port_runtime<D: PortDriver>(
    driver: D,
    config: RuntimeConfig,
) -> (PortRuntimeHandle, std::thread::JoinHandle<()>) {
    create_port_runtime_boxed(Box::new(driver), config)
}

/// Create a port runtime from a boxed driver.
pub fn create_port_runtime_boxed(
    mut driver: Box<dyn PortDriver>,
    config: RuntimeConfig,
) -> (PortRuntimeHandle, std::thread::JoinHandle<()>) {
    // The one site that binds a port to its trace configuration and exception
    // list. C does it inside `registerPort` — the sole path into the port list
    // — so no port can exist without them (asynManager.c:503, :611-637). Every
    // port creator in this crate (`PortManager::register_port`, the
    // `drvAsyn*PortConfigure` iocsh commands, driver-owned ports) funnels
    // through here, so binding here is what makes that true for the port too.
    config.services.bind(driver.base_mut());

    // C `registerInterface(asynCommonType)` — reached from `registerPort` for
    // every port — calls `initPortConnect` and then `portConnectTimerCallback`
    // (asynManager.c:2131-2136), which queues a connect at
    // `asynQueuePriorityConnect` the moment the port exists (:3252-3266). An
    // auto-connect port is therefore brought up BY REGISTRATION, not by whichever
    // record first happens to do I/O: `CNCT` reads 1 straight after
    // `drvAsynIPPortConfigure`, and a port no record ever talks to still comes up.
    //
    // Arming the actor's connect deadline is that queued request: the actor runs
    // `service_connect_timer` ahead of anything in its queue, which is what C's
    // Connect priority buys. It re-arms itself at `secondsBetweenPortConnect` on
    // failure (C :3281), so a port whose device is down keeps trying without a
    // single request ever being submitted.
    let connect_at_registration = driver.base().auto_connect && !driver.base().is_connected();
    if connect_at_registration {
        driver.base_mut().connect_retry_at = Some(std::time::Instant::now());
    }

    let port_name = driver.base().port_name.clone();
    let can_block = driver.base().flags.can_block;
    let multi_device = driver.base().flags.multi_device;
    let max_addr = driver.base().max_addr as i32;
    // The driver's interface declaration, taken once here — C's port
    // registration is where `registerInterface` is called too, and the set never
    // changes afterwards (asynManager.c:2100-2120).
    let interfaces = driver.capabilities();

    // Event broadcast
    let (event_tx, _) = broadcast::channel(256);

    // Runtime-private shutdown channel
    let (shutdown_tx, shutdown_rx) = mpsc::channel::<()>(1);

    // Completion notification (actor thread → shutdown_and_wait)
    let (completion_tx, completion_rx) = std::sync::mpsc::channel::<()>();

    // Share interrupt state (broadcast + mailboxes) so subscribers registered
    // via PortHandle receive notifications from the driver's call_param_callbacks.
    let shared_intr_state = driver.base().interrupts.shared_state();
    let handle_interrupts = Arc::new(InterruptManager::from_shared_state(shared_intr_state));

    // Actor channel
    let (tx, rx) = mpsc::channel(config.channel_capacity);
    let actor = PortActor::new(driver, rx);
    let actor_id = actor.id();

    let event_tx_clone = event_tx.clone();
    let name_clone = port_name.clone();

    // C's `waitConnect` (asynManager.c:2135, :3288-3336): registration waits on
    // the port's *connect exception* for at most `autoConnectTimeout`, so the
    // line of st.cmd after `drvAsynIPPortConfigure` already sees a live port.
    // The waiter is armed before the actor thread exists, so the connect it
    // waits for cannot fire ahead of it.
    let connect_wait =
        connect_at_registration.then(|| ConnectWaiter::arm(&config.services, &port_name));

    let join_handle = std::thread::Builder::new()
        .name(format!("asyn-runtime-{port_name}"))
        .spawn(move || {
            let _ = event_tx_clone.send(RuntimeEvent::Started {
                port_name: name_clone.clone(),
            });
            actor.run_with_shutdown(shutdown_rx);
            let _ = event_tx_clone.send(RuntimeEvent::Stopped {
                port_name: name_clone,
            });
            let _ = completion_tx.send(());
        })
        .expect("failed to spawn port runtime thread");

    if let Some(waiter) = connect_wait {
        // A timeout is not a failure: C ignores `waitConnect`'s status here and
        // the port's own retry timer carries on (asynManager.c:2135, :3281).
        let _ = waiter.wait(config.auto_connect_timeout);
    }

    let mut port_handle = PortHandle::new(tx, port_name.clone(), handle_interrupts, actor_id);
    port_handle.set_can_block(can_block);
    port_handle.set_capabilities(multi_device, max_addr);
    port_handle.set_interfaces(interfaces);
    let client = InProcessClient::new(port_handle.clone());

    let handle = PortRuntimeHandle {
        port_handle,
        client,
        event_tx,
        shutdown_tx: Arc::new(std::sync::Mutex::new(Some(shutdown_tx))),
        completion_rx: Arc::new(std::sync::Mutex::new(Some(completion_rx))),
        port_name,
    };

    (handle, join_handle)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::param::ParamType;
    use crate::port::{PortDriverBase, PortFlags};

    struct TestPort {
        base: PortDriverBase,
    }

    impl TestPort {
        fn new(name: &str) -> Self {
            let mut base = PortDriverBase::new(name, 1, PortFlags::default());
            base.create_param("VAL", ParamType::Int32).unwrap();
            base.create_param("F64", ParamType::Float64).unwrap();
            Self { base }
        }
    }

    impl PortDriver for TestPort {
        fn base(&self) -> &PortDriverBase {
            &self.base
        }
        fn base_mut(&mut self) -> &mut PortDriverBase {
            &mut self.base
        }
    }

    #[test]
    fn port_runtime_int32_roundtrip() {
        let (handle, _jh) = create_port_runtime(TestPort::new("rt_test"), RuntimeConfig::default());

        handle.port_handle().write_int32_blocking(0, 0, 42).unwrap();
        assert_eq!(handle.port_handle().read_int32_blocking(0, 0).unwrap(), 42);
    }

    #[test]
    fn port_runtime_client_roundtrip() {
        use crate::protocol::command::PortCommand;
        use crate::protocol::reply::ReplyPayload;
        use crate::protocol::request::{PortRequest, ProtocolPriority, RequestMeta};
        use crate::protocol::value::ParamValue;
        use crate::transport::RuntimeClient;

        let (handle, _jh) =
            create_port_runtime(TestPort::new("rt_client"), RuntimeConfig::default());

        let client = handle.client();

        // Write via client
        let req = PortRequest {
            meta: RequestMeta {
                request_id: 1,
                port_name: "rt_client".into(),
                addr: 0,
                reason: 0,
                timeout_ms: 5000,
                priority: ProtocolPriority::Medium,
                block_token: None,
            },
            command: PortCommand::Int32Write { value: 77 },
        };
        let reply = client.request_blocking(req).unwrap();
        assert_eq!(reply.payload, ReplyPayload::Ack);

        // Read via client
        let req = PortRequest {
            meta: RequestMeta {
                request_id: 2,
                port_name: "rt_client".into(),
                addr: 0,
                reason: 0,
                timeout_ms: 5000,
                priority: ProtocolPriority::Medium,
                block_token: None,
            },
            command: PortCommand::Int32Read,
        };
        let reply = client.request_blocking(req).unwrap();
        match reply.payload {
            ReplyPayload::Value(ParamValue::Int32(v)) => assert_eq!(v, 77),
            _ => panic!("expected Int32 value"),
        }
    }

    #[test]
    fn port_runtime_shutdown() {
        let (handle, jh) =
            create_port_runtime(TestPort::new("rt_shutdown"), RuntimeConfig::default());

        // Dropping the handle should cause the actor to stop
        drop(handle);
        let result = jh.join();
        assert!(result.is_ok());
    }

    #[test]
    fn port_runtime_explicit_shutdown() {
        let (handle, _jh) = create_port_runtime(
            TestPort::new("rt_explicit_shutdown"),
            RuntimeConfig::default(),
        );

        // Write a value first
        handle.port_handle().write_int32_blocking(0, 0, 42).unwrap();

        // Explicit shutdown should cause the actor to stop
        handle.shutdown_and_wait();
    }

    #[test]
    fn port_runtime_shutdown_while_handles_exist() {
        let (handle, _jh) = create_port_runtime(
            TestPort::new("rt_shutdown_handles"),
            RuntimeConfig::default(),
        );

        // Clone the handle (simulating other code holding a reference)
        let handle2 = handle.clone();

        // Explicit shutdown should work even with outstanding clones
        handle.shutdown_and_wait();

        // Subsequent operations on the cloned handle should fail gracefully
        let result = handle2.port_handle().write_int32_blocking(0, 0, 99);
        assert!(result.is_err());
    }

    #[test]
    fn port_runtime_event_subscription() {
        let (handle, _jh) =
            create_port_runtime(TestPort::new("rt_events"), RuntimeConfig::default());

        let mut rx = handle.subscribe_events();

        // Give the actor thread time to emit Started event
        std::thread::sleep(std::time::Duration::from_millis(10));

        // Check for started event (may or may not have been received depending on timing)
        match rx.try_recv() {
            Ok(RuntimeEvent::Started { port_name }) => {
                assert_eq!(port_name, "rt_events");
            }
            _ => {} // Timing-dependent, OK to miss
        }
    }

    #[test]
    fn port_runtime_port_name() {
        let (handle, _jh) =
            create_port_runtime(TestPort::new("named_port"), RuntimeConfig::default());
        assert_eq!(handle.port_name(), "named_port");
    }
}