cloudtoid_interprocess/
lib.rs

1#![warn(missing_docs)]
2#![doc = include_str!("../README.md")]
3
4//! Shared-memory byte queues compatible with Cloudtoid.Interprocess protocol v3.
5//!
6//! Publishers reserve concurrently. Readers serialize consumption. A paused live
7//! participant retains ownership; recovery only reclaims proven-abandoned work.
8//! The queue is transient: after the last endpoint closes or exits, unread
9//! messages are lost. Reopening the same name creates a fresh, empty queue.
10
11#[cfg(not(all(
12    target_pointer_width = "64",
13    target_endian = "little",
14    target_has_atomic = "64"
15)))]
16compile_error!("Protocol v3 requires a little-endian 64-bit target with native 64-bit atomics");
17
18mod platform;
19mod queue;
20
21use std::{fmt, io, path::PathBuf};
22
23pub use queue::{Publisher, Subscriber, MAX_PUBLISHERS};
24
25/// Queue identity and message-buffer capacity. Every participant must agree.
26#[derive(Clone, Debug)]
27#[non_exhaustive]
28pub struct Options {
29    /// Queue name; use at most 24 UTF-8 bytes for portability.
30    pub name: String,
31    /// Shared storage directory on Unix; ignored on Windows.
32    pub path: PathBuf,
33    /// Message buffer bytes, excluding metadata; greater than 16 and divisible by 8.
34    pub capacity: usize,
35}
36
37impl Options {
38    /// Uses the operating system temporary directory for shared storage.
39    pub fn new(name: impl Into<String>, capacity: usize) -> Self {
40        Self {
41            name: name.into(),
42            path: std::env::temp_dir(),
43            capacity,
44        }
45    }
46
47    /// Selects a shared storage directory on Unix.
48    pub fn with_path(mut self, path: impl Into<PathBuf>) -> Self {
49        self.path = path.into();
50        self
51    }
52
53    fn validate(&self) -> Result<()> {
54        if cfg!(target_os = "macos") && self.name.len() > 24 {
55            return Err(Error::Invalid(
56                "queue name exceeds the macOS limit of 24 UTF-8 bytes",
57            ));
58        }
59        if cfg!(target_os = "linux") && self.name.len() > 245 {
60            return Err(Error::Invalid(
61                "queue name exceeds the Linux limit of 245 UTF-8 bytes",
62            ));
63        }
64        if self.name.is_empty()
65            || self.name.contains(['\0', '/'])
66            || (cfg!(windows) && self.name.contains('\\'))
67            || self.name == "."
68            || self.name == ".."
69        {
70            return Err(Error::Invalid("queue name must be a nonempty file name"));
71        }
72        if self.capacity <= 16 || !self.capacity.is_multiple_of(8) {
73            return Err(Error::Invalid(
74                "capacity must exceed 16 bytes and be a multiple of 8",
75            ));
76        }
77        if self
78            .capacity
79            .checked_add(queue::BUFFER_OFFSET)
80            .is_none_or(|n| n > isize::MAX as usize)
81        {
82            return Err(Error::Invalid("queue mapping is too large"));
83        }
84        Ok(())
85    }
86}
87
88/// Queue failures; full queues are retryable, while corruption requires a fresh queue.
89#[derive(Debug)]
90#[non_exhaustive]
91pub enum Error {
92    /// The queue has insufficient space, or recovery temporarily closed admission.
93    Full,
94    /// Invalid queue configuration or message length.
95    Invalid(&'static str),
96    /// An existing queue has a different capacity.
97    CapacityMismatch,
98    /// All publisher registrations are occupied.
99    PublisherLimit,
100    /// A lifetime counter cannot advance without overflowing.
101    Exhausted,
102    /// The shared queue contains an invalid record or registration.
103    Corrupt,
104    /// An operating system operation failed.
105    Io(io::Error),
106}
107
108impl Error {
109    /// Whether space or recovery admission is temporarily unavailable.
110    pub fn is_full(&self) -> bool {
111        matches!(self, Self::Full)
112    }
113}
114
115impl fmt::Display for Error {
116    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
117        match self {
118            Self::Full => f.write_str("queue is full or temporarily unavailable during recovery"),
119            Self::Invalid(message) => f.write_str(message),
120            Self::CapacityMismatch => f.write_str("capacity does not match the existing queue"),
121            Self::PublisherLimit => f.write_str("the queue already has 2048 connected publishers"),
122            Self::Exhausted => f.write_str("queue lifetime counter exhausted; use a fresh queue"),
123            Self::Corrupt => f.write_str("corrupt or inconsistent shared queue state"),
124            Self::Io(error) => error.fmt(f),
125        }
126    }
127}
128
129impl std::error::Error for Error {
130    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
131        match self {
132            Self::Io(error) => Some(error),
133            _ => None,
134        }
135    }
136}
137impl From<io::Error> for Error {
138    fn from(error: io::Error) -> Self {
139        Self::Io(error)
140    }
141}
142/// Result of a queue operation.
143pub type Result<T> = std::result::Result<T, Error>;
cloudtoid_interprocess/lib.rs

cloudtoid_interprocess/
lib.rs
