Use protocol compiler in future.rs. Also split recv into recv and try_recv.

Author: eholk

src/libcore/future.rs

@@ -12,6 +12,7 @@
  */
 
 import either::either;
+import pipes::recv;
 
 export future;
 export extensions;
@@ -60,10 +61,6 @@ fn from_value<A>(+val: A) -> future<A> {
 }
 
 fn macros() {
-    #macro[
-        [#recv[chan],
-         chan.recv()(chan)]
-    ];
     #macro[
         [#move[x],
          unsafe { let y <- *ptr::addr_of(x); y }]
@@ -77,14 +74,13 @@ fn from_port<A:send>(-port: future_pipe::client::waiting<A>) -> future<A> {
     The first time that the value is requested the task will block
     waiting for the result to be received on the port.
     "];
-    import future_pipe::client::recv;
 
     let port = ~mut some(port);
     do from_fn |move port| {
         let mut port_ = none;
         port_ <-> *port;
         let port = option::unwrap(port_);
-        alt (#recv(port)) {
+        alt recv(port) {
           future_pipe::completed(data, _next) { #move(data) }
         }
     }
@@ -137,55 +133,12 @@ fn with<A,B>(future: future<A>, blk: fn(A) -> B) -> B {
     blk(*v)
 }
 
-// The pipe protocol, generated by pipec
-/*
 proto! future_pipe {
     waiting:recv<T:send> {
         completed(T) -> terminated
     }
 
-    terminated { }
-}
-*/
-mod future_pipe {
-    fn init<T: send>() -> (client::waiting<T>, server::waiting<T>) {
-        { let (s, c) = pipes::entangle(); (c, s) }
-    }
-    enum waiting<T: send> { completed(T, client::terminated), }
-    enum terminated { }
-    mod client {
-        impl recv<T: send> for waiting<T> {
-            fn recv() -> extern fn(+waiting<T>) -> future_pipe::waiting<T> {
-                fn recv<T: send>(+pipe: waiting<T>) ->
-                   future_pipe::waiting<T> {
-                    option::unwrap(pipes::recv(pipe))
-                }
-                recv
-            }
-        }
-        type waiting<T: send> = pipes::recv_packet<future_pipe::waiting<T>>;
-        type terminated = pipes::send_packet<future_pipe::terminated>;
-    }
-    mod server {
-        fn completed<T: send>(+pipe: waiting<T>, +x_0: T) -> terminated {
-            {
-                let (s, c) = pipes::entangle();
-                let message = future_pipe::completed(x_0, s);
-                pipes::send(pipe, message);
-                c
-            }
-        }
-        type waiting<T: send> = pipes::send_packet<future_pipe::waiting<T>>;
-        impl recv for terminated {
-            fn recv() -> extern fn(+terminated) -> future_pipe::terminated {
-                fn recv(+pipe: terminated) -> future_pipe::terminated {
-                    option::unwrap(pipes::recv(pipe))
-                }
-                recv
-            }
-        }
-        type terminated = pipes::recv_packet<future_pipe::terminated>;
-    }
+    terminated:send { }
 }
 
 #[test]

src/libcore/pipes.rs

@@ -117,7 +117,11 @@ fn send<T: send>(-p: send_packet<T>, -payload: T) {
     }
 }
 
-fn recv<T: send>(-p: recv_packet<T>) -> option<T> {
+fn recv<T: send>(-p: recv_packet<T>) -> T {
+    option::unwrap(try_recv(p))
+}
+
+fn try_recv<T: send>(-p: recv_packet<T>) -> option<T> {
     let p_ = p.unwrap();
     let p = unsafe { uniquify(p_) };
     let this = rustrt::rust_get_task();
@@ -288,10 +292,10 @@ fn select2<A: send, B: send>(
 
     unsafe {
         alt i {
-          0 { left((recv(recv_packet(transmute(a))),
+          0 { left((try_recv(recv_packet(transmute(a))),
                     recv_packet(transmute(b)))) }
           1 { right((recv_packet(transmute(a)),
-                     recv(recv_packet(transmute(b))))) }
+                     try_recv(recv_packet(transmute(b))))) }
           _ { fail "select2 return an invalid packet" }
         }
     }
@@ -312,7 +316,7 @@ fn select<T: send>(+endpoints: ~[recv_packet<T>])
     do vec::consume(endpoints) |i, p| {
         let p = recv_packet(unsafe { unsafe::transmute(p) });
         if i == ready {
-            result = recv(p);
+            result = try_recv(p);
         }
         else {
             vec::push(remaining, p);

src/test/bench/msgsend-ring-pipes.rs

@@ -43,7 +43,7 @@ fn thread_ring(i: uint,
         num_port2 <-> num_port;
         num_chan = some(ring::client::num(option::unwrap(num_chan2), i * j));
         let port = option::unwrap(num_port2);
-        alt (option::unwrap(recv(port))) {
+        alt recv(port) {
           ring::num(_n, p) {
             //log(error, _n);
             num_port = some(#move(p));

src/test/bench/shootout-k-nucleotide-pipes.rs

@@ -43,7 +43,7 @@ mod stream {
             let mut endp = none;
             endp <-> self.endp;
             let streamp::data(x, endp) = unwrap(
-                pipes::recv(unwrap(endp)));
+                pipes::try_recv(unwrap(endp)));
             self.endp = some(endp);
             x
         }

src/test/run-pass/pipe-bank-proto.rs

@@ -4,7 +4,7 @@
 //
 // http://theincredibleholk.wordpress.com/2012/07/06/rusty-pipes/
 
-import pipes::recv;
+import pipes::try_recv;
 
 type username = str;
 type password = str;
@@ -43,7 +43,7 @@ fn bank_client(+bank: bank::client::login) {
     import bank::*;
 
     let bank = client::login(bank, "theincredibleholk", "1234");
-    let bank = alt recv(bank) {
+    let bank = alt try_recv(bank) {
       some(ok(connected)) {
         #move(connected)
       }
@@ -53,7 +53,7 @@ fn bank_client(+bank: bank::client::login) {
 
     let bank = client::deposit(bank, 100.00);
     let bank = client::withdrawal(bank, 50.00);
-    alt recv(bank) {
+    alt try_recv(bank) {
       some(money(m, _)) {
         io::println("Yay! I got money!");
       }

src/test/run-pass/pipe-detect-term.rs

@@ -6,7 +6,7 @@ use std;
 import std::timer::sleep;
 import std::uv;
 
-import pipes::{recv};
+import pipes::{try_recv, recv};
 
 proto! oneshot {
     waiting:send {
@@ -20,7 +20,7 @@ fn main() {
     let iotask = uv::global_loop::get();
 
     let c = pipes::spawn_service(oneshot::init, |p| { 
-        alt recv(p) {
+        alt try_recv(p) {
           some(*) { fail }
           none { }
         }

src/test/run-pass/pipe-pingpong-proto.rs

@@ -21,14 +21,14 @@ mod test {
 
         let chan = client::ping(chan);
         log(error, "Sent ping");
-        let pong(_chan) = option::unwrap(recv(chan));
+        let pong(_chan) = recv(chan);
         log(error, "Received pong");
     }
 
     fn server(-chan: pingpong::server::ping) {
         import pingpong::server;
 
-        let ping(chan) = option::unwrap(recv(chan));
+        let ping(chan) = recv(chan);
         log(error, "Received ping");
         let _chan = server::pong(chan);
         log(error, "Sent pong");

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

@@ -26,14 +26,9 @@ fn main() {
 
     let iotask = uv::global_loop::get();
 
-    #macro[
-        [#recv[chan],
-         chan.recv()(chan)]
-    ];
-
     let c = pipes::spawn_service(stream::init, |p| { 
         #error("waiting for pipes");
-        let stream::send(x, p) = option::unwrap(recv(p));
+        let stream::send(x, p) = recv(p);
         #error("got pipes");
         let (left, right) : (oneshot::server::waiting,
                              oneshot::server::waiting)
@@ -44,7 +39,7 @@ fn main() {
         assert i == 0;
 
         #error("waiting for pipes");
-        let stream::send(x, _) = option::unwrap(recv(p));
+        let stream::send(x, _) = recv(p);
         #error("got pipes");
         let (left, right) : (oneshot::server::waiting,
                              oneshot::server::waiting)