cleaning up uv_async stuff and stubbing uv_timer

1 files changed, 188 insertions, 38 deletions

diff --git a/src/libstd/uvtmp.rs b/src/libstd/uvtmp.rs
index 8a61df37982..75e962bd331 100644
--- a/src/libstd/uvtmp.rs
+++ b/src/libstd/uvtmp.rs
@@ -1,26 +1,31 @@
 // Some temporary libuv hacks for servo
 
 // UV2
+
+// these are processed solely in the
+// process_operation() crust fn below
 enum uv_operation {
-    op_async_init([u8])
+    op_async_init([u8]),
+    op_close(uv_handle, *ctypes::void)
 }
 
-type uv_async = {
-    id: [u8],
-    loop: uv_loop
-};
+enum uv_handle {
+    uv_async([u8], uv_loop)
+}
 
 enum uv_msg {
     // requests from library users
     msg_run(comm::chan<bool>),
     msg_run_in_bg(),
-    msg_loop_delete(),
-    msg_async_init(fn~(uv_async), fn~(uv_async)),
+    msg_async_init(fn~(uv_handle), fn~(uv_handle)),
     msg_async_send([u8]),
+    msg_close(uv_handle, fn~()),
 
     // dispatches from libuv
     uv_async_init([u8], *ctypes::void),
-    uv_async_send([u8])
+    uv_async_send([u8]),
+    uv_close([u8]),
+    uv_end()
 }
 
 type uv_loop_data = {
@@ -65,7 +70,10 @@ native mod rustrt {
         loop: *ctypes::void,
         data: *uv_loop_data);
     fn rust_uvtmp_uv_bind_op_cb(loop: *ctypes::void, cb: *u8) -> *ctypes::void;
+    fn rust_uvtmp_uv_stop_op_cb(handle: *ctypes::void);
     fn rust_uvtmp_uv_run(loop_handle: *ctypes::void);
+    fn rust_uvtmp_uv_close(handle: *ctypes::void, cb: *u8);
+    fn rust_uvtmp_uv_close_async(handle: *ctypes::void);
     fn rust_uvtmp_uv_async_send(handle: *ctypes::void);
     fn rust_uvtmp_uv_async_init(
         loop_handle: *ctypes::void,
@@ -74,7 +82,8 @@ native mod rustrt {
 }
 
 mod uv {
-    export loop_new, run, run_in_bg, async_init, async_send;
+    export loop_new, run, close, run_in_bg, async_init, async_send,
+           timer_init;
 
     // public functions
     fn loop_new() -> uv_loop unsafe {
@@ -129,10 +138,14 @@ mod uv {
             // all state goes here
             let handles: map::map<[u8], *ctypes::void> =
                 map::new_bytes_hash();
-            let async_cbs: map::map<[u8], fn~(uv_async)> =
+            let id_to_handle: map::map<[u8], uv_handle> =
+                map::new_bytes_hash();
+            let async_cbs: map::map<[u8], fn~(uv_handle)> =
                 map::new_bytes_hash();
             let async_init_after_cbs: map::map<[u8],
-                                               fn~(uv_async)> =
+                                               fn~(uv_handle)> =
+                map::new_bytes_hash();
+            let close_callbacks: map::map<[u8], fn~()> =
                 map::new_bytes_hash();
 
             // the main loop that this task blocks on.
@@ -152,7 +165,9 @@ mod uv {
                         rustrt::rust_uvtmp_uv_run(loop_handle);
                         // when we're done, msg the
                         // end chan
+                        rustrt::rust_uvtmp_uv_stop_op_cb(op_handle);
                         comm::send(end_chan, true);
+                        comm::send(rust_loop_chan, uv_end);
                     };
                   }
                   
@@ -163,6 +178,34 @@ mod uv {
                     };
                   }
                   
+                  msg_close(handle, cb) {
+                    let id = get_id_from_handle(handle);
+                    close_callbacks.insert(id, cb);
+                    let handle_ptr = handles.get(id);
+                    let op = op_close(handle, handle_ptr);
+
+                    pass_to_libuv(op_handle, operation_chan, op);
+                  }
+                  uv_close(id) {
+                    handles.remove(id);
+                    let handle = id_to_handle.get(id);
+                    id_to_handle.remove(id);
+                    alt handle {
+                      uv_async(id, _) {
+                        async_cbs.remove(id);
+                      }
+                      _ {
+                        fail "unknown form of uv_handle encountered "
+                            + "in uv_close handler";
+                      }
+                    }
+                    let cb = close_callbacks.get(id);
+                    close_callbacks.remove(id);
+                    task::spawn {||
+                        cb();
+                    };
+                  }
+                  
                   msg_async_init(callback, after_cb) {
                     // create a new async handle
                     // with the id as the handle's
@@ -172,9 +215,7 @@ mod uv {
                     async_cbs.insert(id, callback);
                     async_init_after_cbs.insert(id, after_cb);
                     let op = op_async_init(id);
-                    comm::send(operation_chan, op);
-                    rustrt::rust_uvtmp_uv_async_send(op_handle);
-                    io::println("MSG_ASYNC_INIT");
+                    pass_to_libuv(op_handle, operation_chan, op);
                   }
                   uv_async_init(id, async_handle) {
                     // libuv created a handle, which is
@@ -184,22 +225,25 @@ mod uv {
                     handles.insert(id, async_handle);
                     let after_cb = async_init_after_cbs.get(id);
                     async_init_after_cbs.remove(id);
+                    let async = uv_async(id, rust_loop_chan);
+                    id_to_handle.insert(id, copy(async));
                     task::spawn {||
-                        let async: uv_async = {
-                            id: id,
-                            loop: rust_loop_chan
-                        };
                         after_cb(async);
                     };
                   }
 
                   msg_async_send(id) {
                     let async_handle = handles.get(id);
-                    rustrt::rust_uvtmp_uv_async_send(async_handle);
+                    do_send(async_handle);
                   }
                   uv_async_send(id) {
                     let async_cb = async_cbs.get(id);
-                    async_cb({id: id, loop: rust_loop_chan});
+                    task::spawn {||
+                        async_cb(uv_async(id, rust_loop_chan));
+                    };
+                  }
+                  uv_end() {
+                    keep_going = false;
                   }
 
                   _ { fail "unknown form of uv_msg received"; }
@@ -222,17 +266,45 @@ mod uv {
 
     fn async_init (
         loop: uv_loop,
-        async_cb: fn~(uv_async),
-        after_cb: fn~(uv_async)) {
+        async_cb: fn~(uv_handle),
+        after_cb: fn~(uv_handle)) {
         let msg = msg_async_init(async_cb, after_cb);
         comm::send(loop, msg);
     }
 
-    fn async_send(async: uv_async) {
-        comm::send(async.loop, msg_async_send(async.id));
+    fn async_send(async: uv_handle) {
+        alt async {
+          uv_async(id, loop) {
+            comm::send(loop, msg_async_send(id));
+          }
+          _ {
+            fail "attempting to call async_send() with a" +
+                " uv_async uv_handle";
+          }
+        }
+    }
+
+    fn close(h: uv_handle, cb: fn~()) {
+        let loop_chan = get_loop_chan_from_handle(h);
+        comm::send(loop_chan, msg_close(h, cb)); 
+    }
+
+    fn timer_init(loop: uv_loop, after_cb: fn~(uv_handle)) {
+        let msg = msg_timer_init(after_cb);
+        comm::send(loop, msg);
     }
 
     // internal functions
+    fn pass_to_libuv(
+            op_handle: *ctypes::void,
+            operation_chan: comm::chan<uv_operation>,
+            op: uv_operation) unsafe {
+        comm::send(operation_chan, copy(op));
+        do_send(op_handle);
+    }
+    fn do_send(h: *ctypes::void) {
+        rustrt::rust_uvtmp_uv_async_send(h);
+    }
     fn gen_handle_id() -> [u8] {
         ret rand::mk_rng().gen_bytes(16u);
     }
@@ -240,24 +312,45 @@ mod uv {
         ret vec::unsafe::from_buf(buf, 16u); 
     }
 
-    fn get_loop_chan_from(data: *uv_loop_data)
-            -> comm::chan<uv_msg> unsafe {
+    fn get_loop_chan_from_data(data: *uv_loop_data)
+            -> uv_loop unsafe {
         ret (*data).rust_loop_chan;
     }
 
+    fn get_loop_chan_from_handle(handle: uv_handle)
+        -> uv_loop {
+        alt handle {
+          uv_async(id,loop) {
+            ret loop;
+          }
+          _ {
+            fail "unknown form of uv_handle for get_loop_chan_from "
+                 + " handle";
+          }
+        }
+    }
+
+    fn get_id_from_handle(handle: uv_handle) -> [u8] {
+        alt handle {
+          uv_async(id,loop) {
+            ret id;
+          }
+          _ {
+            fail "unknown form of uv_handle for get_id_from handle";
+          }
+        }
+    }
+
     // crust
     crust fn process_operation(
             loop: *ctypes::void,
             data: *uv_loop_data) unsafe {
-        io::println("IN PROCESS_OPERATION");
         let op_port = (*data).operation_port;
-        let loop_chan = get_loop_chan_from(data);
+        let loop_chan = get_loop_chan_from_data(data);
         let op_pending = comm::peek(op_port);
         while(op_pending) {
-            io::println("OPERATION PENDING!");
             alt comm::recv(op_port) {
               op_async_init(id) {
-                io::println("OP_ASYNC_INIT");
                 let id_ptr = vec::unsafe::to_ptr(id);
                 let async_handle = rustrt::rust_uvtmp_uv_async_init(
                     loop,
@@ -267,21 +360,61 @@ mod uv {
                     id,
                     async_handle));
               }
+              op_close(handle, handle_ptr) {
+                handle_op_close(handle, handle_ptr);
+              }
               
               _ { fail "unknown form of uv_operation received"; }
             }
             op_pending = comm::peek(op_port);
         }
-        io::println("NO MORE OPERATIONS PENDING!");
+    }
+
+    fn handle_op_close(handle: uv_handle, handle_ptr: *ctypes::void) {
+        // it's just like im doing C
+        alt handle {
+          uv_async(id, loop) {
+            let cb = process_close_async;
+            rustrt::rust_uvtmp_uv_close(
+                handle_ptr, cb);
+          }
+          _ {
+            fail "unknown form of uv_handle encountered " +
+                "in process_operation/op_close";
+          }
+        }
     }
 
     crust fn process_async_send(id_buf: *u8, data: *uv_loop_data)
             unsafe {
         let handle_id = get_handle_id_from(id_buf);
-        let loop_chan = get_loop_chan_from(data);
+        let loop_chan = get_loop_chan_from_data(data);
         comm::send(loop_chan, uv_async_send(handle_id));
     }
 
+    fn process_close_common(id: [u8], data: *uv_loop_data)
+        unsafe {
+        // notify the rust loop that their handle is closed, then
+        // the caller will invoke a per-handle-type c++ func to
+        // free allocated memory
+        let loop_chan = get_loop_chan_from_data(data);
+        comm::send(loop_chan, uv_close(id));
+    }
+
+    crust fn process_close_async(
+        id_buf: *u8,
+        handle_ptr: *ctypes::void,
+        data: *uv_loop_data)
+        unsafe {
+        let id = get_handle_id_from(id_buf);
+        rustrt::rust_uvtmp_uv_close_async(handle_ptr);
+        // at this point, the handle and its data has been
+        // released. notify the rust loop to remove the
+        // handle and its data and call the user-supplied
+        // close cb
+        process_close_common(id, data);
+    }
+
     
 }
 
@@ -295,14 +428,31 @@ fn test_uvtmp_uv_new_loop_no_handles() {
 #[test]
 fn test_uvtmp_uv_simple_async() {
     let test_loop = uv::loop_new();
-    let cb: fn~(uv_async) = fn~(h: uv_async) {
-        io::println("HELLO FROM ASYNC CALLBACK!");
-    };
-    uv::async_init(test_loop, cb) {|new_async|
-        io::println("NEW_ASYNC CREATED!");
+    let exit_port = comm::port::<bool>();
+    let exit_chan = comm::chan::<bool>(exit_port);
+    uv::async_init(test_loop, {|new_async|
+        uv::close(new_async) {||
+            comm::send(exit_chan, true);
+        };
+    }, {|new_async|
         uv::async_send(new_async);
-    };
+    });
+    uv::run(test_loop);
+    assert comm::recv(exit_port);
+}
+
+#[test]
+fn test_uvtmp_uv_timer() {
+    let test_loop = uv::loop_new();
+    let exit_port = comm::port::<bool>();
+    let exit_chan = comm::chan::<bool>(exit_port);
+    uv::timer(test_loop, {|new_timer|
+        uv::timer_start(new_async) {||
+            comm::send(exit_chan, true);
+        };
+    }); 
     uv::run(test_loop);
+    assert comm::recv(exit_port);
 }
 
 // END OF UV2