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.
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.
13 Higher level communication abstractions.
17 #[allow(missing_doc)];
21 /// An extension of `pipes::stream` that allows both sending and receiving.
22 pub struct DuplexStream<S, R> {
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 })
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) {
40 pub fn try_send(&self, x: S) -> bool {
43 pub fn recv(&self) -> R {
46 pub fn try_recv(&self) -> comm::TryRecvResult<R> {
49 pub fn recv_opt(&self) -> Option<R> {
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> }
59 impl<S: Send> SyncSender<S> {
60 pub fn send(&self, val: S) {
61 assert!(self.try_send(val), "SyncSender.send: receiving port closed");
64 /// Sends a message, or report if the receiver has closed the connection
66 pub fn try_send(&self, val: S) -> bool {
67 self.duplex_stream.try_send(val) && self.duplex_stream.recv_opt().is_some()
71 impl<R: Send> SyncReceiver<R> {
72 pub fn recv(&self) -> R {
73 self.recv_opt().expect("SyncReceiver.recv: sending channel closed")
76 pub fn recv_opt(&self) -> Option<R> {
77 self.duplex_stream.recv_opt().map(|val| {
78 self.duplex_stream.try_send(());
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) }
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 })
101 use comm::{duplex, rendezvous};
105 pub fn DuplexStream1() {
106 let (left, right) = duplex();
111 assert!(left.recv() == 123);
112 assert!(right.recv() == ~"abc");
116 pub fn basic_rendezvous_test() {
117 let (port, chan) = rendezvous();
123 assert!(port.recv() == "abc");
128 // Rendezvous streams should be able to handle any number of messages being sent
129 let (port, chan) = rendezvous();
131 for _ in range(0, 10000) { chan.send(()); }
133 for _ in range(0, 10000) { port.recv(); }
137 fn send_and_fail_and_try_recv() {
138 let (port, chan) = rendezvous();
140 chan.duplex_stream.send(()); // Can't access this field outside this module
147 fn try_send_and_recv_then_fail_before_ack() {
148 let (port, chan) = rendezvous();
150 port.duplex_stream.recv();
158 fn send_and_recv_then_fail_before_ack() {
159 let (port, chan) = rendezvous();
161 port.duplex_stream.recv();