Add std::comm with DuplexStream

src/libstd/arc.rs

@@ -420,12 +420,11 @@ mod tests {
         let v = ~[1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
         let arc_v = arc::arc(v);
 
-        let p = port();
-        let c = chan(p);
+        let (c, p) = pipes::stream();
 
         do task::spawn() {
-            let p = port();
-            c.send(chan(p));
+            let p = pipes::port_set();
+            c.send(p.chan());
 
             let arc_v = p.recv();
 

src/libstd/comm.rs

@@ -0,0 +1,76 @@
+/*!
+
+Higher level communication abstractions.
+
+*/
+
+// NB: transitionary, de-mode-ing.
+#[forbid(deprecated_mode)];
+#[forbid(deprecated_pattern)];
+
+// Make sure we follow the new conventions
+#[forbid(non_camel_case_types)];
+
+import pipes::{channel, recv, chan, port, selectable};
+
+export DuplexStream;
+
+/// An extension of `pipes::stream` that allows both sending and receiving.
+struct DuplexStream<T: send, U: send> : channel<T>, recv<U>, selectable {
+    priv chan: chan<T>;
+    priv port: port<U>;
+
+    fn send(+x: T) {
+        self.chan.send(x)
+    }
+
+    fn try_send(+x: T) -> bool {
+        self.chan.try_send(x)
+    }
+
+    fn recv() -> U {
+        self.port.recv()
+    }
+
+    fn try_recv() -> option<U> {
+        self.port.try_recv()
+    }
+
+    pure fn peek() -> bool {
+        self.port.peek()
+    }
+
+    pure fn header() -> *pipes::packet_header {
+        self.port.header()
+    }
+}
+
+/// Creates a bidirectional stream.
+fn DuplexStream<T: send, U: send>()
+    -> (DuplexStream<T, U>, DuplexStream<U, T>)
+{
+    let (c2, p1) = pipes::stream();
+    let (c1, p2) = pipes::stream();
+    (DuplexStream {
+        chan: c1,
+        port: p1
+    },
+     DuplexStream {
+         chan: c2,
+         port: p2
+     })
+}
+
+#[cfg(test)]
+mod test {
+    #[test]
+    fn DuplexStream1() {
+        let (left, right) = DuplexStream();
+
+        left.send(~"abc");
+        right.send(123);
+
+        assert left.recv() == 123;
+        assert right.recv() == ~"abc";
+    }
+}

src/libstd/net_ip.rs

@@ -19,6 +19,7 @@ import create_uv_getaddrinfo_t = uv::ll::getaddrinfo_t;
 import set_data_for_req = uv::ll::set_data_for_req;
 import get_data_for_req = uv::ll::get_data_for_req;
 import ll = uv::ll;
+import comm = core::comm;
 
 export ip_addr, parse_addr_err;
 export format_addr;
@@ -85,7 +86,7 @@ enum ip_get_addr_err {
  */
 fn get_addr(++node: ~str, iotask: iotask)
         -> result::result<~[ip_addr], ip_get_addr_err> unsafe {
-    do comm::listen |output_ch| {
+    do core::comm::listen |output_ch| {
         do str::as_buf(node) |node_ptr, len| {
             log(debug, fmt!{"slice len %?", len});
             let handle = create_uv_getaddrinfo_t();