agentmux_srv/backend/reactive/
poller.rs

1// Copyright 2025, Command Line Inc.
2// Copyright 2025-2026, AgentMux Corp.
3// SPDX-License-Identifier: Apache-2.0
4
5
6use std::collections::HashMap;
7use std::sync::{Mutex, RwLock};
8
9use super::handler::ReactiveHandler;
10use super::types::{PollerConfig, PollerStatus};
11use super::now_unix_millis;
12
13/// Poller state for cross-host message polling from AgentMux.
14pub struct Poller {
15    config: RwLock<PollerConfig>,
16    _handler: &'static ReactiveHandler,
17    running: Mutex<bool>,
18    poll_count: Mutex<u64>,
19    injections_count: Mutex<u64>,
20    last_poll: Mutex<Option<u64>>,
21    last_error: Mutex<Option<String>>,
22}
23
24impl Poller {
25    /// Create a new poller with the given config.
26    pub fn new(config: PollerConfig, handler: &'static ReactiveHandler) -> Self {
27        Self {
28            config: RwLock::new(config),
29            _handler: handler,
30            running: Mutex::new(false),
31            poll_count: Mutex::new(0),
32            injections_count: Mutex::new(0),
33            last_poll: Mutex::new(None),
34            last_error: Mutex::new(None),
35        }
36    }
37
38    /// Check if the poller is configured (has URL and token).
39    #[allow(dead_code)]
40    pub fn is_configured(&self) -> bool {
41        let config = self.config.read().unwrap();
42        config.muxbus_url.is_some() && config.muxbus_token.is_some()
43    }
44
45    /// Check if the poller is running.
46    pub fn is_running(&self) -> bool {
47        *self.running.lock().unwrap()
48    }
49
50    /// Get poller statistics.
51    pub fn stats(&self) -> HashMap<String, serde_json::Value> {
52        let mut m = HashMap::new();
53        m.insert(
54            "poll_count".to_string(),
55            serde_json::json!(*self.poll_count.lock().unwrap()),
56        );
57        m.insert(
58            "injections_count".to_string(),
59            serde_json::json!(*self.injections_count.lock().unwrap()),
60        );
61        m.insert(
62            "last_poll".to_string(),
63            serde_json::json!(*self.last_poll.lock().unwrap()),
64        );
65        m.insert(
66            "last_error".to_string(),
67            serde_json::json!(*self.last_error.lock().unwrap()),
68        );
69        m
70    }
71
72    /// Get poller status.
73    pub fn status(&self) -> PollerStatus {
74        let config = self.config.read().unwrap();
75        PollerStatus {
76            configured: config.muxbus_url.is_some() && config.muxbus_token.is_some(),
77            running: self.is_running(),
78            url: config.muxbus_url.clone(),
79            has_token: config.muxbus_token.is_some(),
80            poll_count: *self.poll_count.lock().unwrap(),
81            injections_count: *self.injections_count.lock().unwrap(),
82            last_poll: *self.last_poll.lock().unwrap(),
83        }
84    }
85
86    /// Reconfigure the poller with new URL and token.
87    pub fn reconfigure(&self, url: Option<String>, token: Option<String>) {
88        let mut config = self.config.write().unwrap();
89        config.muxbus_url = url;
90        config.muxbus_token = token;
91    }
92
93    /// Record a successful poll.
94    #[allow(dead_code)]
95    pub fn record_poll(&self) {
96        *self.poll_count.lock().unwrap() += 1;
97        *self.last_poll.lock().unwrap() = Some(now_unix_millis());
98        *self.last_error.lock().unwrap() = None;
99    }
100
101    /// Record a poll error.
102    #[allow(dead_code)]
103    pub fn record_error(&self, err: &str) {
104        *self.last_error.lock().unwrap() = Some(err.to_string());
105    }
106
107    /// Record injections delivered.
108    #[allow(dead_code)]
109    pub fn record_injections(&self, count: u64) {
110        *self.injections_count.lock().unwrap() += count;
111    }
112
113    /// Set the running state.
114    #[allow(dead_code)]
115    pub fn set_running(&self, running: bool) {
116        *self.running.lock().unwrap() = running;
117    }
118}
agentmux_srv/backend/reactive/poller.rs

agentmux_srv/backend/reactive/
poller.rs
