diff --git a/examples/polling_echo/main.odin b/examples/polling_echo/main.odin new file mode 100644 index 0000000..d2790d4 --- /dev/null +++ b/examples/polling_echo/main.odin @@ -0,0 +1,152 @@ +package main + +import zmq "../../" +import "core:fmt" +import "core:log" +import "core:os" + +main :: proc() { + context.logger = log.create_console_logger(.Debug, {}) + + major, minor, patch := zmq.version() + log.infof( + "zmq echo (libzmq %d.%d.%d) over inproc; DEALER/DEALER via zmq_poll", + major, + minor, + patch, + ) + + ctx := zmq.ctx_new() + defer zmq.ctx_term(ctx) + + // a is user-facing (stdin/stdout); b is the echoer. DEALER/DEALER is + // async two-way: either end may send/recv freely, so the poll loop + // needs no request/reply state tracking (unlike REQ/REP). + a, serr := zmq.socket(ctx, .DEALER) + if serr != nil { + log.panic(zmq.strerror(serr)) + } + defer zmq.close(a) + if err := zmq.bind(a, "inproc://echo"); err != nil { + log.panic(zmq.strerror(err)) + } + + b, cerr := zmq.socket(ctx, .DEALER) + if cerr != nil { + log.panic(zmq.strerror(cerr)) + } + defer zmq.close(b) + if err := zmq.connect(b, "inproc://echo"); err != nil { + log.panic(zmq.strerror(err)) + } + + echo(a, b) +} + +echo :: proc(a, b: zmq.Socket) { + pending: [dynamic]u8 // stdin line assembly + defer delete(pending) + msg: [255]u8 + + // Poll stdin + both DEALERs together. Each item polls its socket + // when non-nil, otherwise its fd. + items := [3]zmq.Poll_Item { + {fd = 0, events = {.POLLIN}}, // stdin + {socket = a, events = {.POLLIN}}, // reply arrived from b + {socket = b, events = {.POLLIN}}, // request arrived from a + } + + // in_flight tracks lines sent via a whose reply hasn't been printed yet. + // On stdin EOF we stop reading stdin but keep looping until every + // in-flight echo has been printed, so piped input isn't truncated. + in_flight := 0 + eof := false + run := true + + fmt.print("> ") + for run { + _, err := zmq.poll(items[:], -1) // block until something is ready + if err != nil { + fmt.eprintfln("poll: %s", zmq.strerror(err)) + return + } + + // stdin ready: read what's available, assemble into lines. + if !eof && .POLLIN in items[0].revents { + tmp: [256]u8 + n, _ := os.read(os.stdin, tmp[:]) + if n == 0 { + // EOF: flush any trailing partial line, then stop + // watching stdin and drain remaining replies. + eof = true + items[0].events = {} + if len(pending) > 0 { + if _, e := zmq.send(a, pending[:]); e != nil { + fmt.eprintfln("send a: %s", zmq.strerror(e)) + } else { + in_flight += 1 + } + resize(&pending, 0) + } + } else { + append(&pending, ..tmp[:n]) + } + } + + // a got a reply from b -> print it. + if .POLLIN in items[1].revents { + n, e := zmq.recv(a, msg[:]) + if e != nil { + fmt.eprintfln("recv a: %s", zmq.strerror(e)) + } else { + fmt.printf("< %s\n> ", string(msg[:n])) + in_flight -= 1 + } + } + + // b got a request from a -> echo it back. + if .POLLIN in items[2].revents { + n, e := zmq.recv(b, msg[:]) + if e != nil { + fmt.eprintfln("recv b: %s", zmq.strerror(e)) + } else { + if _, se := zmq.send(b, msg[:n]); se != nil { + fmt.eprintfln("send b: %s", zmq.strerror(se)) + } + } + } + + // Drain complete lines from the stdin buffer -> send via a. + if !eof { + for { + idx := -1 + for i in 0 ..< len(pending) { + if pending[i] == '\n' { + idx = i + break + } + } + if idx < 0 { + break + } + line := pending[:idx] + if len(line) > 0 { + if _, e := zmq.send(a, line); e != nil { + fmt.eprintfln("send a: %s", zmq.strerror(e)) + } else { + in_flight += 1 + } + } + // consume the line + its newline + consumed := idx + 1 + copy(pending[:], pending[consumed:]) + resize(&pending, len(pending) - consumed) + } + } + + if eof && in_flight == 0 { + run = false + } + } +} + diff --git a/test_error/main.odin b/test_error/main.odin index 0efae95..de73849 100644 --- a/test_error/main.odin +++ b/test_error/main.odin @@ -2,7 +2,7 @@ package main import "core:fmt" import "core:os" -import "zmq" +import zmq "../" main :: proc() { ctx := zmq.ctx_new() diff --git a/zmq.odin b/zmq.odin index 8b56f3b..b055443 100644 --- a/zmq.odin +++ b/zmq.odin @@ -33,6 +33,8 @@ foreign lib { @(link_name = "zmq_recv") _recv :: proc(s: Socket, buf: rawptr, len: c.size_t, flags: Flags) -> c.int --- + @(link_name = "zmq_poll") + _poll :: proc(items: rawptr, nitems: c.int, timeout: c.long) -> c.int --- @(link_name = "zmq_strerror") _strerror :: proc(errnum: c.int) -> cstring --- @(link_name = "zmq_version") @@ -55,14 +57,34 @@ Socket_Type :: enum c.int { } Flag :: enum c.int { - DONTWAIT = 1, - SNDMORE = 2, + DONTWAIT, + SNDMORE, } Flags :: bit_set[Flag;c.int] NoBlock :: Flags{.DONTWAIT} +// zmq_poll event flags (zmq.h ZMQ_POLLIN etc.). bit_set uses enum values as +// bit indices, so these are positions (0..3), not the ZMQ_POLL* masks (1,2,4,8): +// {.POLLIN} -> bit 0 -> mask 1 == ZMQ_POLLIN, {.POLLOUT} -> bit 1 -> mask 2, etc. +Poll_Event :: enum c.short { + POLLIN = 0, + POLLOUT = 1, + POLLERR = 2, + POLLPRI = 3, +} +Poll_Events :: bit_set[Poll_Event;c.short] + +// Mirrors C zmq_pollitem_t (16 B: ptr / int / short / short). +// socket non-nil => poll the ØMQ socket; otherwise poll fd. +Poll_Item :: struct { + socket: Socket, + fd: c.int, + events: Poll_Events, + revents: Poll_Events, +} + Error :: union #shared_nil { os.Platform_Error, ZMQ_Error, @@ -158,3 +180,12 @@ recv :: proc(s: Socket, buf: []u8, flags: Flags = Flags{}) -> (int, Error) { return int(n), nil } +// timeout_ms -1 = block forever, 0 = non-blocking. +poll :: proc(items: []Poll_Item, timeout_ms: c.long) -> (int, Error) { + n := _poll(rawptr(&items[0]), c.int(len(items)), timeout_ms) + if n < 0 { + return -1, last_error() + } + return int(n), nil +} + diff --git a/zmq_test.odin b/zmq_test.odin index 72c5adc..259c950 100644 --- a/zmq_test.odin +++ b/zmq_test.odin @@ -5,9 +5,80 @@ package zmq import "core:testing" @(test) -test_error_type :: proc(t: ^testing.T) { - e: Error = .EFSM +test_version :: proc(t: ^testing.T) { + ma, mi, pa := version() - testing.expect_value(t, size_of(e), 8) + testing.expect_value(t, ma, 4) + testing.expect_value(t, mi, 3) + testing.expect_value(t, pa, 5) +} + +@(test) +test_simple_const_echo :: proc(t: ^testing.T) { + ctx := ctx_new() + defer ctx_term(ctx) + + srv, serr := socket(ctx, .REP) + testing.expect_value(t, serr, nil) + if serr != nil { + testing.fail_now(t, "Can't create server socket") + } + defer close(srv) + + if err := bind(srv, "inproc://echo"); err != nil { + testing.expect_value(t, err, nil) + testing.fail_now(t, "Can't bind server socket") + } + + cli, cerr := socket(ctx, .REQ) + testing.expect_value(t, cerr, nil) + if cerr != nil { + testing.fail_now(t, "Can't create client socket") + } + defer close(cli) + err := connect(cli, "inproc://echo") + testing.expect_value(t, err, nil) + if err != nil { + testing.fail_now(t, "Can't connect client to server") + } + + msg :: "Hello, World" + buf: [len(msg)]u8 + + _, err = send_const(cli, msg) + testing.expect_value(t, err, nil) + if err != nil { + testing.fail_now(t, "Can't send message") + } + + { + n := 0 + n, err = recv(srv, buf[:]) + testing.expect_value(t, n, len(msg)) + testing.expect_value(t, err, nil) + if err != nil || n != len(msg) { + testing.fail_now(t, "Can't receive message") + } + + testing.expect_value(t, string(buf[:n]), string(msg)) + + _, err := send(srv, buf[:n]) + testing.expect_value(t, err, nil) + if err != nil { + testing.fail_now(t, "Can't send message") + } + } + + { + m := 0 + m, err = recv(cli, buf[:]) + testing.expect_value(t, m, len(msg)) + testing.expect_value(t, err, nil) + if err != nil || m != len(msg) { + testing.fail_now(t, "Can't receive message") + } + + testing.expect_value(t, string(buf[:m]), string(msg)) + } }