Added select2 for pipes.

Author: eholk

src/libcore/pipes.rs

@@ -1,6 +1,7 @@
 // Runtime support for pipes.
 
-import unsafe::{forget, reinterpret_cast};
+import unsafe::{forget, reinterpret_cast, transmute};
+import either::{either, left, right};
 
 enum state {
     empty,
@@ -243,6 +244,30 @@ fn wait_many(pkts: ~[&a.packet_header]) -> uint {
     ready_packet
 }
 
+fn select2<A: send, B: send>(
+    +a: recv_packet<A>,
+    +b: recv_packet<B>)
+    -> either<(option<A>, recv_packet<B>), (recv_packet<A>, option<B>)>
+{
+    let a = unsafe { uniquify(a.unwrap()) };
+    let b = unsafe { uniquify(b.unwrap()) };
+    let i = {
+        let headers = ~[&a.header,
+                        &b.header];
+        wait_many(headers)
+    };
+
+    unsafe {
+        alt i {
+          0 { left((recv(recv_packet(transmute(a))),
+                    recv_packet(transmute(b)))) }
+          1 { right((recv_packet(transmute(a)),
+                     recv(recv_packet(transmute(b))))) }
+          _ { fail "select2 return an invalid packet" }
+        }
+    }
+}
+
 #[doc = "Waits on a set of endpoints. Returns a message, its index,
  and a list of the remaining endpoints."]
 fn select<T: send>(+endpoints: ~[recv_packet<T>])

src/test/run-pass/pipe-select.rs

@@ -72,4 +72,34 @@ fn main() {
     sleep(iotask, 1000);
 
     signal(c2);
-}
+
+    test_select2();
+}
+
+fn test_select2() {
+    let (ac, ap) = stream::init();
+    let (bc, bp) = stream::init();
+
+    stream::client::send(ac, 42);
+
+    alt pipes::select2(ap, bp) {
+      either::left(*) { }
+      either::right(*) { fail }
+    }
+
+    stream::client::send(bc, "abc");
+
+    #error("done with first select2");
+
+    let (ac, ap) = stream::init();
+    let (bc, bp) = stream::init();
+
+    stream::client::send(bc, "abc");
+
+    alt pipes::select2(ap, bp) {
+      either::left(*) { fail }
+      either::right(*) { }
+    }
+
+    stream::client::send(ac, 42);
+}