pubsub.rs

use std::hash::Hash;
use std::sync::mpsc::{self, Sender, Receiver};
use std::collections::HashMap;

#[cfg(test)]
mod test {
    #[test]
    fn simple_network() {
        use super::*;

        let mut builder = Publisher::new();
        let subscriber = builder.add_subscriber(&["widgets"]);
        let publisher = builder.build();

        publisher.publish("widgets", "sprocket");

        for (topic, content) in subscriber.fetch() {
            println!("{}: {}", topic, content);
        }
    }
}

pub struct Subscriber<Topic, Content> {
    inbox: Receiver<(Topic, Content)>,
}

impl<Topic, Content> Subscriber<Topic, Content> {
    pub fn fetch(&self) -> Vec<(Topic, Content)> {
        let mut messages = vec![];
        while let Ok(message) = self.inbox.try_recv() {
            messages.push(message);
        }
        messages
    }
}

#[derive(Clone)]
pub struct Publisher<Topic: Hash + Eq + Clone, Content: Clone> {
    subscribers: HashMap<Topic, Vec<Sender<(Topic, Content)>>>,
}

impl<Topic: Hash + Eq + Clone, Content: Clone> Publisher<Topic, Content> {
    pub fn new() -> Builder<Topic, Content> {
        Builder {
            publisher: Publisher {
                subscribers: HashMap::new()
            }
        }
    }

    pub fn publish(&self, topic: Topic, content: Content) {
        let outbox = match self.subscribers.get(&topic) {
            Some(o) => o,
            None => return,
        };

        for subscriber in outbox {
            subscriber.send((topic.clone(), content.clone())).unwrap_or(());
        }
    }
}

pub struct Builder<Topic: Hash + Eq + Clone, Content: Clone> {
    publisher: Publisher<Topic, Content>,
}

impl<Topic: Hash + Eq + Clone, Content: Clone> Builder<Topic, Content> {
    pub fn add_subscriber(&mut self, topics: &[Topic]) -> Subscriber<Topic, Content> {
        let (tx, rx) = mpsc::channel();
        for topic in topics {
            let topic = topic.clone();
            let subscriber_list = self.publisher.subscribers.entry(topic);
            subscriber_list.or_insert_with(|| Vec::new()).push(tx.clone());
        }

        Subscriber { inbox: rx }
    }

    pub fn build(self) -> Publisher<Topic, Content> {
        self.publisher
    }
}