|
| 1 | +## In-library CBOR-over-socket server — the *remote channel* of the |
| 2 | +## in-process/remote split. |
| 3 | +## |
| 4 | +## Included into `timer.nim` only under `-d:ffiIpcServe`, so it shares the |
| 5 | +## module scope: it can build a `MyTimer` and call the library's async procs |
| 6 | +## (`myTimerEcho`, …) **directly** — a plain Nim call, native, in-process, with |
| 7 | +## zero serialization between the socket layer and the logic. CBOR exists only |
| 8 | +## at the socket edge: each request is decoded, dispatched to the matching proc, |
| 9 | +## and the result re-encoded. There is no FFI boundary and no callback bridge |
| 10 | +## inside the server because the server *is* the library. |
| 11 | +## |
| 12 | +## Exposed as `my_timer_serve(address)` (C ABI) so a trivial host can start it; |
| 13 | +## the wire protocol is the same `proto.h` framing the C client/server already |
| 14 | +## speak, so the existing CBOR client works against this server unchanged. |
| 15 | + |
| 16 | +import std/[os, strutils] |
| 17 | +import chronos |
| 18 | +import chronos/transports/stream |
| 19 | +import results |
| 20 | +import ffi/cbor_serial |
| 21 | +import ffi/ffi_events |
| 22 | + |
| 23 | +# The async procs fire library events (e.g. echo -> onEchoFired). Off the FFI |
| 24 | +# thread that would log "event registry not set"; give this thread an empty |
| 25 | +# registry so dispatch finds zero listeners and stays quiet. Delivering events |
| 26 | +# to remote clients is separate, future work. |
| 27 | +var serveEventRegistry: FFIEventRegistry |
| 28 | + |
| 29 | +# Wire request envelopes. The field names mirror the generated CBOR request |
| 30 | +# ABI (each `{.ffi.}` proc packs its args under their Nim parameter names), so |
| 31 | +# the same bytes the C client builds decode straight into these. |
| 32 | +type |
| 33 | + EchoReqEnv = object |
| 34 | + req: EchoRequest |
| 35 | + ComplexReqEnv = object |
| 36 | + req: ComplexRequest |
| 37 | + ScheduleReqEnv = object |
| 38 | + job: JobSpec |
| 39 | + retry: RetryPolicy |
| 40 | + schedule: ScheduleConfig |
| 41 | + |
| 42 | +const |
| 43 | + RetOk: int32 = 0 # mirrors RET_OK / RET_ERR in my_timer_cbor.h |
| 44 | + RetErr: int32 = 1 |
| 45 | + |
| 46 | +proc toBytes(s: string): seq[byte] = |
| 47 | + var b = newSeq[byte](s.len) |
| 48 | + if s.len > 0: |
| 49 | + copyMem(addr b[0], unsafeAddr s[0], s.len) |
| 50 | + return b |
| 51 | + |
| 52 | +proc toStr(b: openArray[byte]): string = |
| 53 | + var s = newString(b.len) |
| 54 | + if b.len > 0: |
| 55 | + copyMem(addr s[0], unsafeAddr b[0], b.len) |
| 56 | + return s |
| 57 | + |
| 58 | +proc readU32(t: StreamTransport): Future[uint32] {.async.} = |
| 59 | + # Frame integers are network byte order (big-endian), per proto.h. |
| 60 | + var b: array[4, byte] |
| 61 | + await t.readExactly(addr b[0], 4) |
| 62 | + return |
| 63 | + (uint32(b[0]) shl 24) or (uint32(b[1]) shl 16) or (uint32(b[2]) shl 8) or |
| 64 | + uint32(b[3]) |
| 65 | + |
| 66 | +proc writeResponse(t: StreamTransport, ret: int32, body: seq[byte]) {.async.} = |
| 67 | + # Response frame: [i32 ret][u32 len][body], big-endian. |
| 68 | + var hdr = newSeq[byte](8) |
| 69 | + let r = cast[uint32](ret) |
| 70 | + hdr[0] = byte(r shr 24) |
| 71 | + hdr[1] = byte(r shr 16) |
| 72 | + hdr[2] = byte(r shr 8) |
| 73 | + hdr[3] = byte(r and 0xFF) |
| 74 | + let l = uint32(body.len) |
| 75 | + hdr[4] = byte(l shr 24) |
| 76 | + hdr[5] = byte(l shr 16) |
| 77 | + hdr[6] = byte(l shr 8) |
| 78 | + hdr[7] = byte(l and 0xFF) |
| 79 | + discard await t.write(hdr) |
| 80 | + if body.len > 0: |
| 81 | + discard await t.write(body) |
| 82 | + |
| 83 | +proc dispatch(meth: string, payload: seq[byte]): Future[(int32, seq[byte])] {.async.} = |
| 84 | + # The library's own state object — no FFIContext needed, we call the async |
| 85 | + # procs directly. `MyTimer` only carries a name, so building it per request |
| 86 | + # is free and keeps the handler stateless. |
| 87 | + let timer = MyTimer(name: "ipc-server") |
| 88 | + case meth |
| 89 | + of "version": |
| 90 | + let r = await myTimerVersion(timer) |
| 91 | + if r.isOk: |
| 92 | + return (RetOk, cborEncode(r.get())) # string -> CBOR text |
| 93 | + return (RetErr, toBytes(r.error())) |
| 94 | + of "echo": |
| 95 | + let env = cborDecode(payload, EchoReqEnv) |
| 96 | + if env.isErr: |
| 97 | + return (RetErr, toBytes(env.error())) |
| 98 | + let r = await myTimerEcho(timer, env.get().req) |
| 99 | + if r.isOk: |
| 100 | + return (RetOk, cborEncode(r.get())) |
| 101 | + return (RetErr, toBytes(r.error())) |
| 102 | + of "complex": |
| 103 | + let env = cborDecode(payload, ComplexReqEnv) |
| 104 | + if env.isErr: |
| 105 | + return (RetErr, toBytes(env.error())) |
| 106 | + let r = await myTimerComplex(timer, env.get().req) |
| 107 | + if r.isOk: |
| 108 | + return (RetOk, cborEncode(r.get())) |
| 109 | + return (RetErr, toBytes(r.error())) |
| 110 | + of "schedule": |
| 111 | + let env = cborDecode(payload, ScheduleReqEnv) |
| 112 | + if env.isErr: |
| 113 | + return (RetErr, toBytes(env.error())) |
| 114 | + let r = |
| 115 | + await myTimerSchedule(timer, env.get().job, env.get().retry, env.get().schedule) |
| 116 | + if r.isOk: |
| 117 | + return (RetOk, cborEncode(r.get())) |
| 118 | + return (RetErr, toBytes(r.error())) |
| 119 | + else: |
| 120 | + return (RetErr, toBytes("unknown method: " & meth)) |
| 121 | + |
| 122 | +proc onConnection( |
| 123 | + server: StreamServer, transp: StreamTransport |
| 124 | +) {.async: (raises: []).} = |
| 125 | + try: |
| 126 | + while not transp.atEof(): |
| 127 | + let mlen = await readU32(transp) |
| 128 | + var mbytes = newSeq[byte](int(mlen)) |
| 129 | + if mlen > 0'u32: |
| 130 | + await transp.readExactly(addr mbytes[0], int(mlen)) |
| 131 | + let plen = await readU32(transp) |
| 132 | + var payload = newSeq[byte](int(plen)) |
| 133 | + if plen > 0'u32: |
| 134 | + await transp.readExactly(addr payload[0], int(plen)) |
| 135 | + let (ret, body) = await dispatch(toStr(mbytes), payload) |
| 136 | + await writeResponse(transp, ret, body) |
| 137 | + except CatchableError: |
| 138 | + discard # peer closed or malformed frame: drop the connection |
| 139 | + try: |
| 140 | + await transp.closeWait() |
| 141 | + except CatchableError: |
| 142 | + discard |
| 143 | + |
| 144 | +proc serveLoop(address: string) {.async.} = |
| 145 | + var server: StreamServer |
| 146 | + if address.startsWith("unix:"): |
| 147 | + let path = address[5 ..^ 1] |
| 148 | + removeFile(path) # clear a stale socket so bind() succeeds |
| 149 | + server = createStreamServer(initTAddress(path), onConnection, {ReuseAddr}) |
| 150 | + elif address.startsWith("tcp:"): |
| 151 | + let hp = address[4 ..^ 1] |
| 152 | + let sep = hp.rfind(':') |
| 153 | + doAssert sep > 0, "tcp address must be tcp:<host>:<port>" |
| 154 | + server = |
| 155 | + createStreamServer( |
| 156 | + initTAddress(hp[0 ..< sep], Port(parseInt(hp[sep + 1 ..^ 1]))), |
| 157 | + onConnection, |
| 158 | + {ReuseAddr}, |
| 159 | + ) |
| 160 | + else: |
| 161 | + raise newException(ValueError, "address must be unix:<path> or tcp:<host>:<port>") |
| 162 | + server.start() |
| 163 | + echo "[serve] listening on ", address |
| 164 | + await server.join() |
| 165 | + |
| 166 | +proc my_timer_serve(address: cstring): cint {.exportc, cdecl, dynlib.} = |
| 167 | + ## C entry point: boot the library runtime and run the socket server forever. |
| 168 | + ## Blocks the calling thread. `address` is "unix:<path>" or "tcp:<host>:<port>". |
| 169 | + initializeLibrary() |
| 170 | + initEventRegistry(serveEventRegistry) |
| 171 | + ffiCurrentEventRegistry = addr serveEventRegistry |
| 172 | + try: |
| 173 | + waitFor serveLoop($address) |
| 174 | + except CatchableError as e: |
| 175 | + echo "[serve] error: ", e.msg |
| 176 | + return 1 |
| 177 | + return 0 |
0 commit comments