feat: add udp wrappers

Author: Chronostasys

planglib/std/__private.pi

@@ -23,4 +23,5 @@ use std::task::executor;
 use std::task::helper;
 use std::task::delay;
 use std::task::tcp;
+use std::task::udp;
 use std::json::encode;

planglib/std/libuv.pi

@@ -5,6 +5,7 @@ var UV_ASYNC = 1 as u32;
 var UV_IDLE = 6 as u32;
 var UV_TIMER = 13 as u32;
 var UV_TCP = 12 as u32;
+var UV_UDP = 14 as u32;
 
 
 // below are request types
@@ -17,6 +18,7 @@ pub struct uv_timer_t {}
 pub struct uv_tcp_t {}
 pub struct uv_stream_t {}
 pub struct uv_write_t {}
+pub struct uv_udp_t {}
 
 pub struct uv_buf_t {
     pub base:i64;
@@ -164,4 +166,17 @@ pub fn get_data_for_handle_raw<H|T> (handle:*H) *T {
     return unsafe_cast<T>(re);
 }
 
+pub fn uv_udp_init(loop:*uv_loop_t, handle:*uv_udp_t) i32;
+
+pub fn uv_udp_bind(handle:*uv_udp_t, addr:*sockaddr_in, flags:u32) i32;
+
+pub fn uv_udp_recv_start(handle:*uv_udp_t, alloc_cb:*(), recv_cb:*()) i32;
+
+pub fn uv_udp_send(req:*uv_write_t, handle:*uv_udp_t, bufs:*uv_buf_t, nbufs:u32, addr:*sockaddr_in, cb:*()) i32;
+
+pub fn new_uv_udp_t() *uv_udp_t {
+    let re = gc::malloc_pinned(uv_handle_size(UV_UDP) as i64);
+    return unsafe_cast<uv_udp_t>(re);
+}
+
 

planglib/std/task/reactor.pi

@@ -25,7 +25,7 @@ struct StopEV {
 }
 
 
-type EV =   TCPListenEV | TimerEV | StreamReadEV | ReadStopEV | WriteEV |StopEV;
+type EV = TCPListenEV | TimerEV | StreamReadEV | ReadStopEV | WriteEV | StopEV | UDPRecvEV | UDPSendEV;
 
 
 
@@ -54,6 +54,20 @@ struct StreamReadEV {
     buf: [u8];
 }
 
+struct UDPRecvEV {
+    handle: *libuv::uv_udp_t;
+    buf: [u8];
+    read_cb: |i64|=>void;
+    total_read: i64;
+}
+
+struct UDPSendEV {
+    handle: *libuv::uv_udp_t; 
+    buf: [u8];
+    write_cb: |i32|=>void;
+    addr: libuv::sockaddr_in;
+}
+
 var GLOBAL_REACTOR = new_uv_reactor();
 
 
@@ -113,7 +127,6 @@ fn write_cb(req:*libuv::uv_write_t, status:i32) void {
 
 use std::io;
 
-
 fn async_cb(async_t:*libuv::uv_async_t) void {
     let reactor = libuv::get_data_for_handle<libuv::uv_async_t|UVReactor>(async_t);
     let ch = reactor.ch;
@@ -172,6 +185,25 @@ fn async_cb(async_t:*libuv::uv_async_t) void {
                 libuv::uv_stop(reactor.loop);
                 libuv::uv_loop_close(GLOBAL_REACTOR.loop);
             }
+            UDPRecvEV(ev) => {
+                gc::keep_alive_pinned(ev.handle);
+                let cb = unsafe_cast<()>(&udp_recv_cb);
+                let alloc_cb = unsafe_cast<()>(&udp_alloc_cb);
+                libuv::uv_udp_recv_start(ev.handle, alloc_cb, cb);
+                libuv::set_data_for_handle(ev.handle, &ev);
+            }
+            UDPSendEV(ev) => {
+                let req = libuv::new_uv_write_t();
+                gc::keep_alive_pinned(req);
+                let raw_slice_p = &ev.buf[0];
+                gc::pin(raw_slice_p);
+                let buf = libuv::uv_buf_init(&ev.buf[0], ev.buf.len() as u32);
+                let buf_p = &buf;
+                gc::pin(buf_p);
+                libuv::set_data_for_handle(req, &ev);
+                let cb = unsafe_cast<()>(&udp_send_cb);
+                libuv::uv_udp_send(req, ev.handle, &buf, 1 as u32, &ev.addr, cb);
+            }
             _ => {
             }
         }
@@ -290,6 +322,41 @@ impl UVReactor {
         return;
     }
 
+    pub fn new_udp_socket(udp_h:*libuv::uv_udp_t, ip:string, port:i32) void {
+        let addr = libuv::sockaddr_in{};
+        libuv::uv_ip4_addr(ip.cstr(), port, &addr);
+        libuv::uv_udp_init(self.loop, udp_h);
+        libuv::uv_udp_bind(udp_h, &addr, 0 as u32);
+        return;
+    }
+
+    pub fn udp_recv(udp:*libuv::uv_udp_t, buf:[u8], readcb:|i64|=>void) void {
+        let ev = UDPRecvEV{
+            handle: udp,
+            buf: buf,
+            read_cb: readcb,
+            total_read: 0,
+        };
+        let e = ev as EV;
+        self.ch.send(e);
+        libuv::uv_async_send(self.async_t);
+        return;
+    }
+
+    pub fn udp_send(udp:*libuv::uv_udp_t, buf:[u8], ip:string, port:i32, writecb:|i32|=>void) void {
+        let addr = libuv::sockaddr_in{};
+        libuv::uv_ip4_addr(ip.cstr(), port, &addr);
+        let ev = UDPSendEV{
+            handle: udp,
+            buf: buf, 
+            write_cb: writecb,
+            addr: addr,
+        };
+        let e = ev as EV;
+        self.ch.send(e);
+        libuv::uv_async_send(self.async_t);
+        return;
+    }
 }
 
 
@@ -304,3 +371,27 @@ pub fn stop_global_reactor() void {
     GLOBAL_REACTOR.stop();
     return;
 }
+
+fn udp_recv_cb(handle:*libuv::uv_udp_t, nread:i64, _buf:*libuv::uv_buf_t) void {
+    let ev = libuv::get_data_for_handle<libuv::uv_udp_t|UDPRecvEV>(handle);
+    ev.total_read = nread + ev.total_read;
+    ev.read_cb(nread);
+    return;
+}
+
+fn udp_send_cb(req:*libuv::uv_write_t, status:i32) void {
+    let ev = libuv::get_data_for_handle<libuv::uv_write_t|UDPSendEV>(req);
+    ev.write_cb(status);
+    return;
+}
+
+fn udp_alloc_cb(handle:*(), _suggested_size:i64, buf:*libuv::uv_buf_t) void {
+    let ev = libuv::get_data_for_handle<_|UDPRecvEV>(handle);
+    let ptr = &ev.buf[ev.total_read];
+    let l = ev.buf.len() - ev.total_read;
+    if l<=0 {
+        l = 0;
+    }
+    * buf = libuv::uv_buf_init(ptr, l as u32);
+    return;
+}

planglib/std/task/udp.pi

@@ -0,0 +1,124 @@
+use std::task::Task;
+use std::task::reactor;
+use std::task::executor;
+use std::libuv;
+use std::slice::SliceExt;
+
+use std::io;
+
+pub struct UDPSocket {
+    handle:*libuv::uv_udp_t;
+}
+
+pub fn new_udp_socket(handle:*libuv::uv_udp_t) UDPSocket {
+    return UDPSocket{
+        handle:handle,
+    };
+}
+
+impl UDPSocket {
+    pub fn bind(ip:string, port:i32) void {
+        reactor::GLOBAL_REACTOR.new_udp_socket(self.handle, ip, port);
+        return;
+    }
+
+    pub fn read_async(buf:[u8]) Task<i64> {
+        // pin the buffer
+        let buf_ptr = &buf[0];
+        gc::pin(buf_ptr);
+        let read_task = UDPReadTask{
+            first:true,
+            ready:false,
+            handle: self.handle,
+            read_len:0,
+            buf:buf,
+        };
+
+        return read_task as Task<i64>;
+    }
+
+    pub fn send_async(buf:[u8], ip:string, port:i32) Task<i32> {
+        let write_task = UDPWriteTask{
+            first:true,
+            ready:false,
+            handle: self.handle,
+            buf:buf,
+            status:0 as i32,
+            ip:ip,
+            port:port,
+        };
+        return write_task as Task<i32>;
+    }
+}
+
+struct UDPWriteTask {
+    first:bool;
+    ready:bool;
+    handle:*libuv::uv_udp_t;
+    buf: [u8];
+    status:i32;
+    ip:string;
+    port:i32;
+}
+
+impl Task<i32> for UDPWriteTask {
+    fn poll(wk:||=>void) Option<i32> {
+        if self.first {
+            self.first = false;
+            let write_cb = |status| => {
+                if self.ready {
+                    return;
+                }
+                self.status = status;
+                self.ready = true;
+                wk();
+                return;
+            };
+            reactor::GLOBAL_REACTOR.udp_send(self.handle, self.buf, self.ip, self.port, write_cb);
+        }
+
+        if self.ready {
+            return self.status as Option<i32>;
+        }
+        return None{} as Option<i32>;
+    }
+}
+
+struct UDPReadTask {
+    first:bool;
+    ready:bool;
+    handle:*libuv::uv_udp_t;
+    read_len:i64;
+    buf: [u8];
+}
+
+impl Task<i64> for UDPReadTask {
+    fn poll(wk:||=>void) Option<i64> {
+        if self.first {
+            self.first = false;
+            let read_cb = |nread| => {
+                if self.ready {
+                    return;
+                }
+                if nread<0 {
+                    // error
+                    self.read_len = -1;
+                    self.ready = true;
+                    wk();
+                    return;
+                }
+                self.read_len = nread + self.read_len;
+                if self.read_len >= self.buf.len() {
+                    self.ready = true;
+                    wk();
+                }
+                return;
+            };
+            reactor::GLOBAL_REACTOR.udp_recv(self.handle, self.buf, read_cb);
+        }
+        if self.ready {
+            return self.read_len as Option<i64>;
+        }
+        return None{} as Option<i64>;
+    }
+} 

test/main.pi

@@ -73,6 +73,7 @@ async fn main() Task<()> {
     _io::test_io();
     // future_test::test_future();
     std_test::test_std();
+    // await std_test::test_udp();
 
     await std_test::test_nested_async_closure();
     await std_test::test_nested_async_closure_in_normal_f();

test/test/std_test.pi

@@ -226,4 +226,18 @@ pub struct JSONObj<T> {
     b: string;
 }
 
+use std::task::udp;
+use std::libuv;
+
+pub async fn test_udp() Task<()> {
+    let udp_t = udp::new_udp_socket(libuv::new_uv_udp_t());
+    udp_t.bind("127.0.0.1", 8080 as i32);
+    let buf = [u8 * 10;];
+    let nread = await udp_t.read_async(buf);
+    println!("read: ", nread);
+    let s = string_from_bytes(buf);
+    println!("udp msp:", s);
+
+    return ();
+}