pattern 3

Chapter08/pattern3.rs

@@ -3,35 +3,64 @@ use std::sync::mpsc::{channel,Sender,Receiver};
 use std::time;
 extern crate rand;
 
-fn new_ping() -> (Sender<u64>, Receiver<u64>) {
+enum Address {
+   Ping,
+   Pong
+}
+enum Message {
+   PingPlus(u64),
+   PongPlus(u64),
+}
+
+fn new_ping() -> (Sender<Message>, Receiver<(Address,Message)>) {
    let (pinginsend,pinginrecv) = channel();
    let (pingoutsend,pingoutrecv) = channel();
    let mut ping = 1;
    thread::spawn(move || {
       let t = time::Duration::from_millis(1000);
       loop {
-         let n = pinginrecv.recv().unwrap();
-         ping += n;
+         let msg = pinginrecv.recv().unwrap();
+         match msg {
+            Message::PingPlus(n) => { ping += n; },
+            _ => panic!("Unexpected message")
+         }
          println!("ping {}", ping);
          thread::sleep(t);
-         pingoutsend.send(ping).unwrap();
+         pingoutsend.send((
+            Address::Pong,
+            Message::PongPlus(ping)
+         )).unwrap();
+         pingoutsend.send((
+            Address::Pong,
+            Message::PongPlus(ping)
+         )).unwrap();
       }
    });
    (pinginsend, pingoutrecv)
 }
 
-fn new_pong() -> (Sender<u64>, Receiver<u64>) {
+fn new_pong() -> (Sender<Message>, Receiver<(Address,Message)>) {
    let (ponginsend,ponginrecv) = channel();
    let (pongoutsend,pongoutrecv) = channel();
-   let mut pong = 2;
+   let mut pong = 1;
    thread::spawn(move || {
       let t = time::Duration::from_millis(1000);
       loop {
-         let n = ponginrecv.recv().unwrap();
-         pong += n;
+         let msg = ponginrecv.recv().unwrap();
+         match msg {
+            Message::PongPlus(n) => { pong += n; },
+            _ => panic!("Unexpected message")
+         }
          println!("pong {}", pong);
          thread::sleep(t);
-         pongoutsend.send(pong).unwrap();
+         pongoutsend.send((
+            Address::Ping,
+            Message::PingPlus(pong)
+         )).unwrap();
+         pongoutsend.send((
+            Address::Ping,
+            Message::PingPlus(pong)
+         )).unwrap();
       }
    });
    (ponginsend, pongoutrecv)
@@ -40,15 +69,39 @@ fn new_pong() -> (Sender<u64>, Receiver<u64>) {
 fn main() {
    let pings = vec![new_ping(), new_ping(), new_ping()];
    let pongs = vec![new_pong(), new_pong(), new_pong()];
+
+   //Start the action
+   pings[0].0.send(Message::PingPlus(1)).unwrap();
+
+   //This thread will be the router
+   let t = time::Duration::from_millis(10);
    loop {
-      let mut d = 3;
+      let mut mail = Vec::new();
 
-      let (ref pingin,ref pingout) = pings[(rand::random::<u64>() % 3) as usize];
-      pingin.send(d).unwrap();
-      d = pingout.recv().unwrap();
+      for (_,r) in pings.iter() {
+         for (addr,msg) in r.try_iter() {
+            mail.push((addr,msg));
+         }
+      }
+      for (_,r) in pongs.iter() {
+         for (addr,msg) in r.try_iter() {
+            mail.push((addr,msg));
+         }
+      }
+
+      for (addr,msg) in mail.into_iter() {
+         match addr {
+            Address::Ping => {
+               let (ref s,_) = pings[(rand::random::<u32>() as usize) % pings.len()];
+               s.send(msg).unwrap();
+            },
+            Address::Pong => {
+               let (ref s,_) = pongs[(rand::random::<u32>() as usize) % pongs.len()];
+               s.send(msg).unwrap();
+            }
+         }
+      }
 
-      let (ref pongin,ref pongout) = pongs[(rand::random::<u64>() % 3) as usize];
-      pongin.send(d).unwrap();
-      pongout.recv().unwrap();
+      thread::sleep(t);
    }
 }