use std::cell::RefCell;
use std::rc::Rc;
use crate::component::{Component, ComponentNode};
use crate::port::{bind_or_panic, sink_of, PortName, PortOwner, PublishIf, SinkHandle};
struct HubInner<T: 'static> {
subs: RefCell<Vec<Rc<dyn SinkHandle<T>>>>,
}
impl<T: 'static> HubInner<T> {
fn broadcast(&self, item: &T) {
let subs: Vec<Rc<dyn SinkHandle<T>>> = self.subs.borrow().clone();
for sub in subs {
sub.deliver(item);
}
}
}
impl<T: 'static> PublishIf<T> for HubInner<T> {
fn write(&self, item: &T) {
self.broadcast(item);
}
}
pub struct PublishExport<T: 'static> {
inner: Rc<HubInner<T>>,
}
impl<T: 'static> PublishExport<T> {
pub fn connect(&self, owner: &dyn PortOwner, name: PortName<dyn PublishIf<T>>) {
bind_or_panic(owner, name, self.inner.clone() as Rc<dyn PublishIf<T>>);
}
}
pub struct SubscribeExport<T: 'static> {
inner: Rc<HubInner<T>>,
}
impl<T: 'static> SubscribeExport<T> {
pub fn connect(&self, owner: &dyn PortOwner, name: PortName<dyn SinkHandle<T>>) {
match sink_of(owner, name) {
Ok(sink) => self.inner.subs.borrow_mut().push(sink),
Err(e) => panic!("{e}"),
}
}
}
pub struct AnalysisBus<T: 'static> {
inner: Rc<HubInner<T>>,
}
impl<T: 'static> Clone for AnalysisBus<T> {
fn clone(&self) -> Self {
AnalysisBus { inner: self.inner.clone() }
}
}
impl<T: 'static> Default for AnalysisBus<T> {
fn default() -> Self {
AnalysisBus::new()
}
}
impl<T: 'static> AnalysisBus<T> {
pub fn new() -> AnalysisBus<T> {
AnalysisBus { inner: Rc::new(HubInner { subs: RefCell::new(Vec::new()) }) }
}
pub fn pub_export(&self) -> PublishExport<T> {
PublishExport { inner: self.inner.clone() }
}
pub fn sub_export(&self) -> SubscribeExport<T> {
SubscribeExport { inner: self.inner.clone() }
}
pub fn write(&self, item: &T) {
self.inner.broadcast(item);
}
pub fn subscriber_count(&self) -> usize {
self.inner.subs.borrow().len()
}
}
impl<T: 'static> Component for AnalysisBus<T> {}
impl<T: 'static> ComponentNode for AnalysisBus<T> {
fn node_name(&self) -> &'static str {
"AnalysisBus"
}
fn children_mut(&mut self) -> Vec<(String, &mut (dyn ComponentNode + 'static))> {
Vec::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::port::{
PortField, PortName, PortOwner, PublishPort, SinkHandle, SubscribePort, Subscriber,
};
use crate::shared::RustdvShared;
use std::any::Any;
#[derive(Default)]
struct Tally {
seen: Vec<u8>,
}
impl Subscriber<u8> for Tally {
fn write(&mut self, item: &u8) {
self.seen.push(*item);
}
}
struct Source {
ap: PublishPort<u8>,
}
impl Source {
const AP: PortName<dyn PublishIf<u8>> = PortName::new("ap");
}
impl PortOwner for Source {
fn owner_port_slot(&self, name: &str) -> Option<Rc<dyn Any>> {
(name == "ap").then(|| self.ap.slot_any())
}
fn owner_label(&self) -> &'static str {
"Source"
}
}
struct Listener {
input: SubscribePort<u8>,
tally: RustdvShared<Tally>,
}
impl Listener {
const INPUT: PortName<dyn SinkHandle<u8>> = PortName::new("input");
fn new() -> Listener {
let l = Listener { input: SubscribePort::default(), tally: RustdvShared::default() };
l.input.subscribe(l.tally.clone());
l
}
}
impl PortOwner for Listener {
fn owner_port_slot(&self, name: &str) -> Option<Rc<dyn Any>> {
(name == "input").then(|| self.input.slot_any())
}
fn owner_label(&self) -> &'static str {
"Listener"
}
}
#[test]
fn one_write_reaches_every_subscriber() {
let bus: AnalysisBus<u8> = AnalysisBus::new();
let src = Source { ap: PublishPort::default() };
let a = Listener::new();
let b = Listener::new();
bus.pub_export().connect(&src, Source::AP);
bus.sub_export().connect(&a, Listener::INPUT);
bus.sub_export().connect(&b, Listener::INPUT);
assert_eq!(bus.subscriber_count(), 2);
src.ap.write(&7);
assert_eq!(a.tally.get().seen, vec![7]);
assert_eq!(b.tally.get().seen, vec![7], "several subscribers is what makes it a broadcast");
}
#[test]
fn subscribers_are_called_in_connection_order() {
let bus: AnalysisBus<u8> = AnalysisBus::new();
let src = Source { ap: PublishPort::default() };
let first = Listener::new();
let second = Listener::new();
bus.pub_export().connect(&src, Source::AP);
bus.sub_export().connect(&first, Listener::INPUT);
bus.sub_export().connect(&second, Listener::INPUT);
for n in 1..=3u8 {
src.ap.write(&n);
}
assert_eq!(first.tally.get().seen, vec![1, 2, 3]);
assert_eq!(second.tally.get().seen, vec![1, 2, 3]);
}
#[test]
fn the_bus_stores_nothing() {
let bus: AnalysisBus<u8> = AnalysisBus::new();
let src = Source { ap: PublishPort::default() };
bus.pub_export().connect(&src, Source::AP);
src.ap.write(&1); src.ap.write(&2);
let late = Listener::new();
bus.sub_export().connect(&late, Listener::INPUT);
assert!(late.tally.get().seen.is_empty(), "nothing was buffered for a late subscriber");
src.ap.write(&3);
assert_eq!(late.tally.get().seen, vec![3], "only what arrives after it connects");
}
#[test]
fn writing_with_no_subscribers_is_legal() {
let bus: AnalysisBus<u8> = AnalysisBus::new();
let src = Source { ap: PublishPort::default() };
bus.pub_export().connect(&src, Source::AP);
assert_eq!(bus.subscriber_count(), 0);
src.ap.write(&1); }
#[test]
fn an_unconnected_publish_port_does_not_panic() {
let src = Source { ap: PublishPort::default() };
assert!(!src.ap.has_subscribers());
src.ap.write(&1); }
#[test]
fn delivery_happens_before_write_returns() {
let bus: AnalysisBus<u8> = AnalysisBus::new();
let src = Source { ap: PublishPort::default() };
let sub = Listener::new();
bus.pub_export().connect(&src, Source::AP);
bus.sub_export().connect(&sub, Listener::INPUT);
src.ap.write(&5);
assert_eq!(sub.tally.get().seen, vec![5], "already delivered, no scheduling in between");
}
#[test]
#[should_panic(expected = "has no subscriber")]
fn connecting_a_subscriber_with_no_sink_is_a_named_error() {
let bus: AnalysisBus<u8> = AnalysisBus::new();
let bare = Listener { input: SubscribePort::default(), tally: RustdvShared::default() };
bus.sub_export().connect(&bare, Listener::INPUT);
}
#[test]
fn a_clone_is_the_same_bus() {
let bus: AnalysisBus<u8> = AnalysisBus::new();
let other = bus.clone();
let sub = Listener::new();
other.sub_export().connect(&sub, Listener::INPUT);
assert_eq!(bus.subscriber_count(), 1, "two handles, one subscriber list");
bus.write(&4);
assert_eq!(sub.tally.get().seen, vec![4]);
}
}