Add option to run event loop once

base/stream.jl

@@ -2,7 +2,6 @@
 include("uv_constants.jl")
 
 ## types ##
-
 typealias Executable Union(Vector{ByteString},Function)
 typealias Callback Union(Function,Bool)
 type WaitTask 
@@ -113,6 +112,7 @@ end
 wait_connect_filter(w::AsyncStream, args...) = !w.open
 wait_readable_filter(w::AsyncStream, args...) = nb_available(w.buffer) <= 0
 wait_readnb_filter(w::(AsyncStream,Int), args...) = w[1].open && (nb_available(w[1].buffer) < w[2])
+wait_readbyte_filter(w::(AsyncStream,Uint8), args...) = w[1].open && (search(w[1].buffer,w[2]) <= 0)
 wait_readline_filter(w::AsyncStream, args...) = w.open && (search(w.buffer,'\n') <= 0)
 
 function wait(forwhat::Vector, notify_list_name, filter_fcn)
@@ -143,6 +143,7 @@ wait_readable(x) = wait(x, :readnotify, wait_readable_filter)
 wait_readline(x) = wait(x, :readnotify, wait_readline_filter)
 wait_readnb(x::(AsyncStream,Int)) = wait(x, :readnotify, wait_readnb_filter)
 wait_readnb(x::AsyncStream,b::Int) = wait_readnb((x,b))
+wait_readbyte(x::AsyncStream,c::Uint8) = wait((x,c), :readnotify, wait_readbyte_filter)
 
 #from `connect`
 function _uv_hook_connectcb(sock::AsyncStream, status::Int32)
@@ -306,7 +307,15 @@ eventloop() = ccall(:jl_global_event_loop,Ptr{Void},())
 function run_event_loop(loop::Ptr{Void})
     ccall(:jl_run_event_loop,Void,(Ptr{Void},),loop)
 end
+function process_events(loop::Ptr{Void})
+    ccall(:jl_process_events,Int32,(Ptr{Void},),loop)
+end
+function run_event_loop_once(loop::Ptr{Void})
+    ccall(:jl_run_once,Int32,(Ptr{Void},),loop)
+end
+process_events() = process_events(eventloop())
 run_event_loop() = run_event_loop(eventloop())
+run_event_loop_once() = run_event_loop_once(eventloop())
 
 ##pipe functions
 malloc_pipe() = c_malloc(_sizeof_uv_pipe)
@@ -382,6 +391,7 @@ end
 function read(this::AsyncStream,::Type{Uint8})
     buf = this.buffer
     assert(buf.seekable == false)
+    start_reading(this)
     wait_readnb(this,1)
     read(buf,Uint8)
 end
@@ -394,6 +404,14 @@ function readline(this::AsyncStream)
     readline(buf)
 end
 
+function readuntil(this::AsyncStream,c::Uint8)
+    buf = this.buffer
+    assert(buf.seekable == false)
+    start_reading(this)
+    wait_readbyte(this,c)
+    readuntil(buf,c)
+end
+
 function finish_read(pipe::NamedPipe)
     close(pipe) #handles to UV and ios will be invalid after this point
 end

src/jl_uv.c

@@ -220,14 +220,19 @@ DLLEXPORT uv_tcp_t *jl_make_tcp(uv_loop_t* loop, jl_value_t *julia_struct)
 
 /** This file contains wrappers for most of libuv's stream functionailty. Once we can allocate structs in Julia, this file will be removed */
 
+DLLEXPORT int jl_run_once(uv_loop_t *loop)
+{
+    if (loop) return uv_run(loop,UV_RUN_ONCE);
+}
+
 DLLEXPORT void jl_run_event_loop(uv_loop_t *loop)
 {
     if (loop) uv_run(loop,UV_RUN_DEFAULT);
 }
 
-DLLEXPORT void jl_process_events(uv_loop_t *loop)
+DLLEXPORT int jl_process_events(uv_loop_t *loop)
 {
-    if (loop) uv_run(loop,UV_RUN_NOWAIT);
+    if (loop) return uv_run(loop,UV_RUN_NOWAIT);
 }
 
 DLLEXPORT uv_pipe_t *jl_init_pipe(uv_pipe_t *pipe, int writable, int julia_only, jl_value_t *julia_struct)

src/julia.expmap

@@ -238,6 +238,7 @@
     jl_reshape_array;
     jl_restore_system_image;
     jl_run_event_loop;
+    jl_run_once;
     jl_save_system_image;
     jl_set_const;
     jl_set_current_module;

src/julia.h

@@ -807,7 +807,8 @@ DLLEXPORT int jl_spawn(char *name, char **argv, uv_loop_t *loop,
                        uv_handle_type stdout_type,uv_pipe_t *stdout_pipe,
                        uv_handle_type stderr_type,uv_pipe_t *stderr_pipe);
 DLLEXPORT void jl_run_event_loop(uv_loop_t *loop);
-DLLEXPORT void jl_process_events(uv_loop_t *loop);
+DLLEXPORT int jl_run_once(uv_loop_t *loop);
+DLLEXPORT int jl_process_events(uv_loop_t *loop);
 
 DLLEXPORT uv_loop_t *jl_global_event_loop();