msg type -> wire::push() -> update handle wire into module/service/ect

crates/common/src/wire.rs

@@ -0,0 +1,151 @@
+use std::any::Any;
+use std::fmt;
+use std::hash::{Hash, Hasher};
+use std::sync::{Arc, Mutex};
+
+use iced::futures::channel::mpsc;
+
+/// Which namespace an endpoint lives in. Module ids and service route keys
+/// are separate namespaces, so a module and a service may share a name.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
+pub enum Namespace {
+    Module,
+    Service,
+}
+
+/// A wire's sender: a module id or a service route key.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
+pub struct Endpoint {
+    pub ns: Namespace,
+    pub name: &'static str,
+}
+
+impl Endpoint {
+    pub const fn module(name: &'static str) -> Self {
+        Self {
+            ns: Namespace::Module,
+            name,
+        }
+    }
+
+    pub const fn service(name: &'static str) -> Self {
+        Self {
+            ns: Namespace::Service,
+            name,
+        }
+    }
+}
+
+/// Where a wire goes. Core owns the routing rule for each variant.
+#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)]
+pub enum Target {
+    /// One module, by id.
+    Module(&'static str),
+    /// One running service's inbox, by route key.
+    Service(&'static str),
+    /// Every module subscribed to a route key (a service's fan-out).
+    Topic(&'static str),
+}
+
+/// One typed message on the wire. Core routes by `to`; the receiver
+/// downcasts `payload` to one of *its own* types. Sender and receiver keep
+/// their own concrete types — nothing here enumerates them.
+#[derive(Clone)]
+pub struct Wire {
+    pub from: Endpoint,
+    pub to: Target,
+    pub payload: Arc<dyn Any + Send + Sync>,
+}
+
+impl Wire {
+    /// A module-local message: addressed straight back to one module.
+    pub fn module<T: Any + Send + Sync>(from: Endpoint, id: &'static str, msg: T) -> Self {
+        Self {
+            from,
+            to: Target::Module(id),
+            payload: Arc::new(msg),
+        }
+    }
+
+    /// Sent into a service's inbox.
+    pub fn service<T: Any + Send + Sync>(from: Endpoint, key: &'static str, msg: T) -> Self {
+        Self {
+            from,
+            to: Target::Service(key),
+            payload: Arc::new(msg),
+        }
+    }
+
+    /// Published by a service to every module subscribed to `key`.
+    pub fn topic<T: Any + Send + Sync>(from: Endpoint, key: &'static str, msg: T) -> Self {
+        Self {
+            from,
+            to: Target::Topic(key),
+            payload: Arc::new(msg),
+        }
+    }
+
+    /// Downcasts to the receiver's own message type.
+    pub fn downcast<T: Any + Send + Sync>(&self) -> Option<&T> {
+        self.payload.downcast_ref()
+    }
+}
+
+impl fmt::Debug for Wire {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Wire")
+            .field("from", &self.from)
+            .field("to", &self.to)
+            .field("payload", &self.payload.type_id())
+            .finish()
+    }
+}
+
+/// A service's inbound mailbox: modules send `Wire`s here, the service's
+/// subscription task reads them. Cloneable; the receiver is taken once.
+///
+/// The `Arc<Mutex<Option<..>>>` lets `Bar` keep a clone and hand the same
+/// value to `Subscription::run_with` on every rebuild without moving the
+/// single-consumer receiver out of its reach.
+#[derive(Clone)]
+pub struct Inbox {
+    key: &'static str,
+    tx: mpsc::UnboundedSender<Wire>,
+    rx: Arc<Mutex<Option<mpsc::UnboundedReceiver<Wire>>>>,
+}
+
+impl Inbox {
+    /// Creates the channel backing a service's inbox.
+    pub fn new(key: &'static str) -> Self {
+        let (tx, rx) = mpsc::unbounded();
+        Self {
+            key,
+            tx,
+            rx: Arc::new(Mutex::new(Some(rx))),
+        }
+    }
+
+    /// The service's route key, used as the `Topic` it publishes under.
+    pub fn key(&self) -> &'static str {
+        self.key
+    }
+
+    /// Sends a wire into the service; `false` if the service is gone.
+    pub fn send(&self, wire: Wire) -> bool {
+        self.tx.unbounded_send(wire).is_ok()
+    }
+
+    /// Takes the receiver; the service stream calls this once, when iced
+    /// first starts it.
+    pub fn take(&self) -> Option<mpsc::UnboundedReceiver<Wire>> {
+        self.rx.lock().unwrap().take()
+    }
+}
+
+impl Hash for Inbox {
+    /// Identity for `Subscription::run_with`: the route key, so returning
+    /// the same service subscription each pass does not restart its stream.
+    fn hash<H: Hasher>(&self, state: &mut H) {
+        self.key.hash(state);
+    }
+}