0
\$\begingroup\$

I have written a sample Observer/Subscriber design pattern where, in my application there is a networ_receiver thread, mq_processor thread and 3 subscriber threads. Each of the subscriber will subscribe to multiple messages (distinguished by message ID). Upon reception of network data network_receiver, puts the data into a message queue shared between network_receiver and mq_processor. mq_processor further puts the data into relevant queues shared between mq_processor and different subscriber depending upon the subscription.

use std::sync::mpsc;
use std::thread;
use std::collections::HashMap;

fn network_receiver(tx: mpsc::Sender<Vec<u8>>) {
    println!("network receiver");
    let mut buff = [0u8; 2048];
    loop {
        // receive data from network
        let recvd_len = 128;
        let data = buff[..recvd_len].to_vec();
        
        let _ = tx.send(data);
    }
}

fn mq_processor(rx: mpsc::Receiver<Vec<u8>>, observers: HashMap<i32, Vec<mpsc::Sender::<Vec<u8>>>>) {
    println!("external mq processor");
    loop {
        let data = rx.recv().unwrap();
        if data.len() > 3 {
            let buff_slice: [u8; 4] = data.as_slice()[0..4].try_into().ok().unwrap();
            let msg_id = i32::from_be_bytes(buff_slice);
            if let Some(observers) = observers.get(&msg_id) {
                for observer in observers {
                    let _ = observer.send(data.clone());
                }
            }
        }
    }
}

fn subscriber1(name: &str, rx: mpsc::Receiver<Vec<u8>>) {
    println!("subscriber: {}", name);
    loop {
        let data = rx.recv().unwrap();
        println!("{} received data", name);
        let buff_slice: [u8; 4] = data.as_slice()[0..4].try_into().ok().unwrap();
        let msg_id = i32::from_be_bytes(buff_slice);
        
        match msg_id {
            1 => println!("{} received message with message-id {}", name, msg_id),
            2 => println!("{} received message with message-id {}", name, msg_id),
            3 => println!("{} received message with message-id {}", name, msg_id),
            4 => println!("{} received message with message-id {}", name, msg_id),
            8 => println!("{} received message with message-id {}", name, msg_id),
            9 => println!("{} received message with message-id {}", name, msg_id),
            _ => {},
        }
    }
}

fn subscriber2(name: &str, rx: mpsc::Receiver<Vec<u8>>) {
    println!("subscriber: {}", name);
    loop {
        let data = rx.recv().unwrap();
        println!("{} received data", name);
        let buff_slice: [u8; 4] = data.as_slice()[0..4].try_into().ok().unwrap();
        let msg_id = i32::from_be_bytes(buff_slice);
        
        match msg_id {
            2 => println!("{} received message with message-id {}", name, msg_id),
            4 => println!("{} received message with message-id {}", name, msg_id),
            5 => println!("{} received message with message-id {}", name, msg_id),
            7 => println!("{} received message with message-id {}", name, msg_id),
            8 => println!("{} received message with message-id {}", name, msg_id),
            9 => println!("{} received message with message-id {}", name, msg_id),
            _ => {},
        }
    }
}

fn subscriber3(name: &str, rx: mpsc::Receiver<Vec<u8>>) {
    println!("subscriber: {}", name);
    loop {
        let data = rx.recv().unwrap();
        println!("{} received data", name);
        let buff_slice: [u8; 4] = data.as_slice()[0..4].try_into().ok().unwrap();
        let msg_id = i32::from_be_bytes(buff_slice);
        
        match msg_id {
            2 => println!("{} received message with message-id {}", name, msg_id),
            3 => println!("{} received message with message-id {}", name, msg_id),
            4 => println!("{} received message with message-id {}", name, msg_id),
            5 => println!("{} received message with message-id {}", name, msg_id),
            6 => println!("{} received message with message-id {}", name, msg_id),
            9 => println!("{} received message with message-id {}", name, msg_id),
            10 => println!("{} received message with message-id {}", name, msg_id),
            _ => {},
        }
    }
}

fn main() {
    let msg_id1 = 1;
    let msg_id2 = 2;
    let msg_id3 = 3;
    let msg_id4 = 4;
    let msg_id5 = 5;
    let msg_id6 = 6;
    let msg_id7 = 7;
    let msg_id8 = 8;
    let msg_id9 = 9;
    let msg_id10 = 10;
    
    // 3 subscriber threads for the messages received from network
    let mut msg_id_subscriber_mapper: HashMap<i32, Vec<mpsc::Sender::<Vec<u8>>>> = HashMap::new();
    
    let (mq_processor_tx1, subscriber_rx1) = mpsc::channel::<Vec<u8>>();
    let (mq_processor_tx2, subscriber_rx2) = mpsc::channel::<Vec<u8>>();
    let (mq_processor_tx3, subscriber_rx3) = mpsc::channel::<Vec<u8>>();
    
    msg_id_subscriber_mapper.insert(msg_id1, vec![mq_processor_tx1.clone()]);
    msg_id_subscriber_mapper.insert(msg_id2, vec![mq_processor_tx1.clone(), mq_processor_tx2.clone(), mq_processor_tx3.clone()]);
    msg_id_subscriber_mapper.insert(msg_id3, vec![mq_processor_tx1.clone(), mq_processor_tx3.clone()]);
    msg_id_subscriber_mapper.insert(msg_id4, vec![mq_processor_tx1.clone(), mq_processor_tx2.clone(), mq_processor_tx3.clone()]);
    msg_id_subscriber_mapper.insert(msg_id5, vec![mq_processor_tx2.clone(), mq_processor_tx3.clone()]);
    msg_id_subscriber_mapper.insert(msg_id6, vec![mq_processor_tx3.clone()]);
    msg_id_subscriber_mapper.insert(msg_id7, vec![mq_processor_tx2.clone()]);
    msg_id_subscriber_mapper.insert(msg_id8, vec![mq_processor_tx1.clone(), mq_processor_tx2.clone()]);
    msg_id_subscriber_mapper.insert(msg_id9, vec![mq_processor_tx1.clone(), mq_processor_tx2.clone(), mq_processor_tx3.clone()]);
    msg_id_subscriber_mapper.insert(msg_id10, vec![mq_processor_tx3.clone()]);
    
    let subscriber1 = thread::spawn(move || subscriber1("subscriber1", subscriber_rx1));
    let subscriber2 = thread::spawn(move || subscriber2("subscriber2", subscriber_rx2));
    let subscriber3 = thread::spawn(move || subscriber3("subscriber3", subscriber_rx3));
    
    let (network_tx, mq_processor_rx) = mpsc::channel::<Vec<u8>>();
    let network_receiver_handle = thread::spawn(move || network_receiver(network_tx));
    let mq_processor_handle = thread::spawn(move || mq_processor(mq_processor_rx, msg_id_subscriber_mapper));
    
    let _ = network_receiver_handle.join();
    let _ = mq_processor_handle.join();
    let _ = subscriber1.join();
    let _ = subscriber2.join();
    let _ = subscriber3.join();
}
\$\endgroup\$
2
  • \$\begingroup\$ Are you sure the code you posted works as intended? \$\endgroup\$
    – corvus_192
    Commented May 3 at 9:34
  • \$\begingroup\$ @corvus_192 yes it works. Here is the modified code link which you can also verify play.rust-lang.org/… \$\endgroup\$
    – Harry
    Commented May 3 at 11:36

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.