This commit is contained in:
@@ -8,6 +8,7 @@
|
||||
|
||||
pub use lumbridge_pty::{CommandConfig, PtyOptions, TerminalSize};
|
||||
use lumbridge_pty::{ExitStatus, OutputEvent, PtyError, PtySession};
|
||||
use std::collections::BTreeMap;
|
||||
use std::num::NonZeroUsize;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender, TryRecvError, TrySendError};
|
||||
@@ -113,6 +114,142 @@ pub enum RuntimeActorError {
|
||||
InvalidPollInterval,
|
||||
}
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum RuntimeRegistryError {
|
||||
#[error("a runtime session already exists for this pane")]
|
||||
SessionAlreadyExists,
|
||||
#[error("no runtime session exists for this pane")]
|
||||
SessionNotFound,
|
||||
#[error(transparent)]
|
||||
Actor(#[from] RuntimeActorError),
|
||||
}
|
||||
|
||||
/// Owns independent runtime actors by stable pane identity.
|
||||
///
|
||||
/// Layout attachment is deliberately absent from this type. Detaching a panel
|
||||
/// must not remove its actor; only an explicit shutdown terminates the session.
|
||||
pub struct RuntimeRegistry<K> {
|
||||
actors: BTreeMap<K, RuntimeActor>,
|
||||
}
|
||||
|
||||
impl<K> Default for RuntimeRegistry<K> {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
actors: BTreeMap::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<K: Ord> RuntimeRegistry<K> {
|
||||
#[must_use]
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn len(&self) -> usize {
|
||||
self.actors.len()
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.actors.is_empty()
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn contains(&self, pane: &K) -> bool {
|
||||
self.actors.contains_key(pane)
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn session_id(&self, pane: &K) -> Option<RuntimeSessionId> {
|
||||
self.actors.get(pane).map(RuntimeActor::session_id)
|
||||
}
|
||||
|
||||
#[must_use]
|
||||
pub fn process_id(&self, pane: &K) -> Option<u32> {
|
||||
self.actors.get(pane).and_then(RuntimeActor::process_id)
|
||||
}
|
||||
|
||||
/// Starts and registers one pane-owned actor.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Rejects duplicate pane ownership or propagates PTY/actor startup errors.
|
||||
pub fn spawn(
|
||||
&mut self,
|
||||
pane: K,
|
||||
command: CommandConfig,
|
||||
pty_options: PtyOptions,
|
||||
actor_options: RuntimeActorOptions,
|
||||
) -> Result<RuntimeSessionId, RuntimeRegistryError> {
|
||||
if self.actors.contains_key(&pane) {
|
||||
return Err(RuntimeRegistryError::SessionAlreadyExists);
|
||||
}
|
||||
let actor = RuntimeActor::spawn(command, pty_options, actor_options)?;
|
||||
let session_id = actor.session_id();
|
||||
self.actors.insert(pane, actor);
|
||||
Ok(session_id)
|
||||
}
|
||||
|
||||
/// Enqueues a command for one pane without blocking.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Reports a missing pane, actor disconnection, or bounded backpressure.
|
||||
pub fn try_send(&self, pane: &K, command: RuntimeCommand) -> Result<(), RuntimeRegistryError> {
|
||||
self.actors
|
||||
.get(pane)
|
||||
.ok_or(RuntimeRegistryError::SessionNotFound)?
|
||||
.try_send(command)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Polls one pane's next ordered event without blocking.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Reports a missing pane or actor disconnection.
|
||||
pub fn try_recv(&self, pane: &K) -> Result<Option<RuntimeEvent>, RuntimeRegistryError> {
|
||||
Ok(self
|
||||
.actors
|
||||
.get(pane)
|
||||
.ok_or(RuntimeRegistryError::SessionNotFound)?
|
||||
.try_recv()?)
|
||||
}
|
||||
|
||||
/// Waits for one pane's next ordered event up to `timeout`.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Reports a missing pane, timeout, or actor disconnection.
|
||||
pub fn recv_timeout(
|
||||
&self,
|
||||
pane: &K,
|
||||
timeout: Duration,
|
||||
) -> Result<RuntimeEvent, RuntimeRegistryError> {
|
||||
Ok(self
|
||||
.actors
|
||||
.get(pane)
|
||||
.ok_or(RuntimeRegistryError::SessionNotFound)?
|
||||
.recv_timeout(timeout)?)
|
||||
}
|
||||
|
||||
/// Explicitly terminates and removes one pane-owned actor.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Reports a missing pane, actor disconnection, or actor-thread panic.
|
||||
pub fn shutdown(&mut self, pane: &K) -> Result<(), RuntimeRegistryError> {
|
||||
let actor = self
|
||||
.actors
|
||||
.remove(pane)
|
||||
.ok_or(RuntimeRegistryError::SessionNotFound)?;
|
||||
actor.shutdown()?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// UI-side owner of one runtime actor and its bounded channels.
|
||||
pub struct RuntimeActor {
|
||||
session_id: RuntimeSessionId,
|
||||
@@ -411,6 +548,7 @@ fn send_fault(
|
||||
mod tests {
|
||||
use super::{
|
||||
RuntimeActor, RuntimeActorError, RuntimeActorOptions, RuntimeCommand, RuntimeEvent,
|
||||
RuntimeRegistry, RuntimeRegistryError,
|
||||
};
|
||||
use lumbridge_pty::{CommandConfig, PtyOptions, TerminalSize};
|
||||
use std::time::{Duration, Instant};
|
||||
@@ -447,6 +585,36 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
fn registry_events_until_exit(registry: &RuntimeRegistry<u8>, pane: u8) -> Vec<RuntimeEvent> {
|
||||
let deadline = Instant::now() + TEST_TIMEOUT;
|
||||
let mut events = Vec::new();
|
||||
loop {
|
||||
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||
assert!(!remaining.is_zero(), "timed out waiting for registry exit");
|
||||
let event = registry
|
||||
.recv_timeout(&pane, remaining)
|
||||
.expect("runtime registry event");
|
||||
let exited = matches!(event, RuntimeEvent::Exited { .. });
|
||||
events.push(event);
|
||||
if exited {
|
||||
return events;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn output_text(events: &[RuntimeEvent]) -> String {
|
||||
let bytes = events
|
||||
.iter()
|
||||
.filter_map(|event| match event {
|
||||
RuntimeEvent::Output { bytes, .. } => Some(bytes.as_slice()),
|
||||
_ => None,
|
||||
})
|
||||
.flatten()
|
||||
.copied()
|
||||
.collect::<Vec<_>>();
|
||||
String::from_utf8_lossy(&bytes).replace('\r', "")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn emits_started_ordered_output_and_exit() {
|
||||
let actor = actor("printf 'first\\n'; printf 'second\\n'; exit 7");
|
||||
@@ -555,4 +723,97 @@ mod tests {
|
||||
Err(RuntimeActorError::InvalidPollInterval)
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn registry_routes_input_and_output_by_pane() {
|
||||
let mut registry = RuntimeRegistry::new();
|
||||
let options = RuntimeActorOptions::default();
|
||||
let left_id = registry
|
||||
.spawn(
|
||||
1,
|
||||
shell("IFS= read -r line; printf 'left:%s\\n' \"$line\""),
|
||||
PtyOptions::default(),
|
||||
options,
|
||||
)
|
||||
.unwrap();
|
||||
let right_id = registry
|
||||
.spawn(
|
||||
2,
|
||||
shell("IFS= read -r line; printf 'right:%s\\n' \"$line\""),
|
||||
PtyOptions::default(),
|
||||
options,
|
||||
)
|
||||
.unwrap();
|
||||
assert_ne!(left_id, right_id);
|
||||
assert_eq!(registry.len(), 2);
|
||||
|
||||
registry
|
||||
.try_send(&1, RuntimeCommand::Input(b"alpha\n".to_vec()))
|
||||
.unwrap();
|
||||
registry
|
||||
.try_send(&2, RuntimeCommand::Input(b"beta\n".to_vec()))
|
||||
.unwrap();
|
||||
|
||||
let left = output_text(®istry_events_until_exit(®istry, 1));
|
||||
let right = output_text(®istry_events_until_exit(®istry, 2));
|
||||
assert!(left.contains("left:alpha"), "unexpected output: {left:?}");
|
||||
assert!(!left.contains("beta"), "cross-pane output: {left:?}");
|
||||
assert!(right.contains("right:beta"), "unexpected output: {right:?}");
|
||||
assert!(!right.contains("alpha"), "cross-pane output: {right:?}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn registry_rejects_duplicates_and_missing_panes() {
|
||||
let mut registry = RuntimeRegistry::new();
|
||||
registry
|
||||
.spawn(
|
||||
1,
|
||||
shell("sleep 30"),
|
||||
PtyOptions::default(),
|
||||
RuntimeActorOptions::default(),
|
||||
)
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
registry.spawn(
|
||||
1,
|
||||
shell("exit 0"),
|
||||
PtyOptions::default(),
|
||||
RuntimeActorOptions::default(),
|
||||
),
|
||||
Err(RuntimeRegistryError::SessionAlreadyExists)
|
||||
));
|
||||
assert!(matches!(
|
||||
registry.try_send(&2, RuntimeCommand::CloseInput),
|
||||
Err(RuntimeRegistryError::SessionNotFound)
|
||||
));
|
||||
assert!(matches!(
|
||||
registry.try_recv(&2),
|
||||
Err(RuntimeRegistryError::SessionNotFound)
|
||||
));
|
||||
registry.shutdown(&1).unwrap();
|
||||
assert!(registry.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn shutting_down_one_registry_entry_leaves_its_sibling_running() {
|
||||
let mut registry = RuntimeRegistry::new();
|
||||
for pane in [1, 2] {
|
||||
registry
|
||||
.spawn(
|
||||
pane,
|
||||
shell("IFS= read -r line; printf 'got:%s\\n' \"$line\""),
|
||||
PtyOptions::default(),
|
||||
RuntimeActorOptions::default(),
|
||||
)
|
||||
.unwrap();
|
||||
}
|
||||
registry.shutdown(&1).unwrap();
|
||||
assert!(!registry.contains(&1));
|
||||
assert!(registry.contains(&2));
|
||||
registry
|
||||
.try_send(&2, RuntimeCommand::Input(b"still-running\n".to_vec()))
|
||||
.unwrap();
|
||||
let events = registry_events_until_exit(®istry, 2);
|
||||
assert!(output_text(&events).contains("got:still-running"));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user