diff --git a/main.odin b/main.odin index 88d089a..0b6c294 100644 --- a/main.odin +++ b/main.odin @@ -1,5 +1,74 @@ package main -main :: proc () { - +import "core:bufio" +import "core:fmt" +import "core:os" +import zmq "zmq" + +main :: proc() { + major, minor, patch := zmq.version() + fmt.printfln("zmq echo (libzmq %d.%d.%d) over inproc; type lines, Ctrl-D to quit", major, minor, patch) + + ctx := zmq.new_context() + defer zmq.term(ctx) + + rep := must_socket(ctx, .REP, "rep socket") + defer zmq.close(rep) + must(zmq.bind(rep, "inproc://echo"), "bind") + + req := must_socket(ctx, .REQ, "req socket") + defer zmq.close(req) + must(zmq.connect(req, "inproc://echo"), "connect") + + echo(req, rep) +} + +echo :: proc(req, rep: zmq.Socket) { + scanner: bufio.Scanner + bufio.scanner_init(&scanner, os.to_reader(os.stdin)) + defer bufio.scanner_destroy(&scanner) + + buf: [1024]u8 + for bufio.scan(&scanner) { + line := bufio.scanner_text(&scanner) + if len(line) == 0 { + continue + } + if _, err := zmq.send(req, transmute([]u8)line); !zmq.is_ok(err) { + fmt.eprintfln("send: %s", zmq.error_string(err)) + return + } + n, rerr := zmq.recv(rep, buf[:]) + if !zmq.is_ok(rerr) { + fmt.eprintfln("rep recv: %s", zmq.error_string(rerr)) + return + } + if _, werr := zmq.send(rep, buf[:n]); !zmq.is_ok(werr) { + fmt.eprintfln("rep send: %s", zmq.error_string(werr)) + return + } + m, qerr := zmq.recv(req, buf[:]) + if !zmq.is_ok(qerr) { + fmt.eprintfln("req recv: %s", zmq.error_string(qerr)) + return + } + fmt.println(string(buf[:m])) + } +} + +must :: proc(err: zmq.Error, at: string) { + if !zmq.is_ok(err) { + fatal(at, err) + } +} + +must_socket :: proc(ctx: zmq.Context, type: zmq.Socket_Type, at: string) -> zmq.Socket { + s, err := zmq.socket(ctx, type) + must(err, at) + return s +} + +fatal :: proc(at: string, err: zmq.Error) { + fmt.eprintfln("%s: %s", at, zmq.error_string(err)) + os.exit(1) } diff --git a/zmq/zmq.odin b/zmq/zmq.odin new file mode 100644 index 0000000..a54f156 --- /dev/null +++ b/zmq/zmq.odin @@ -0,0 +1,119 @@ +package zmq + +import "core:c" + +foreign import lib "system:zmq" + +@(default_calling_convention = "c") +foreign lib { + @(link_name = "zmq_ctx_new") _ctx_new :: proc() -> rawptr --- + @(link_name = "zmq_ctx_term") _ctx_term :: proc(ctx: rawptr) -> c.int --- + @(link_name = "zmq_socket") _socket :: proc(ctx: rawptr, type: Socket_Type) -> rawptr --- + @(link_name = "zmq_close") _close :: proc(s: rawptr) -> c.int --- + @(link_name = "zmq_bind") _bind :: proc(s: rawptr, addr: cstring) -> c.int --- + @(link_name = "zmq_connect") _connect :: proc(s: rawptr, addr: cstring) -> c.int --- + @(link_name = "zmq_send") _send :: proc(s: rawptr, buf: rawptr, len: c.size_t, flags: c.int) -> c.int --- + @(link_name = "zmq_recv") _recv :: proc(s: rawptr, buf: rawptr, len: c.size_t, flags: c.int) -> c.int --- + @(link_name = "zmq_errno") _errno :: proc() -> c.int --- + @(link_name = "zmq_strerror") _strerror :: proc(errnum: c.int) -> cstring --- + @(link_name = "zmq_version") _version :: proc(major, minor, patch: ^c.int) --- +} + +Context :: distinct rawptr +Socket :: distinct rawptr + +Socket_Type :: enum c.int { + PAIR = 0, + PUB = 1, + SUB = 2, + REQ = 3, + REP = 4, + DEALER = 5, + ROUTER = 6, + PULL = 7, + PUSH = 8, + XPUB = 9, + XSUB = 10, + STREAM = 11, +} + +Flag :: enum c.int { + DONTWAIT = 1, + SNDMORE = 2, +} + +Flags :: bit_set[Flag; c.int] + +NoBlock :: Flags{.DONTWAIT} + +Error :: distinct c.int + +OK :: Error(0) + +is_ok :: proc(e: Error) -> bool { return c.int(e) == 0 } + +error_string :: proc(e: Error) -> string { + return string(_strerror(c.int(e))) +} + +last_error :: proc() -> Error { return Error(_errno()) } + +version :: proc() -> (major, minor, patch: int) { + ma, mi, pa: c.int = 0, 0, 0 + _version(&ma, &mi, &pa) + return int(ma), int(mi), int(pa) +} + +new_context :: proc() -> Context { + return Context(_ctx_new()) +} + +term :: proc(ctx: Context) -> Error { + return _ctx_term(rawptr(ctx)) == 0 ? OK : last_error() +} + +socket :: proc(ctx: Context, type: Socket_Type) -> (Socket, Error) { + s := _socket(rawptr(ctx), type) + if s == nil { + return nil, last_error() + } + return Socket(s), OK +} + +close :: proc(s: Socket) -> Error { + return _close(rawptr(s)) == 0 ? OK : last_error() +} + +bind :: proc(s: Socket, addr: string) -> Error { + return endpoint(rawptr(s), addr, _bind) +} + +connect :: proc(s: Socket, addr: string) -> Error { + return endpoint(rawptr(s), addr, _connect) +} + +send :: proc(s: Socket, buf: []u8, flags: Flags = Flags{}) -> (int, Error) { + n := _send(rawptr(s), rawptr(&buf[0]), c.size_t(len(buf)), transmute(c.int)flags) + if n < 0 { + return -1, last_error() + } + return int(n), OK +} + +recv :: proc(s: Socket, buf: []u8, flags: Flags = Flags{}) -> (int, Error) { + n := _recv(rawptr(s), rawptr(&buf[0]), c.size_t(len(buf)), transmute(c.int)flags) + if n < 0 { + return -1, last_error() + } + return int(n), OK +} + +endpoint :: proc(s: rawptr, addr: string, f: proc "c" (s: rawptr, addr: cstring) -> c.int) -> Error { + buf: [256]u8 + n := copy(buf[:], addr) + if n >= len(buf) { + return Error(7) + } + buf[n] = 0 + return f(s, cstring(rawptr(&buf[0]))) == 0 ? OK : last_error() +}