Added peek for pipes.

src/libcore/pipes.rs

@@ -144,6 +144,15 @@ fn recv<T: send>(-p: recv_packet<T>) -> option<T> {
     }
 }
 
+/// Returns true if messages are available.
+fn peek<T: send>(p: recv_packet<T>) -> bool {
+    alt p.header().state {
+      empty { false }
+      blocked { fail "peeking on blocked packet" }
+      full | terminated { true }
+    }
+}
+
 fn sender_terminate<T: send>(p: *packet<T>) {
     let p = unsafe { uniquify(p) };
     alt swap_state_rel(p.header.state, terminated) {
@@ -337,6 +346,20 @@ fn unwrap() -> *packet<T> {
         p <-> self.p;
         option::unwrap(p)
     }
+
+    fn header() -> &self.packet_header {
+        alt self.p {
+          some(packet) {
+            unsafe {
+                let packet = uniquify(packet);
+                let header = reinterpret_cast(&packet.header);
+                forget(packet);
+                header
+            }
+          }
+          none { fail "packet already consumed" }
+        }
+    }
 }
 
 fn entangle<T: send>() -> (send_packet<T>, recv_packet<T>) {

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

+// xfail-pretty
+
+use std;
+import std::timer::sleep;
+import std::uv;
+
+proto! oneshot {
+    waiting:send {
+        signal -> signaled
+    }
+
+    signaled:send { }
+}
+
+fn main() {
+    let (c, p) = oneshot::init();
+
+    assert !pipes::peek(p);
+
+    oneshot::client::signal(c);
+
+    assert pipes::peek(p);
+}