src/libsync/comm.rs

   1 // Copyright 2012-2013 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 /*!
  12
  13 Higher level communication abstractions.
  14
  15 */
  16
  17 #[allow(missing_doc)];
  18
  19 use std::comm;
  20
  21 /// An extension of `pipes::stream` that allows both sending and receiving.
  22 pub struct DuplexStream<S, R> {
  23     priv tx: Sender<S>,
  24     priv rx: Receiver<R>,
  25 }
  26
  27 /// Creates a bidirectional stream.
  28 pub fn duplex<S: Send, R: Send>() -> (DuplexStream<S, R>, DuplexStream<R, S>) {
  29     let (tx1, rx1) = channel();
  30     let (tx2, rx2) = channel();
  31     (DuplexStream { tx: tx1, rx: rx2 },
  32      DuplexStream { tx: tx2, rx: rx1 })
  33 }
  34
  35 // Allow these methods to be used without import:
  36 impl<S:Send,R:Send> DuplexStream<S, R> {
  37     pub fn send(&self, x: S) {
  38         self.tx.send(x)
  39     }
  40     pub fn try_send(&self, x: S) -> bool {
  41         self.tx.try_send(x)
  42     }
  43     pub fn recv(&self) -> R {
  44         self.rx.recv()
  45     }
  46     pub fn try_recv(&self) -> comm::TryRecvResult<R> {
  47         self.rx.try_recv()
  48     }
  49     pub fn recv_opt(&self) -> Option<R> {
  50         self.rx.recv_opt()
  51     }
  52 }
  53
  54 /// An extension of `pipes::stream` that provides synchronous message sending.
  55 pub struct SyncSender<S> { priv duplex_stream: DuplexStream<S, ()> }
  56 /// An extension of `pipes::stream` that acknowledges each message received.
  57 pub struct SyncReceiver<R> { priv duplex_stream: DuplexStream<(), R> }
  58
  59 impl<S: Send> SyncSender<S> {
  60     pub fn send(&self, val: S) {
  61         assert!(self.try_send(val), "SyncSender.send: receiving port closed");
  62     }
  63
  64     /// Sends a message, or report if the receiver has closed the connection
  65     /// before receiving.
  66     pub fn try_send(&self, val: S) -> bool {
  67         self.duplex_stream.try_send(val) && self.duplex_stream.recv_opt().is_some()
  68     }
  69 }
  70
  71 impl<R: Send> SyncReceiver<R> {
  72     pub fn recv(&self) -> R {
  73         self.recv_opt().expect("SyncReceiver.recv: sending channel closed")
  74     }
  75
  76     pub fn recv_opt(&self) -> Option<R> {
  77         self.duplex_stream.recv_opt().map(|val| {
  78             self.duplex_stream.try_send(());
  79             val
  80         })
  81     }
  82
  83     pub fn try_recv(&self) -> comm::TryRecvResult<R> {
  84         match self.duplex_stream.try_recv() {
  85             comm::Data(t) => { self.duplex_stream.try_send(()); comm::Data(t) }
  86             state => state,
  87         }
  88     }
  89 }
  90
  91 /// Creates a stream whose channel, upon sending a message, blocks until the
  92 /// message is received.
  93 pub fn rendezvous<T: Send>() -> (SyncReceiver<T>, SyncSender<T>) {
  94     let (chan_stream, port_stream) = duplex();
  95     (SyncReceiver { duplex_stream: port_stream },
  96      SyncSender { duplex_stream: chan_stream })
  97 }
  98
  99 #[cfg(test)]
 100 mod test {
 101     use comm::{duplex, rendezvous};
 102
 103
 104     #[test]
 105     pub fn DuplexStream1() {
 106         let (left, right) = duplex();
 107
 108         left.send(~"abc");
 109         right.send(123);
 110
 111         assert!(left.recv() == 123);
 112         assert!(right.recv() == ~"abc");
 113     }
 114
 115     #[test]
 116     pub fn basic_rendezvous_test() {
 117         let (port, chan) = rendezvous();
 118
 119         spawn(proc() {
 120             chan.send("abc");
 121         });
 122
 123         assert!(port.recv() == "abc");
 124     }
 125
 126     #[test]
 127     fn recv_a_lot() {
 128         // Rendezvous streams should be able to handle any number of messages being sent
 129         let (port, chan) = rendezvous();
 130         spawn(proc() {
 131             for _ in range(0, 10000) { chan.send(()); }
 132         });
 133         for _ in range(0, 10000) { port.recv(); }
 134     }
 135
 136     #[test]
 137     fn send_and_fail_and_try_recv() {
 138         let (port, chan) = rendezvous();
 139         spawn(proc() {
 140             chan.duplex_stream.send(()); // Can't access this field outside this module
 141             fail!()
 142         });
 143         port.recv()
 144     }
 145
 146     #[test]
 147     fn try_send_and_recv_then_fail_before_ack() {
 148         let (port, chan) = rendezvous();
 149         spawn(proc() {
 150             port.duplex_stream.recv();
 151             fail!()
 152         });
 153         chan.try_send(());
 154     }
 155
 156     #[test]
 157     #[should_fail]
 158     fn send_and_recv_then_fail_before_ack() {
 159         let (port, chan) = rendezvous();
 160         spawn(proc() {
 161             port.duplex_stream.recv();
 162             fail!()
 163         });
 164         chan.send(());
 165     }
 166 }