152 lines
4.3 KiB
Rust
152 lines
4.3 KiB
Rust
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);
|
|
}
|
|
}
|