diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4243081..53a6a32 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,7 +12,7 @@ jobs: unit_test_and_coverage: runs-on: ubuntu-latest container: - image: imng/zero-kcov:0.1 + image: imng/zero-kcov:0.2 options: --security-opt seccomp=unconfined steps: - name: Checkout code @@ -20,7 +20,7 @@ jobs: with: token: ${{ secrets.GH_TOKEN }} - - name: Set Zig 0.15.2 as active + - name: Set Zig 0.16.0 as active run: | zig version kcov --version @@ -36,10 +36,10 @@ jobs: if [ -f zig-out/kcov/test/coverage.json ]; then COVERAGE=$(jq -r '.percent_covered' zig-out/kcov/test/coverage.json) echo "coverage=$COVERAGE" >> $GITHUB_OUTPUT - echo "Coverage for Zig 0.15.2: $COVERAGE%" + echo "Coverage for Zig 0.16.0: $COVERAGE%" else echo "coverage=0" >> $GITHUB_OUTPUT - echo "Coverage for Zig 0.15.2: 0%" + echo "Coverage for Zig 0.16.0: 0%" fi - name: Update README with coverage diff --git a/Dockerfiles/alpine.Dockerfile b/Dockerfiles/alpine.Dockerfile index 6f604c0..86b2a59 100644 --- a/Dockerfiles/alpine.Dockerfile +++ b/Dockerfiles/alpine.Dockerfile @@ -28,14 +28,14 @@ RUN mkdir -p /opt/zig-0.15.1 /opt/zig-0.15.2 /opt/zig-0.16.0 /opt/kcov # && rm zig-x86_64-linux-0.15.1.tar.xz # Install Zig 0.15.2 -RUN wget -q https://ziglang.org/download/0.15.2/zig-x86_64-linux-0.15.2.tar.xz \ - && tar -xJf zig-x86_64-linux-0.15.2.tar.xz -C /opt/zig-0.15.2 --strip-components=1 \ - && rm zig-x86_64-linux-0.15.2.tar.xz +# RUN wget -q https://ziglang.org/download/0.15.2/zig-x86_64-linux-0.15.2.tar.xz \ +# && tar -xJf zig-x86_64-linux-0.15.2.tar.xz -C /opt/zig-0.15.2 --strip-components=1 \ +# && rm zig-x86_64-linux-0.15.2.tar.xz # Install Zig 0.16.0 -# RUN wget -q https://ziglang.org/download/0.16.0/zig-x86_64-linux-0.16.0.tar.xz \ -# && tar -xJf zig-x86_64-linux-0.16.0.tar.xz -C /opt/zig-0.16.0 --strip-components=1 \ -# && rm zig-x86_64-linux-0.16.0.tar.xz +RUN wget -q https://ziglang.org/download/0.16.0/zig-x86_64-linux-0.16.0.tar.xz \ + && tar -xJf zig-x86_64-linux-0.16.0.tar.xz -C /opt/zig-0.16.0 --strip-components=1 \ + && rm zig-x86_64-linux-0.16.0.tar.xz # Install kcov from source RUN wget -q https://github.com/SimonKagstrom/kcov/archive/refs/heads/master.tar.gz \ @@ -52,7 +52,7 @@ RUN wget -q https://github.com/SimonKagstrom/kcov/archive/refs/heads/master.tar. FROM alpine:latest LABEL maintainer="im-ng" LABEL description="Multi-version Zig CI container with kcov coverage support" -LABEL version="0.1" +LABEL version="0.2" # Update dependencies for zig, zero, kcov RUN apk add --no-cache \ @@ -64,10 +64,11 @@ RUN apk add --no-cache \ binutils-dev curl-dev elfutils-dev # Create working directories -RUN mkdir -p /usr/local/zig-0.15.2 /app +# RUN mkdir -p /usr/local/zig-0.15.2 /app +RUN mkdir -p /usr/local/zig-0.16.0 /app COPY --from=builder /opt/kcov* /usr/ -COPY --from=builder /opt/zig-0.15.2 /usr/local/zig-0.15.2/ +COPY --from=builder /opt/zig-0.16.0 /usr/local/zig-0.16.0/ # RUN ls -alt /usr/local/zig-0.15.2/ # RUN ls -alth @@ -75,9 +76,9 @@ COPY --from=builder /opt/zig-0.15.2 /usr/local/zig-0.15.2/ # ENV ZIG151=/opt/zig/zig-0.15.1 # ENV ZIG152=/opt/zig/zig-0.15.2 # ENV ZIG160=/opt/zig/zig-0.16.0 -ENV ZIG=/usr/local/zig-0.15.2 +ENV ZIG=/usr/local/zig-0.16.0 -RUN ln -s /usr/local/zig-0.15.2/zig /usr/local/bin/zig +RUN ln -s /usr/local/zig-0.16.0/zig /usr/local/bin/zig # RUN chmod -R 777 /usr/local/zig-0.15.2 # ENV PATH="${ZIG151}:${ZIG152}:${ZIG160}:${PATH}" # ENV PATH="${ZIG}:${PATH}" diff --git a/Makefile b/Makefile index 8964054..4116666 100644 --- a/Makefile +++ b/Makefile @@ -9,20 +9,22 @@ top: clean: rm -rf .zig-cache zig-out zig-pkg - rm -rf examples/zero-auth/.zig-cache examples/zero-auth/zig-out - rm -rf examples/zero-basic/.zig-cache examples/zero-basic/zig-out - rm -rf examples/zero-cronz/.zig-cache examples/zero-cronz/zig-out - rm -rf examples/zero-migration/.zig-cache examples/zero-migration/zig-out - rm -rf examples/zero-mqtt-publisher/.zig-cache examples/zero-mqtt-publisher/zig-out - rm -rf examples/zero-mqtt-subscriber/.zig-cache examples/zero-mqtt-subscriber/zig-out - rm -rf examples/zero-redis/.zig-cache examples/zero-redis/zig-out - rm -rf examples/zero-service-client/.zig-cache examples/zero-service-client/zig-out - rm -rf examples/zero-stream/.zig-cache examples/zero-stream/zig-out - rm -rf examples/zero-todo-htmx/.zig-cache examples/zero-todo-htmx/zig-out - rm -rf examples/zero-websocket/.zig-cache examples/zero-websocket/zig-out - rm -rf examples/zero-kafka-publisher/.zig-cache examples/zero-kafka-publisher/zig-out - rm -rf examples/zero-kafka-subscriber/.zig-cache examples/zero-kafka-subscriber/zig-out - rm -rf examples/zero-sqlite/.zig-cache examples/zero-sqlite/zig-out + rm -rf examples/zero-auth/.zig-cache examples/zero-auth/zig-out examples/zero-auth/zig-pkg + rm -rf examples/zero-basic/.zig-cache examples/zero-basic/zig-out examples/zero-basic/zig-pkg + rm -rf examples/zero-cronz/.zig-cache examples/zero-cronz/zig-out examples/zero-cronz/zig-pkg + rm -rf examples/zero-migration/.zig-cache examples/zero-migration/zig-out examples/zero-migration/zig-pkg + rm -rf examples/zero-mqtt-publisher/.zig-cache examples/zero-mqtt-publisher/zig-out examples/zero-mqtt-publisher/zig-pkg + rm -rf examples/zero-mqtt-subscriber/.zig-cache examples/zero-mqtt-subscriber/zig-out examples/zero-mqtt-subscriber/zig-pkg + rm -rf examples/zero-redis/.zig-cache examples/zero-redis/zig-out examples/zero-redis/zig-pkg + rm -rf examples/zero-service-client/.zig-cache examples/zero-service-client/zig-out examples/zero-service-client/zig-pkg + rm -rf examples/zero-stream/.zig-cache examples/zero-stream/zig-out examples/zero-stream/zig-pkg + rm -rf examples/zero-todo-htmx/.zig-cache examples/zero-todo-htmx/zig-out examples/zero-todo-htmx/zig-pkg + rm -rf examples/zero-websocket/.zig-cache examples/zero-websocket/zig-out examples/zero-websocket/zig-pkg + rm -rf examples/zero-kafka-publisher/.zig-cache examples/zero-kafka-publisher/zig-out examples/zero-kafka-publisher/zig-pkg + rm -rf examples/zero-kafka-subscriber/.zig-cache examples/zero-kafka-subscriber/zig-out examples/zero-kafka-subscriber/zig-pkg + rm -rf examples/zero-sqlite/.zig-cache examples/zero-sqlite/zig-out examples/zero-sqlite/zig-pkg + rm -rf examples/zero-nats-publisher/.zig-cache examples/zero-nats-publisher/zig-out examples/zero-nats-publisher/zig-pkg + rm -rf examples/zero-nats-subscriber/.zig-cache examples/zero-nats-subscriber/zig-out examples/zero-nats-subscriber/zig-pkg release: zig build --release=fast diff --git a/build.zig b/build.zig index 23e8f41..4c4041c 100644 --- a/build.zig +++ b/build.zig @@ -17,9 +17,6 @@ pub fn build(b: *std.Build) void { const httpz = b.dependency("httpz", .{}); module.addImport("httpz", httpz.module("httpz")); - const metriks = b.dependency("metriks", .{}); - module.addImport("metriks", metriks.module("metriks")); - const env = b.dependency("dotenv", .{}); module.addImport("dotenv", env.module("dotenv")); @@ -44,6 +41,9 @@ pub fn build(b: *std.Build) void { const sqlite = b.dependency("sqlite", .{}); module.addImport("sqlite", sqlite.module("sqlite")); + const nats = b.dependency("nats", .{}); + module.addImport("nats", nats.module("nats")); + // if (b.option( // bool, // "kafka", @@ -55,7 +55,9 @@ pub fn build(b: *std.Build) void { module.addIncludePath(.{ .cwd_relative = "/usr/local/Cellar/librdkafka/2.13.0/include" }); module.addLibraryPath(.{ .cwd_relative = "/usr/local/Cellar/librdkafka/2.13.0/lib" }); } - module.linkSystemLibrary("rdkafka", .{ .weak = true }); + module.linkSystemLibrary("rdkafka", .{ + .weak = true, + }); const test_module = b.createModule(.{ .root_source_file = b.path("src/tests.zig"), @@ -64,7 +66,6 @@ pub fn build(b: *std.Build) void { }); test_module.addImport("pg", pgz.module("pg")); test_module.addImport("httpz", httpz.module("httpz")); - test_module.addImport("metriks", metriks.module("metriks")); test_module.addImport("dotenv", env.module("dotenv")); test_module.addImport("zul", zul.module("zul")); test_module.addImport("rediz", rediz.module("okredis")); @@ -73,6 +74,7 @@ pub fn build(b: *std.Build) void { test_module.addImport("mqttz", mqttz.module("mqttz")); test_module.addImport("jwt", jwt.module("zig-jwt")); test_module.addImport("sqlite", sqlite.module("sqlite")); + test_module.addImport("nats", nats.module("nats")); test_module.addImport("zero", module); if (builtin.os.tag == .macos) { @@ -83,7 +85,6 @@ pub fn build(b: *std.Build) void { const unit_tests = b.addTest(.{ .root_module = test_module, - .use_llvm = true, }); const test_step = b.step("test", "Run tests"); diff --git a/build.zig.zon b/build.zig.zon index bbf7cea..fcccad9 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -2,61 +2,61 @@ .name = .zero, .version = "0.0.2", .fingerprint = 0xabdef192c03b44cb, - .minimum_zig_version = "0.15.2", + .minimum_zig_version = "0.16.0", .dependencies = .{ - // .dotenv = .{ .path = "../dotenv" }, - // .httpz = .{ .path = "../http.zig" }, - // .metrics = .{ .path = "../metrics.zig" }, + // .jwt = .{ .path = "../zig-jwt" }, + // .regexp = .{ .path = "../zig-regex" }, + // .mqttz = .{ .path = "../mqttz" }, // .pg = .{ .path = "../pg.zig" }, - // .zul = .{ .path = "../zul" }, // .okredis = .{ .path = "../okredis" }, + // .zul = .{ .path = "../zul" }, + // .httpz = .{ .path = "../http.zig" }, // .zdt = .{ .path = "../zdt" }, - // .mqttz = .{ .path = "../mqttz" }, - // .regexp = .{ .path = "../zig-regex" }, - // .jwt = .{ .path = "../zig-jwt" }, - .jwt = .{ - .url = "git+https://github.com/im-ng/zig-jwt.git#0c0fd114a9f3a4f4901fc0eac28821feebfb29f9", - .hash = "zig_jwt-1.3.3-w3q9SOY0BAAfF6i8vq9S4JJR_rw3QSSm8gFQ_RYx3vRQ", + // .dotenv = .{ .path = "../dotenv" }, + // .sqlite = .{ .path = "../zig-sqlite" }, + .zdt = .{ + .url = "git+https://codeberg.org/FObersteiner/zdt.git?ref=v0.9.5#0db1d32005b7058fea0138e97976cc55b58b403d", + .hash = "zdt-0.9.5-xr0_vK7ZEQDFyjfdx_ds5WMEMEUCfUkOnNqLNyvaA8b9", }, - .regexp = .{ - .url = "git+https://github.com/im-ng/zig-regex.git#ee759955e59ae292f4074d62308f953feeecac61", - .hash = "regex-0.1.3-axC352raAQCBo-WcgB_nFMFmO2d3QLBLpE8-ebizdeX_", + .dotenv = .{ + .url = "git+https://github.com/dying-will-bullet/dotenv.git#f6dff7ce2c39ceaafd00114c7d0674376cd82b87", + .hash = "dotenv-0.3.0-hfaNFtm3AACGiIZx5jDPHB4Q1QGs9BEMKH-sq32ZxRXa", }, - .mqttz = .{ - .url = "git+https://github.com/im-ng/mqttz.git#bba199c808200cf51ae9bea86577c646bed627bc", - .hash = "mqttz-0.0.0-ywFE_5QMAwAWoMQe29cn9PMnPf6BYSSiJC-BZPCvw70c", + .httpz = .{ + .url = "git+https://github.com/im-ng/http.zig#a150c6228fdcf5a4ae4cb38f6e0c2e7040acf7cb", + .hash = "httpz-0.0.0-PNVzrPYUCQCsmlsCu1UW6CWzwLann2nDqlwM-olFXom2", }, .pg = .{ - .url = "git+https://github.com/im-ng/pg.zig.git#c187af6760d4a923ff4c4020db76aae3c8b8a273", - .hash = "pg-0.0.0-Wp_7geUtBgC21DUn8zhRP3jDdWwXk3DjUyjFv7301VtI", + .url = "git+https://github.com/karlseguin/pg.zig.git#2e15df7c3b635977bc2733bacd921f3395383860", + .hash = "pg-0.0.0-Wp_7gRK8BgDzp9DPArDotLfTq_aV1NLpNZtWxsOITFdZ", }, .okredis = .{ - .url = "git+https://github.com/im-ng/zig-okredis.git#f53ad9f03a57d41d89b3ee779aaca608e1e4767f", - .hash = "okredis-0.1.0-Cg726vcEBAAjrhwgBJ2cjL4qN3ircjBsX6iTd7wEFiEC", + .url = "git+https://github.com/kristoff-it/zig-okredis.git#c38db4cefd8958dd67730a5965ec247e51a6ef64", + .hash = "okredis-0.1.0-Cg726ncIBAC50n1bNNhRPsT3g27LrKhTttUGu9MHaGJR", }, .zul = .{ - .url = "git+https://github.com/im-ng/zul.git#e770047f208cf4538fcd9fc64550c84fd5492fc4", - .hash = "zul-0.0.0-1oDot9yRBwDA_ovd6GC1M_ViW3LarywMaGrH6vcuEjqv", + .url = "git+https://github.com/karlseguin/zul.git#146f9d5b2238c3a621b96345adf03490900c2fe2", + .hash = "zul-0.0.0-1oDot2KwBwC9c43wC7V9y4-xxg0a_d9okFDcJPhqmije", }, - .httpz = .{ - .url = "git+https://github.com/im-ng/http.zig.git#d21498e2343a4eeab61a57e913f67a3862623d78", - .hash = "httpz-0.0.0-PNVzrEnhBgBwAemUFsKHu7IrGXnIEPAo8oI6HtIaoTan", + .jwt = .{ + .url = "git+https://github.com/deatil/zig-jwt.git#741cb09c871246cc1959cc00c2f758a83fdc3e79", + .hash = "zig_jwt-1.5.15-w3q9SD16BAC6Kuc-1hFYYdoFh9BGo4mhvSSNASQY0EmN", }, - .zdt = .{ - .url = "git+https://codeberg.org/imng/zdt.git#fa55f4080752ce1c48f2bb392e9513f87b59d8cb", - .hash = "zdt-0.8.1-xr0_vABUDwDExZVsT91Qid0tOc_48sI47Xhg6nbFsqqC", + .regexp = .{ + .url = "git+https://github.com/tiehuis/zig-regex.git#49c5694c8b3ee4ae4a7131035a44a69841ef4740", + .hash = "regex-0.1.3-axC357jaAQBRENglwG9NTcuej8pYz1IZmfwER_AXMlHZ", }, - .metriks = .{ - .url = "git+https://github.com/im-ng/metrics.zig.git#2011171f3542f213b4742f96ed0934f56218b8a7", - .hash = "metriks-0.0.0-sq540f2_AQCTNA7_ZQWKehRfXynKHT_0baiw5RFDv_3W", + .mqttz = .{ + .url = "git+https://github.com/karlseguin/mqttz.git#cf8acd2c1acae7f5c1c67b7df53228e7dd8c8c74", + .hash = "mqttz-0.0.0-ywFE_2ZfAwAc6PsPYZq1W7G-EqzyhFPeCbXS_86Qj98V", }, .sqlite = .{ - .url = "git+https://github.com/im-ng/zig-sqlite.git#fb73a6cca771c0c26fbab1a1f7689c23ec786257", - .hash = "sqlite-3.48.0-F2R_a8WODgDamFQx1fOrpgY7IdluD4Sr_P7G0UPxUUMr", + .url = "git+https://github.com/im-ng/zig-sqlite.git#4af3be0231d5dc1765467b87f1d0df51bbeb9121", + .hash = "sqlite-3.48.0-F2R_a0iSDgCgU0ZDXF9fgYM8tauaFdYyt5L9lDYWxjPC", }, - .dotenv = .{ - .url = "git+https://github.com/im-ng/dotenv.git#e5e09fad22499c4bedfb6cb03fb298581de3174f", - .hash = "dotenv-0.2.3-ikMfwFR3AAAdjNnXfM40c1ml45J2cF3axl6p1WGBfcb1", + .nats = .{ + .url = "git+https://github.com/nats-io/nats.zig#883d29c25db420f59133bc5098059191e6ac6b02", + .hash = "nats-0.1.0-991asQCWHwBIwVHZJpXV_b7mll-5F78o_F_vunrfuBm7", }, }, .paths = .{ diff --git a/examples/zero-auth/build.zig.zon b/examples/zero-auth/build.zig.zon index 041dfdc..a503b6f 100644 --- a/examples/zero-auth/build.zig.zon +++ b/examples/zero-auth/build.zig.zon @@ -2,7 +2,7 @@ .name = .authserver, .version = "0.0.1", .fingerprint = 0x1c77681fa9ac8842, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, diff --git a/examples/zero-auth/src/main.zig b/examples/zero-auth/src/main.zig index f2bb466..bfa3456 100644 --- a/examples/zero-auth/src/main.zig +++ b/examples/zero-auth/src/main.zig @@ -6,6 +6,7 @@ const Context = zero.Context; const ClientError = zero.Error.ClientError; const Client = zero.client; const jwtClaims = zero.jwtClaims; +const utils = zero.utils; pub const publicKey = struct { kid: []const u8, @@ -24,12 +25,14 @@ pub const std_options: std.Options = .{ .logFn = zero.logger.custom, }; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app: *App = try App.new(allocator, init.environ_map); try app.get("/basic", basicResponse); diff --git a/examples/zero-basic/build.zig b/examples/zero-basic/build.zig index 1764f4a..7806c14 100644 --- a/examples/zero-basic/build.zig +++ b/examples/zero-basic/build.zig @@ -8,6 +8,7 @@ pub fn build(b: *std.Build) void { const exe = b.addExecutable(.{ .name = "basic", + // .use_llvm = true, .root_module = b.createModule(.{ .root_source_file = b.path("src/main.zig"), .target = target, diff --git a/examples/zero-basic/build.zig.zon b/examples/zero-basic/build.zig.zon index 1e53076..1099df2 100644 --- a/examples/zero-basic/build.zig.zon +++ b/examples/zero-basic/build.zig.zon @@ -2,7 +2,7 @@ .name = .httpserver, .version = "0.0.1", .fingerprint = 0xa328853f6ec9a2ef, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, diff --git a/examples/zero-basic/src/main.zig b/examples/zero-basic/src/main.zig index 36c09e8..5b1a8d1 100644 --- a/examples/zero-basic/src/main.zig +++ b/examples/zero-basic/src/main.zig @@ -12,20 +12,20 @@ pub const std_options: std.Options = .{ .logFn = zero.logger.custom, }; -fn panic(_: []const u8, _: ?*std.builtin.StackTrace, _: ?usize) noreturn { - var it = std.debug.StackIterator.init(@returnAddress(), null); - var ix: usize = 0; +fn panic(msg: []const u8, return_address: ?usize) noreturn { + _ = msg; std.log.err("=== Stack Trace ==============", .{}); - while (it.next()) |frame| : (ix += 1) { - std.log.err("#{d:0>2}: 0x{X:0>16}", .{ ix, frame }); - } + std.debug.dumpCurrentStackTrace(.{ .first_address = return_address }); + std.process.exit(1); } -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); - const app = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try app.get("/", index); diff --git a/examples/zero-cronz/build.zig.zon b/examples/zero-cronz/build.zig.zon index 84b4562..9655d49 100644 --- a/examples/zero-cronz/build.zig.zon +++ b/examples/zero-cronz/build.zig.zon @@ -2,7 +2,7 @@ .name = .zerocronz, .version = "0.0.1", .fingerprint = 0x8e5bb82bf53df309, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, @@ -11,5 +11,4 @@ "build.zig.zon", "src", }, - .use_llvm = true, } diff --git a/examples/zero-cronz/src/main.zig b/examples/zero-cronz/src/main.zig index 002e5c7..2812f4b 100644 --- a/examples/zero-cronz/src/main.zig +++ b/examples/zero-cronz/src/main.zig @@ -9,12 +9,14 @@ pub const std_options: std.Options = .{ .logFn = zero.logger.custom, }; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try app.addCronJob("* * * * * *", "task-1", task1); app.container.log.info("task 1 occurs every 5 seconds of minutes"); diff --git a/examples/zero-kafka-publisher/build.zig.zon b/examples/zero-kafka-publisher/build.zig.zon index 705de11..9d557f1 100644 --- a/examples/zero-kafka-publisher/build.zig.zon +++ b/examples/zero-kafka-publisher/build.zig.zon @@ -2,7 +2,7 @@ .name = .zerokafkapubsub, .version = "0.0.1", .fingerprint = 0xf005043760ca2924, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, @@ -11,5 +11,4 @@ "build.zig.zon", "src", }, - .use_llvm = true, } diff --git a/examples/zero-kafka-publisher/configs/.env b/examples/zero-kafka-publisher/configs/.env index 74de4cc..74d782e 100644 --- a/examples/zero-kafka-publisher/configs/.env +++ b/examples/zero-kafka-publisher/configs/.env @@ -9,8 +9,9 @@ PUBSUB_BROKER="localhost:9092" KAFKA_BATCH_SIZE=1000 KAFKA_BATCH_BYTES=1048576 KAFKA_BATCH_TIMEOUT=300 -# KAFKA_SASL_MECHANISM=PLAINTEXT -KAFKA_SECURITY_PROTOCOL=SASL_PLAINTEXT -KAFKA_SASL_MECHANISM=PLAIN -KAFKA_SASL_USERNAME=admin -KAFKA_SASL_PASSWORD=secret \ No newline at end of file +# switch kafka pods based on this mechanism +KAFKA_SASL_MECHANISM=PLAINTEXT +# KAFKA_SECURITY_PROTOCOL=SASL_PLAINTEXT +# KAFKA_SASL_MECHANISM=PLAIN +# KAFKA_SASL_USERNAME=admin +# KAFKA_SASL_PASSWORD=secret \ No newline at end of file diff --git a/examples/zero-kafka-publisher/src/main.zig b/examples/zero-kafka-publisher/src/main.zig index f383f57..207dc7c 100644 --- a/examples/zero-kafka-publisher/src/main.zig +++ b/examples/zero-kafka-publisher/src/main.zig @@ -17,12 +17,14 @@ const Payload = struct { message: []const u8, }; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try app.addCronJob("* * * * * *", "publisher-1", publishTask1); diff --git a/examples/zero-kafka-subscriber/build.zig.zon b/examples/zero-kafka-subscriber/build.zig.zon index c485134..f08e0f8 100644 --- a/examples/zero-kafka-subscriber/build.zig.zon +++ b/examples/zero-kafka-subscriber/build.zig.zon @@ -2,7 +2,7 @@ .name = .zerokafkapubsub, .version = "0.0.1", .fingerprint = 0xf0050437fca638e1, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, @@ -11,5 +11,4 @@ "build.zig.zon", "src", }, - .use_llvm = true, } diff --git a/examples/zero-kafka-subscriber/configs/.env b/examples/zero-kafka-subscriber/configs/.env index 9e3b34b..fcc42ce 100644 --- a/examples/zero-kafka-subscriber/configs/.env +++ b/examples/zero-kafka-subscriber/configs/.env @@ -2,7 +2,7 @@ APP_NAME=zero-kafka-subscriber APP_VERSION=1.0.0 APP_ENV=dev LOG_LEVEL=debug -HTTP_PORT=8081 +HTTP_PORT=8082 PUBSUB_BACKEND=KAFKA PUBSUB_BROKER="localhost:9092" @@ -10,8 +10,9 @@ CONSUMER_ID="zero-consumer" # KAFKA_BATCH_SIZE=1000 # KAFKA_BATCH_BYTES=1048576 # KAFKA_BATCH_TIMEOUT=300 -# KAFKA_SASL_MECHANISM=PLAINTEXT -KAFKA_SECURITY_PROTOCOL=SASL_PLAINTEXT -KAFKA_SASL_MECHANISM=PLAIN -KAFKA_SASL_USERNAME=admin -KAFKA_SASL_PASSWORD=secret \ No newline at end of file +# switch pods based on this mechanism +KAFKA_SASL_MECHANISM=PLAINTEXT +# KAFKA_SECURITY_PROTOCOL=SASL_PLAINTEXT +# KAFKA_SASL_MECHANISM=PLAIN +# KAFKA_SASL_USERNAME=admin +# KAFKA_SASL_PASSWORD=secret \ No newline at end of file diff --git a/examples/zero-kafka-subscriber/src/main.zig b/examples/zero-kafka-subscriber/src/main.zig index 055b160..55f09aa 100644 --- a/examples/zero-kafka-subscriber/src/main.zig +++ b/examples/zero-kafka-subscriber/src/main.zig @@ -11,12 +11,14 @@ pub const std_options: std.Options = .{ const topicName = "zero-topic"; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try app.addKafkaSubscription(topicName, subscribeTask); @@ -31,10 +33,11 @@ const customMessage = struct { fn subscribeTask(ctx: *Context) !void { const timestamp = try utils.sqlTimestampz(ctx.allocator); //transform ctx.message to custom type in packet read itself - if (ctx.message2) |message| { + if (ctx.message) |message| { + const k = message.kafka; var buffer: []u8 = undefined; buffer = try ctx.allocator.alloc(u8, 1024); - buffer = try std.fmt.bufPrint(buffer, "Received on [{s}] {s}", .{ message.topic, message.payload.? }); + buffer = try std.fmt.bufPrint(buffer, "Received on [{s}] {s}", .{ k.topic, k.payload.? }); ctx.info(timestamp); ctx.info(buffer); diff --git a/examples/zero-migration/build.zig.zon b/examples/zero-migration/build.zig.zon index d74252c..37907d2 100644 --- a/examples/zero-migration/build.zig.zon +++ b/examples/zero-migration/build.zig.zon @@ -2,7 +2,7 @@ .name = .migrations, .version = "0.0.1", .fingerprint = 0x6ea77022efa5d6e6, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, diff --git a/examples/zero-migration/src/main.zig b/examples/zero-migration/src/main.zig index 3444d82..a27c86a 100644 --- a/examples/zero-migration/src/main.zig +++ b/examples/zero-migration/src/main.zig @@ -13,12 +13,14 @@ pub const std_options: std.Options = .{ .logFn = zero.logger.custom, }; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try prepareMigrations(app); diff --git a/examples/zero-mqtt-publisher/build.zig.zon b/examples/zero-mqtt-publisher/build.zig.zon index 37887e0..487ffe6 100644 --- a/examples/zero-mqtt-publisher/build.zig.zon +++ b/examples/zero-mqtt-publisher/build.zig.zon @@ -2,7 +2,7 @@ .name = .zeropubsub, .version = "0.0.1", .fingerprint = 0x8be215153b2ae7f1, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, @@ -11,5 +11,4 @@ "build.zig.zon", "src", }, - .use_llvm = true, } diff --git a/examples/zero-mqtt-publisher/src/main.zig b/examples/zero-mqtt-publisher/src/main.zig index 12ce770..3f0c5f4 100644 --- a/examples/zero-mqtt-publisher/src/main.zig +++ b/examples/zero-mqtt-publisher/src/main.zig @@ -12,12 +12,14 @@ pub const std_options: std.Options = .{ const pubSubTopic = "zero"; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try app.addCronJob("* * * * * *", "publisher-1", publishTask1); diff --git a/examples/zero-mqtt-subscriber/build.zig.zon b/examples/zero-mqtt-subscriber/build.zig.zon index 37887e0..487ffe6 100644 --- a/examples/zero-mqtt-subscriber/build.zig.zon +++ b/examples/zero-mqtt-subscriber/build.zig.zon @@ -2,7 +2,7 @@ .name = .zeropubsub, .version = "0.0.1", .fingerprint = 0x8be215153b2ae7f1, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, @@ -11,5 +11,4 @@ "build.zig.zon", "src", }, - .use_llvm = true, } diff --git a/examples/zero-mqtt-subscriber/src/main.zig b/examples/zero-mqtt-subscriber/src/main.zig index 3e096c0..a5a5963 100644 --- a/examples/zero-mqtt-subscriber/src/main.zig +++ b/examples/zero-mqtt-subscriber/src/main.zig @@ -11,12 +11,14 @@ pub const std_options: std.Options = .{ const pubSubTopic = "zero"; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try app.addSubscription(pubSubTopic, subscribeTask); @@ -38,9 +40,10 @@ fn subscribeTask(ctx: *Context) !void { //transform ctx.message to custom type in packet read itself if (ctx.message) |message| { + const mq = message.mqtt; m = customMessage{}; - m.msg = message.payload.?; - m.topic = message.topic; + m.msg = mq.payload.?; + m.topic = mq.topic; var buffer: []u8 = undefined; buffer = try ctx.allocator.alloc(u8, 1024); diff --git a/examples/zero-nats-publisher/build.zig b/examples/zero-nats-publisher/build.zig new file mode 100644 index 0000000..75cc99b --- /dev/null +++ b/examples/zero-nats-publisher/build.zig @@ -0,0 +1,46 @@ +const std = @import("std"); + +pub fn build(b: *std.Build) void { + const target = b.standardTargetOptions(.{}); + const optimize = b.standardOptimizeOption(.{}); + + const zero = b.dependency("zero", .{}); + + const exe = b.addExecutable(.{ + .name = "pubsub", + .root_module = b.createModule(.{ + .root_source_file = b.path("src/main.zig"), + .target = target, + .optimize = optimize, + }), + }); + + exe.root_module.addImport("zero", zero.module("zero")); + + b.installArtifact(exe); + + const run_cmd = b.addRunArtifact(exe); + run_cmd.step.dependOn(b.getInstallStep()); + if (b.args) |args| { + run_cmd.addArgs(args); + } + + const run_step = b.step("pubsub", "Run zero nats pubsub..."); + run_step.dependOn(&run_cmd.step); + + const test_module = b.createModule(.{ + .root_source_file = b.path("src/t.zig"), + .target = target, + .optimize = optimize, + }); + test_module.addImport("test", zero.module("zero")); + + const unit_tests = b.addTest(.{ + .root_module = test_module, + }); + unit_tests.root_module.addImport("zero", zero.module("zero")); + + const run_exe_tests = b.addRunArtifact(unit_tests); + const test_step = b.step("test", "Run tests"); + test_step.dependOn(&run_exe_tests.step); +} diff --git a/examples/zero-nats-publisher/build.zig.zon b/examples/zero-nats-publisher/build.zig.zon new file mode 100644 index 0000000..539213f --- /dev/null +++ b/examples/zero-nats-publisher/build.zig.zon @@ -0,0 +1,14 @@ +.{ + .name = .zeronatspubsub, + .version = "0.0.1", + .fingerprint = 0x8cb2f601fdc6d999, + .minimum_zig_version = "0.16.0", + .dependencies = .{ + .zero = .{ .path = "../../." }, + }, + .paths = .{ + "build.zig", + "build.zig.zon", + "src", + }, +} diff --git a/examples/zero-nats-publisher/configs/.env b/examples/zero-nats-publisher/configs/.env new file mode 100644 index 0000000..92ac1d7 --- /dev/null +++ b/examples/zero-nats-publisher/configs/.env @@ -0,0 +1,14 @@ +APP_NAME=zero-nats-publisher +APP_VERSION=1.0.0 +APP_ENV=dev +LOG_LEVEL=debug +HTTP_PORT=8092 + +PUBSUB_BACKEND=NATS +PUBSUB_BROKER=nats://localhost:4222 +NATS_STREAM=zero-stream +NATS_SUBJECTS=zero +NATS_MAX_WAIT=5000 +NATS_MAX_PULL_WAIT=5000 +NATS_CONSUMER=zero-consumer +# NATS_CREDS_FILE=/path/to/creds.creds diff --git a/examples/zero-nats-publisher/src/main.zig b/examples/zero-nats-publisher/src/main.zig new file mode 100644 index 0000000..02b574b --- /dev/null +++ b/examples/zero-nats-publisher/src/main.zig @@ -0,0 +1,41 @@ +const std = @import("std"); +const zero = @import("zero"); + +const Allocator = std.mem.Allocator; +const App = zero.App; +const Context = zero.Context; +const utils = zero.utils; + +pub const std_options: std.Options = .{ + .logFn = zero.logger.custom, +}; + +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; + const allocator = gpa.allocator(); + _ = gpa.detectLeaks(); + + const app = try App.new(allocator, init.environ_map); + + try app.addCronJob("* * * * * *", "publisher-1", publishTask1); + + try app.addCronJob("*/10 * * * * *", "publisher-2", publishTask2); + + try app.run(); +} + +fn publishTask1(ctx: *Context) !void { + const timestamp = try utils.sqlTimestampz(ctx.allocator); + try ctx.pubsub.Publish("zero", "publisher 1 says hello! via NATS"); + + ctx.info(timestamp); +} + +fn publishTask2(ctx: *Context) !void { + const timestamp = try utils.sqlTimestampz(ctx.allocator); + try ctx.pubsub.Publish("zero", "publisher 2 says hi! via NATS"); + + ctx.info(timestamp); +} diff --git a/examples/zero-nats-subscriber/build.zig b/examples/zero-nats-subscriber/build.zig new file mode 100644 index 0000000..6cc52e9 --- /dev/null +++ b/examples/zero-nats-subscriber/build.zig @@ -0,0 +1,46 @@ +const std = @import("std"); + +pub fn build(b: *std.Build) void { + const target = b.standardTargetOptions(.{}); + const optimize = b.standardOptimizeOption(.{}); + + const zero = b.dependency("zero", .{}); + + const exe = b.addExecutable(.{ + .name = "pubsub", + .root_module = b.createModule(.{ + .root_source_file = b.path("src/main.zig"), + .target = target, + .optimize = optimize, + }), + }); + + exe.root_module.addImport("zero", zero.module("zero")); + + b.installArtifact(exe); + + const run_cmd = b.addRunArtifact(exe); + run_cmd.step.dependOn(b.getInstallStep()); + if (b.args) |args| { + run_cmd.addArgs(args); + } + + const run_step = b.step("pubsub", "Run zero nats subscriber..."); + run_step.dependOn(&run_cmd.step); + + const test_module = b.createModule(.{ + .root_source_file = b.path("src/t.zig"), + .target = target, + .optimize = optimize, + }); + test_module.addImport("test", zero.module("zero")); + + const unit_tests = b.addTest(.{ + .root_module = test_module, + }); + unit_tests.root_module.addImport("zero", zero.module("zero")); + + const run_exe_tests = b.addRunArtifact(unit_tests); + const test_step = b.step("test", "Run tests"); + test_step.dependOn(&run_exe_tests.step); +} diff --git a/examples/zero-nats-subscriber/build.zig.zon b/examples/zero-nats-subscriber/build.zig.zon new file mode 100644 index 0000000..b88d820 --- /dev/null +++ b/examples/zero-nats-subscriber/build.zig.zon @@ -0,0 +1,14 @@ +.{ + .name = .zeronatssubpub, + .version = "0.0.1", + .fingerprint = 0x8603af67a76a159, + .minimum_zig_version = "0.16.0", + .dependencies = .{ + .zero = .{ .path = "../../." }, + }, + .paths = .{ + "build.zig", + "build.zig.zon", + "src", + }, +} diff --git a/examples/zero-nats-subscriber/configs/.env b/examples/zero-nats-subscriber/configs/.env new file mode 100644 index 0000000..8f5fbb8 --- /dev/null +++ b/examples/zero-nats-subscriber/configs/.env @@ -0,0 +1,14 @@ +APP_NAME=zero-nats-subscriber +APP_VERSION=1.0.0 +APP_ENV=dev +LOG_LEVEL=debug +HTTP_PORT=8093 + +PUBSUB_BACKEND=NATS +PUBSUB_BROKER=nats://localhost:4222 +NATS_STREAM=zero-stream +NATS_SUBJECTS=zero +NATS_MAX_WAIT=5000 +NATS_MAX_PULL_WAIT=5000 +NATS_CONSUMER=zero-consumer +# NATS_CREDS_FILE=/path/to/creds.creds diff --git a/examples/zero-nats-subscriber/src/main.zig b/examples/zero-nats-subscriber/src/main.zig new file mode 100644 index 0000000..72d28a2 --- /dev/null +++ b/examples/zero-nats-subscriber/src/main.zig @@ -0,0 +1,51 @@ +const std = @import("std"); +const zero = @import("zero"); + +const Allocator = std.mem.Allocator; +const App = zero.App; +const Context = zero.Context; +const utils = zero.utils; + +pub const std_options: std.Options = .{ + .logFn = zero.logger.custom, +}; + +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; + const allocator = gpa.allocator(); + _ = gpa.detectLeaks(); + + const app = try App.new(allocator, init.environ_map); + + try app.get("/", index); + + try app.addPubSubSubscription("zero", onMessage); + + try app.run(); +} + +fn index(ctx: *Context) !void { + ctx.response.setStatus(.ok); + ctx.response.body = + \\ NATS Subscriber Demo - Zero Framework + \\ ================================ + \\ + \\ Subscribed to subject: zero + \\ Messages are logged as they arrive. + ; +} + +fn onMessage(ctx: *Context) !void { + if (ctx.message) |message| { + const m = message.nats; + var buffer: [256]u8 = undefined; + const msg = std.fmt.bufPrint( + &buffer, + "received on [{s}] {s}", + .{ m.subject, m.payload }, + ) catch "decode error"; + ctx.info(msg); + } +} diff --git a/examples/zero-redis/build.zig.zon b/examples/zero-redis/build.zig.zon index ac43362..abb1b38 100644 --- a/examples/zero-redis/build.zig.zon +++ b/examples/zero-redis/build.zig.zon @@ -2,7 +2,7 @@ .name = .zeroredis, .version = "0.0.1", .fingerprint = 0x24dda8bd11e0e43f, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, diff --git a/examples/zero-redis/src/main.zig b/examples/zero-redis/src/main.zig index cc78a61..cf79c98 100644 --- a/examples/zero-redis/src/main.zig +++ b/examples/zero-redis/src/main.zig @@ -4,17 +4,20 @@ const zero = @import("zero"); const App = zero.App; const Context = zero.Context; const redis = zero.rediz; +const utils = zero.utils; pub const std_options: std.Options = .{ .logFn = zero.logger.custom, }; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); app.onStatup(prepareCache); diff --git a/examples/zero-service-client/build.zig.zon b/examples/zero-service-client/build.zig.zon index af6a2c2..4ffcbe9 100644 --- a/examples/zero-service-client/build.zig.zon +++ b/examples/zero-service-client/build.zig.zon @@ -2,7 +2,7 @@ .name = .serviceclient, .version = "0.0.1", .fingerprint = 0xe40e052ef8f4f884, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, diff --git a/examples/zero-service-client/src/main.zig b/examples/zero-service-client/src/main.zig index 6e90048..3e78277 100644 --- a/examples/zero-service-client/src/main.zig +++ b/examples/zero-service-client/src/main.zig @@ -23,12 +23,14 @@ pub const publicKeys = struct { keys: []publicKey, }; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try app.addHttpService("auth-service", app.config.get("SERVICE_URL")); diff --git a/examples/zero-sqlite/build.zig.zon b/examples/zero-sqlite/build.zig.zon index 3f0640f..67f0855 100644 --- a/examples/zero-sqlite/build.zig.zon +++ b/examples/zero-sqlite/build.zig.zon @@ -2,7 +2,7 @@ .name = .zerosqlite, .version = "0.0.1", .fingerprint = 0x8ee7340c93003d6e, - .minimum_zig_version = "0.15.2", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../.." }, }, diff --git a/examples/zero-sqlite/src/main.zig b/examples/zero-sqlite/src/main.zig index 75075a6..2b8bb1e 100644 --- a/examples/zero-sqlite/src/main.zig +++ b/examples/zero-sqlite/src/main.zig @@ -32,11 +32,14 @@ const CreateUser = struct { email: []const u8, }; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); + _ = gpa.detectLeaks(); - const app = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); try app.get("/", index); try app.get("/sqlite/init", sqliteInit); diff --git a/examples/zero-stream/build.zig.zon b/examples/zero-stream/build.zig.zon index d5fea5c..d06ed28 100644 --- a/examples/zero-stream/build.zig.zon +++ b/examples/zero-stream/build.zig.zon @@ -2,7 +2,7 @@ .name = .stream, .version = "0.0.1", .fingerprint = 0xf0e9be1c512e7027, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, diff --git a/examples/zero-stream/src/main.zig b/examples/zero-stream/src/main.zig index 848b89a..a007c60 100644 --- a/examples/zero-stream/src/main.zig +++ b/examples/zero-stream/src/main.zig @@ -9,19 +9,21 @@ const Process = zero.process; const Host = zero.host; const utils = zero.utils; const Builder = zero.zul.StringBuilder; -var mutex: std.Thread.Mutex = .{}; +var mutex: std.Io.Mutex = .init; var connections: std.hash_map.StringHashMap(?*zero.WSClient) = undefined; pub const std_options: std.Options = .{ .logFn = zero.logger.custom, }; -pub fn main() !void { - var gpa = std.heap.GeneralPurposeAllocator(.{}){}; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); - defer _ = gpa.detectLeaks(); + _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app = try App.new(allocator, init.environ_map); connections = std.hash_map.StringHashMap(?*zero.WSClient).init(allocator); @@ -39,8 +41,8 @@ pub fn main() !void { } pub fn connect(ctx: *Context) !void { - mutex.lock(); - defer mutex.unlock(); + mutex.lock(utils.io) catch {}; + defer mutex.unlock(utils.io); try connections.put(ctx.request.header("sec-websocket-key").?, ctx.wsClient); } diff --git a/examples/zero-todo-htmx/build.zig b/examples/zero-todo-htmx/build.zig index 866ee26..dae90d9 100644 --- a/examples/zero-todo-htmx/build.zig +++ b/examples/zero-todo-htmx/build.zig @@ -8,6 +8,8 @@ pub fn build(b: *std.Build) void { const exe = b.addExecutable(.{ .name = "todo", + .use_lld = true, + .use_llvm = true, .root_module = b.createModule(.{ .root_source_file = b.path("src/main.zig"), .target = target, @@ -15,6 +17,12 @@ pub fn build(b: *std.Build) void { }), }); + const debug = b.option(bool, "debug", "enable code debug mode") orelse false; + if (debug) { + exe.use_lld = true; + exe.use_llvm = true; + } + exe.root_module.addImport("zero", zero.module("zero")); b.installArtifact(exe); diff --git a/examples/zero-todo-htmx/build.zig.zon b/examples/zero-todo-htmx/build.zig.zon index 64f0bae..38ee109 100644 --- a/examples/zero-todo-htmx/build.zig.zon +++ b/examples/zero-todo-htmx/build.zig.zon @@ -2,7 +2,7 @@ .name = .todo, .version = "0.0.1", .fingerprint = 0x5a0eb6a0f95b9c4a, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, diff --git a/examples/zero-todo-htmx/src/handler.zig b/examples/zero-todo-htmx/src/handler.zig index cc6d743..14d41db 100644 --- a/examples/zero-todo-htmx/src/handler.zig +++ b/examples/zero-todo-htmx/src/handler.zig @@ -14,10 +14,8 @@ const Context = zero.Context; const utils = zero.utils; pub fn getAll(ctx: *Context) !void { - var rows = try ctx.SQL.queryRows(models.getAllTodos, .{}); - defer rows.deinit(); - - // var res = rows.mapper(models.Todo, .{ .dupe = true }); + var _rows = std.array_list.Managed(models.Todo).init(ctx.allocator); + _ = try ctx.SQL.selectSlice(models.Todo, &_rows, models.getAllTodos, .{}); var responses = std.array_list.Managed( models.HandlerTodo, @@ -25,17 +23,15 @@ pub fn getAll(ctx: *Context) !void { ctx.allocator, ); - while (try rows.next()) |row| { - const todo = try row.to(models.Todo, .{}); - + for (_rows.items) |row| { const response = models.HandlerTodo{ - .id = try std.fmt.allocPrint(ctx.allocator, "{d}", .{todo.id.?}), - .description = todo.description, - .task = todo.task, - .isDone = todo.isDone, + .id = try std.fmt.allocPrint(ctx.allocator, "{d}", .{row.id.?}), + .description = row.description, + .task = row.task, + .isDone = row.is_done, .created_at = try utils.DTtimestampz( ctx.allocator, - todo.created_at, + row.created_at, ), }; @@ -61,7 +57,8 @@ pub fn getTodo(ctx: *Context) !void { return; } - var row = ctx.SQL.queryRow( + const row: ?models.Todo = ctx.SQL.select( + models.Todo, models.getTodoByID, .{id}, ) catch |err| { @@ -83,17 +80,15 @@ pub fn getTodo(ctx: *Context) !void { return; } - defer row.?.deinit() catch {}; - - const res = try row.?.to(models.Todo, .{}); + // const res = try row.?.to(models.Todo, .{}); var response = models.HandlerTodo{ - .id = try std.fmt.allocPrint(ctx.allocator, "{d}", .{res.id.?}), - .description = res.description, - .task = res.task, - .isDone = res.isDone, + .id = try std.fmt.allocPrint(ctx.allocator, "{d}", .{row.?.id.?}), + .description = row.?.description, + .task = row.?.task, + .isDone = row.?.is_done, }; - response.created_at = try utils.DTtimestampz(ctx.allocator, res.created_at); + response.created_at = try utils.DTtimestampz(ctx.allocator, row.?.created_at); const list = try helper.getEditItem(ctx, &response); @@ -121,27 +116,25 @@ pub fn persistTodo(ctx: *Context) !void { ctx.info(status); } - var row = try ctx.SQL.queryRow( + const row: ?models.Todo = try ctx.SQL.select( + models.Todo, models.getTodoEntry, .{}, - ) orelse unreachable; - defer row.deinit() catch {}; - - const res = try row.to(models.Todo, .{}); + ); var response = models.HandlerTodo{ .id = try std.fmt.allocPrint( ctx.allocator, "{d}", - .{res.id.?}, + .{row.?.id.?}, ), - .description = res.description, - .task = res.task, - .isDone = res.isDone, + .description = row.?.description, + .task = row.?.task, + .isDone = row.?.is_done, }; response.created_at = try utils.DTtimestampz( ctx.allocator, - res.created_at, + row.?.created_at, ); ctx.response.setStatus(.ok); @@ -201,27 +194,26 @@ pub fn updateTodo(ctx: *Context) !void { ctx.info(status); } - var row = try ctx.SQL.queryRow( + const row: ?models.Todo = try ctx.SQL.select( + models.Todo, models.getTodoByID, .{todoID}, - ) orelse unreachable; - defer row.deinit() catch {}; - - const res = try row.to(models.Todo, .{}); + ); + // const res = try row.to(models.Todo, .{}); var response = models.HandlerTodo{ .id = try std.fmt.allocPrint( ctx.allocator, "{d}", - .{res.id.?}, + .{row.?.id.?}, ), - .description = res.description, - .task = res.task, - .isDone = res.isDone, + .description = row.?.description, + .task = row.?.task, + .isDone = row.?.is_done, }; response.created_at = try utils.DTtimestampz( ctx.allocator, - res.created_at, + row.?.created_at, ); var sb = Builder.init(ctx.allocator); @@ -262,7 +254,7 @@ pub fn markDone(ctx: *Context) !void { ), .description = res.description, .task = res.task, - .isDone = res.isDone, + .isDone = res.is_done, }; response.created_at = try utils.DTtimestampz( ctx.allocator, @@ -307,7 +299,7 @@ pub fn markUndone(ctx: *Context) !void { ), .description = res.description, .task = res.task, - .isDone = res.isDone, + .isDone = res.is_done, }; response.created_at = try utils.DTtimestampz(ctx.allocator, res.created_at); diff --git a/examples/zero-todo-htmx/src/main.zig b/examples/zero-todo-htmx/src/main.zig index ef82eb0..7b99965 100644 --- a/examples/zero-todo-htmx/src/main.zig +++ b/examples/zero-todo-htmx/src/main.zig @@ -16,12 +16,14 @@ pub const std_options: std.Options = .{ .logFn = zero.logger.custom, }; -pub fn main() !void { - var gpa: std.heap.GeneralPurposeAllocator(.{}) = .init; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app: *App = try App.new(allocator, init.environ_map); try migrations.all(app); diff --git a/examples/zero-todo-htmx/src/models.zig b/examples/zero-todo-htmx/src/models.zig index 5dea42a..5acfbdb 100644 --- a/examples/zero-todo-htmx/src/models.zig +++ b/examples/zero-todo-htmx/src/models.zig @@ -18,7 +18,7 @@ pub const Todo = struct { id: ?i32 = 0, task: ?[]const u8 = undefined, description: ?[]const u8 = undefined, - isDone: ?bool = undefined, + is_done: ?bool = undefined, created_at: ?i64 = undefined, }; diff --git a/examples/zero-websocket/build.zig.zon b/examples/zero-websocket/build.zig.zon index 7cd3cc9..4f633ec 100644 --- a/examples/zero-websocket/build.zig.zon +++ b/examples/zero-websocket/build.zig.zon @@ -2,7 +2,7 @@ .name = .websocket, .version = "0.0.1", .fingerprint = 0x42ce80b977d7d790, - .minimum_zig_version = "0.15.1", + .minimum_zig_version = "0.16.0", .dependencies = .{ .zero = .{ .path = "../../." }, }, diff --git a/examples/zero-websocket/src/main.zig b/examples/zero-websocket/src/main.zig index 3ed7ace..d1b05f1 100644 --- a/examples/zero-websocket/src/main.zig +++ b/examples/zero-websocket/src/main.zig @@ -3,6 +3,7 @@ const zero = @import("zero"); const App = zero.App; const Context = zero.Context; +const utils = zero.utils; pub const std_options: std.Options = .{ .logFn = zero.logger.custom, @@ -17,12 +18,14 @@ fn panic(_: []const u8, _: ?*std.builtin.StackTrace, _: ?usize) noreturn { } } -pub fn main() !void { - var gpa = std.heap.GeneralPurposeAllocator(.{}){}; +pub fn main(init: std.process.Init) !void { + utils.setIo(init.io); + + var gpa: std.heap.DebugAllocator(.{}) = .init; const allocator = gpa.allocator(); - defer _ = gpa.detectLeaks(); + _ = gpa.detectLeaks(); - const app: *App = try App.new(allocator); + const app: *App = try App.new(allocator, init.environ_map); try app.addWebsocket(socketHandler); @@ -32,6 +35,9 @@ pub fn main() !void { pub fn socketHandler(ctx: *Context) !void { if (ctx.wsMessage) |msg| { ctx.info(msg); + + try ctx.wsClient.write(msg); + return; } try ctx.wsClient.write("hello!"); diff --git a/src/app.zig b/src/app.zig index 4b9bd98..ce2933a 100644 --- a/src/app.zig +++ b/src/app.zig @@ -1,5 +1,7 @@ const std = @import("std"); const root = @import("zero.zig"); +const EnvMap = std.process.Environ.Map; + const App = @This(); const Self = @This(); const httpz = root.httpz; @@ -23,6 +25,7 @@ pub const swaggerUIBundlerPreset = root.swaggerUIBundlerPreset; pub const swaggerUICss = root.swaggerUICss; pub const swaggerUIJs = root.swaggerUIJs; +envMap: *EnvMap = undefined, log: *root.logger = undefined, config: *root.config = undefined, container: *root.container = undefined, @@ -35,7 +38,7 @@ startupHook: ?*const fn (*root.Context) anyerror!void = null, var hServer: ?*root.httpServer = undefined; var AppInstance: *Self = undefined; -pub fn new(allocator: std.mem.Allocator) !*App { +pub fn new(allocator: std.mem.Allocator, em: *EnvMap) !*App { const app = try allocator.create(App); errdefer allocator.destroy(app); @@ -44,6 +47,7 @@ pub fn new(allocator: std.mem.Allocator) !*App { const config = try root.config.create(.{ .allocator = allocator, .log = log, + .environments = em, }); // reset log level @@ -183,11 +187,15 @@ fn startPubSubSubscriptions(self: Self) !void { } if (self.container.Kakfa) |k| { - if (k.kafkaMode != root.rdkafka.RD_KAFKA_CONSUMER) { - return; + if (k.kafkaMode == root.rdkafka.RD_KAFKA_CONSUMER) { + self.container.log.info("starting kafka subscriptions"); + try k.startSubscription(); } - self.container.log.info("starting kafka subscriptions"); - try k.startSubscription(); + } + + if (self.container.Nats) |n| { + self.container.log.info("starting nats subscriptions"); + try n.startSubscription(); } } @@ -207,13 +215,13 @@ fn startShutdownHandler(_: Self) !void { }, null); } -fn shutdown(_: c_int) callconv(.c) void { +fn shutdown(_: std.os.linux.SIG) callconv(.c) void { if (AppInstance.cronz) |cronz| { cronz.destroy(); AppInstance.log.info("cleaning running cronz"); } - std.Thread.sleep(1_000_000_000); + std.Io.sleep(utils.io, std.Io.Duration.fromSeconds(1), .awake) catch {}; if (hServer) |h| { h.shutdown(); @@ -268,7 +276,7 @@ pub fn prepareHttpServer(self: Self) !std.Thread { } fn favIcon(ctx: *Context) !void { - var f = std.fs.cwd().openFile(constants.FAVICON_FILE_PATH, .{}) catch |err| switch (err) { + var f = std.Io.Dir.cwd().openFile(utils.io, constants.FAVICON_FILE_PATH, .{}) catch |err| switch (err) { else => { var buffer: []u8 = try ctx.allocator.alloc(u8, 100); buffer = try std.fmt.bufPrint(buffer, "favorite icon not found, using default", .{}); @@ -281,10 +289,10 @@ fn favIcon(ctx: *Context) !void { return; }, }; - defer f.close(); + defer f.close(utils.io); // Read the file into a buffer. - const stat = f.stat() catch |err| { + const stat = f.stat(utils.io) catch |err| { var buffer: []u8 = try ctx.allocator.alloc(u8, 100); buffer = try std.fmt.bufPrint(buffer, "favorite icon not found, using default {s}", .{ @errorName(err), @@ -298,19 +306,8 @@ fn favIcon(ctx: *Context) !void { return; }; - const buffer = f.readToEndAlloc(ctx.allocator, stat.size) catch |err| { - var buffer: []u8 = try ctx.allocator.alloc(u8, 100); - buffer = try std.fmt.bufPrint(buffer, "favorite icon not found, using default {s}", .{ - @errorName(err), - }); - ctx.info(buffer); - - ctx.response.setStatus(.ok); - ctx.response.content_type = .ICO; - ctx.response.body = favoriteIcon; - - return; - }; + const buffer = try ctx.allocator.alloc(u8, stat.size); + _ = try f.readPositionalAll(utils.io, buffer, 0); ctx.response.setStatus(.ok); ctx.response.content_type = .ICO; @@ -318,17 +315,13 @@ fn favIcon(ctx: *Context) !void { } fn readFile(ctx: *Context, path: []const u8) ![]const u8 { - var filePath: []u8 = undefined; - filePath = try ctx.allocator.alloc(u8, 100); - filePath = try std.fs.cwd().realpath(path, filePath); - - var f = try std.fs.cwd().openFile(filePath, .{}); - defer f.close(); + var f = try std.Io.Dir.cwd().openFile(utils.io, path, .{}); + defer f.close(utils.io); // Read the file into a buffer. - const stat = try f.stat(); - - const buffer = f.readToEndAlloc(ctx.allocator, stat.size); + const stat = try f.stat(utils.io); + const buffer = try ctx.allocator.alloc(u8, stat.size); + _ = try f.readPositionalAll(utils.io, buffer, 0); return buffer; } @@ -505,6 +498,25 @@ pub fn addKafkaSubscription(self: *Self, topic: []const u8, hook: fn (*root.Cont try self.container.Kakfa.?.addSubscriber(topic, hook); } +pub fn addNatsSubscription(self: *Self, topic: []const u8, hook: fn (*root.Context) anyerror!void) !void { + if (self.container.Nats == null) { + self.container.log.err("pubsub is disabled, topic subscription is not available."); + return; + } + + try self.container.Nats.?.addSubscriber(topic, hook); +} + +/// Subscribe through the unified PubSub interface (backend-agnostic). +pub fn addPubSubSubscription(self: *Self, topic: []const u8, hook: fn (*root.Context) anyerror!void) !void { + if (self.container.pubSub == null) { + self.container.log.err("pubsub is disabled, topic subscription is not available."); + return; + } + + try self.container.pubSub.?.addSubscriber(topic, hook); +} + pub fn addOAuthKeyRefresher(self: *Self) anyerror!void { if (self.httpServer.provider == null) { return; diff --git a/src/config.zig b/src/config.zig index 4fa7a35..30e57fb 100644 --- a/src/config.zig +++ b/src/config.zig @@ -1,4 +1,5 @@ const std = @import("std"); +const builtin = @import("builtin"); const root = @import("zero.zig"); const dotenv = root.dotenv; const constants = root.constants; @@ -7,10 +8,14 @@ const utils = root.utils; const config = @This(); const Self = @This(); +/// Process environment, set once at startup via `setEnviron` (from +/// `std.process.Init.environ`). Under `zig build test`, `std.testing.environ` +/// is used instead. const defaultPath = "./configs"; const defaultFile = "./configs/.env"; -// const defaultFile = "/media/ng/home/zig-self-learning/zero/examples/zero-kafka-subscriber/configs/.env"; +// const defaultFile = "/media/ng/home/zig-self-learning/zero/examples/zero-todo-htmx/configs/.env"; +environments: *std.process.Environ.Map, allocator: std.mem.Allocator, log: *root.logger, @@ -21,6 +26,7 @@ pub fn create(self: Self) !*config { c.* = .{ .allocator = self.allocator, .log = self.log, + .environments = self.environments, }; try loadDefaultEnv(c); @@ -42,7 +48,7 @@ fn isFileRWExist(fn_dir: std.fs.Dir, fn_file_name: []const u8) !bool { } fn loadDefaultEnv(self: *Self) !void { - try dotenv.loadFrom(self.allocator, defaultFile, .{}); + try dotenv.loadFrom(self.allocator, utils.io, self.environments, defaultFile, .{}); const msg = try utils.combine(self.allocator, "Loaded config from file: {s}", .{defaultFile}); self.log.Info(self.allocator, msg); } @@ -57,7 +63,7 @@ fn loadEnvironmentOverrides(self: *Self) !void { finalEnvFile = defaultFile; } - dotenv.loadFrom(self.allocator, finalEnvFile, .{ .override = true }) catch |err| switch (err) { + dotenv.loadFrom(self.allocator, utils.io, self.environments, finalEnvFile, .{ .override = true }) catch |err| switch (err) { error.FileNotFound => { const msg = try utils.combine(self.allocator, "config overriden {s} file not found.", .{finalEnvFile}); self.log.info(msg); @@ -99,8 +105,11 @@ pub fn getIntByType(self: *Self, key: []const u8, comptime T: type) !T { return integer; } -pub fn getOrDefault(_: *Self, key: []const u8, default: []const u8) []const u8 { - const value = std.posix.getenv(key); +pub fn getOrDefault(self: *Self, key: []const u8, default: []const u8) []const u8 { + const value = if (builtin.is_test) + std.testing.environ.getPosix(key) + else + self.environments.get(key); if (value == null) { return default; } diff --git a/src/container.zig b/src/container.zig index 5f94b83..fed240c 100644 --- a/src/container.zig +++ b/src/container.zig @@ -14,6 +14,7 @@ const rdzDatasource = root.rdz; const zeroClient = root.client; const MQTT = root.MQTT; const mqConfig = root.mqConfig; +const natsConfig = root.natsConfig; const rdkafka = root.rdkafka; const kafka = root.kafka; const utils = root.utils; @@ -34,6 +35,8 @@ SQLite: ?*root.SQLite = undefined, services: ?std.StringHashMap(*zeroClient) = undefined, pubsub: ?*root.MQTT = null, Kakfa: ?*root.kafka = null, +Nats: ?*root.nats = null, +pubSub: ?*root.PubSub = null, pub fn create(self: Self) anyerror!*container { const c = try self.allocator.create(container); @@ -113,6 +116,8 @@ fn loadPubSub(self: *Self) !void { try self.loadKafkaPubSub(); } else if (std.mem.eql(u8, "MQTT", pubsub)) { try self.loadMqttPubSub(); + } else if (std.mem.eql(u8, "NATS", pubsub)) { + try self.loadNatsPubSub(); } else { buffer = try std.fmt.bufPrint(buffer, "pubsub is disabled, as pubsub mode is not provided.", .{}); self.log.debug(buffer); @@ -338,6 +343,11 @@ fn loadKafkaPubSub(self: *Self) !void { //do nothing }, } + + // build the unified PubSub dispatcher + const ps = try self.allocator.create(root.PubSub); + ps.* = .{ .ptr = @ptrCast(@alignCast(self.Kakfa)), .vtable = &root.kafka.vtable }; + self.pubSub = ps; } fn loadMqttPubSub(self: *Self) !void { @@ -412,8 +422,60 @@ fn loadMqttPubSub(self: *Self) !void { buffer = try self.allocator.alloc(u8, 256); buffer = try std.fmt.bufPrint(buffer, "connected to MQTT at '{s}:{d}'", .{ hostname, portAsInt }); self.log.info(buffer); + + // build the unified PubSub dispatcher + const ps = try self.allocator.create(root.PubSub); + ps.* = .{ .ptr = @ptrCast(@alignCast(self.pubsub)), .vtable = &root.MQTT.vtable }; + self.pubSub = ps; } +fn loadNatsPubSub(self: *Self) !void { + var buffer: []u8 = undefined; + buffer = try self.allocator.alloc(u8, 512); + + const url = self.config.get("PUBSUB_BROKER"); + if (std.mem.eql(u8, url, "") == true) { + buffer = try std.fmt.bufPrint(buffer, "pubsub is disabled, as nats broker is not provided.", .{}); + self.log.debug(buffer); + return; + } + + const stream = self.config.get("NATS_STREAM"); + const subjects = self.config.getOrDefault("NATS_SUBJECTS", ""); + const max_wait = try self.config.getAsInt("NATS_MAX_WAIT"); + const max_pull_wait = try self.config.getAsInt("NATS_MAX_PULL_WAIT"); + const consumer = self.config.get("NATS_CONSUMER"); + const creds_file = self.config.get("NATS_CREDS_FILE"); + + const config = natsConfig{ + .url = url, + .stream = stream, + .subjects = subjects, + .max_wait_ms = @intCast(max_wait), + .max_pull_wait_ms = @intCast(max_pull_wait), + .consumer = consumer, + .creds_file = creds_file, + }; + + self.Nats = root.nats.create(self, &config) catch |err| { + buffer = try self.allocator.alloc(u8, 256); + buffer = try std.fmt.bufPrint(buffer, "could not connect to NATS at '{s}'", .{url}); + self.log.err(buffer); + self.log.any(err); + return; + }; + + // build the unified PubSub dispatcher + const ps = try self.allocator.create(root.PubSub); + ps.* = .{ .ptr = @ptrCast(@alignCast(self.Nats)), .vtable = &root.nats.vtable }; + self.pubSub = ps; +} + +pub fn natsPullWaitMs(self: *Self) u32 { + return @intCast(self.config.getAsInt("NATS_MAX_PULL_WAIT") catch 5000); +} + + fn loadMetricz(self: *Self) !void { // initialize metrics self.metricz = try Metricz.initialize(self.allocator, .{ .prefix = "", .exclude = null }); @@ -475,22 +537,22 @@ fn loadRedis(self: *Self) !void { const dbInt = try self.config.getAsInt("REDIS_DB"); const portInt = try self.config.getAsInt("REDIS_PORT"); - const addr = try std.net.Address.parseIp4(hostname, portInt); - const connection = try std.net.tcpConnectToAddress(addr); + const addr = try std.Io.net.IpAddress.parseIp4(hostname, portInt); + + const connection = try addr.connect(utils.io, .{ .mode = .stream }); + defer connection.close(utils.io); self.rdz = try rdzDatasource.create(self.allocator); + var reader = connection.reader(utils.io, &self.rdz.?.rbuf); + var writer = connection.writer(utils.io, &self.rdz.?.wbuf); - self.redis = rdzClient.init(connection, .{ - .auth = .{ - .user = null, - .pass = password, - }, - .reader_buffer = &self.rdz.?.rbuf, - .writer_buffer = &self.rdz.?.wbuf, + self.redis = rdzClient.init(utils.io, &reader.interface, &writer.interface, .{ + .user = null, + .pass = password, }) catch |err| { buffer = try std.fmt.bufPrint(buffer, "Failed to connect: {}", .{err}); self.log.err(buffer); - std.posix.exit(1); + std.process.exit(1); }; buffer = try std.fmt.bufPrint(buffer, "connecting to redis at '{s}:{d}' on database {d}", .{ hostname, portInt, dbInt }); @@ -591,10 +653,10 @@ fn loadSQL(self: *Self) !void { }, }; - self.SQL.?.sql = pgz.Pool.init(self.allocator, options) catch |err| { + self.SQL.?.sql = pgz.Pool.init(utils.io, self.allocator, options) catch |err| { buffer = try std.fmt.bufPrint(buffer, "Failed to connect: {}", .{err}); self.log.err(buffer); - std.posix.exit(1); + std.process.exit(1); }; self.SQL.?.options = &options; diff --git a/src/context.zig b/src/context.zig index 94dcec9..ea76a8d 100644 --- a/src/context.zig +++ b/src/context.zig @@ -4,6 +4,7 @@ const httpz = root.httpz; const zeroClient = root.client; const pubSub = root.MQTT; const mqMessage = root.mqMessage; +const natsMessage = root.natsMessage; const Error = root.Error; const Responder = root.responder; const constants = root.constants; @@ -20,12 +21,13 @@ pub const Context = struct { SQL: *root.SQL = undefined, SQLite: *root.SQLite = undefined, Cache: root.rediz.Client = undefined, - MQ: *root.MQTT = undefined, provider: *root.AuthProvider = undefined, + MQ: *root.MQTT = undefined, KF: *root.kafka = undefined, + NATS: *root.nats = undefined, - message: ?*mqMessage = null, - message2: ?*kafkaMessage = null, + pubsub: *root.PubSub = undefined, + message: ?root.pubsubInterface.Message = null, wsMessage: ?[]const u8 = null, wsClient: *root.httpz.websocket.Conn = undefined, @@ -65,6 +67,14 @@ pub const Context = struct { c.KF = k; } + if (container.Nats) |n| { + c.NATS = n; + } + + if (container.pubSub) |ps| { + c.pubsub = ps; + } + return c; } diff --git a/src/cronz/cronz.zig b/src/cronz/cronz.zig index e2e15a4..5cbc19d 100644 --- a/src/cronz/cronz.zig +++ b/src/cronz/cronz.zig @@ -1,6 +1,5 @@ const std = @import("std"); const root = @import("../zero.zig"); -const time = std.time; const arena: type = std.heap.ArenaAllocator; const Thread = std.Thread; const Atomic = std.atomic.Value; @@ -32,11 +31,10 @@ const _res: *httpz.Response = undefined; /// Set by cronz before calling a job's exec callback. Read-only for consumers. pub var current_job_name: ?[]const u8 = null; -ticker: time.Timer = undefined, thread: std.Thread = undefined, container: *root.container = undefined, jobs: std.array_list.Managed(job) = undefined, -mu: std.Thread.Mutex = undefined, +mu: std.Io.Mutex = undefined, running: Atomic(bool) = undefined, request: *httpz.Request = undefined, response: *httpz.Response = undefined, @@ -45,12 +43,11 @@ pub fn create(container: *root.container) !*Cronz { const c = try container.allocator.create(Cronz); errdefer container.allocator.destroy(c); - c.mu = .{}; + c.mu = .init; c.running = Atomic(bool).init(true); c.container = container; - c.ticker = try time.Timer.start(); c.jobs = std.array_list.Managed(job).init(container.allocator); - c.thread = try Thread.spawn(.{}, Cronz.runSchedules, .{ c, std.time.nanoTimestamp() }); + c.thread = try Thread.spawn(.{}, Cronz.runSchedules, .{ c, @as(i128, utils.nowReal().nanoseconds) }); return c; } @@ -79,8 +76,8 @@ fn destroryChildAllocator(self: *Self, ca: *arena) void { pub fn runSchedules(self: *Self, _: i128) void { while (self.running.load(.monotonic)) { - std.Thread.sleep(std.time.ns_per_s); - const now = dateTime.nowUTC(); + std.Io.sleep(utils.io, std.Io.Duration.fromSeconds(1), .awake) catch {}; + const now = dateTime.nowUTC(utils.io); for (self.jobs.items) |j| { if (j.compare(now)) { const ca = self.prepareChildAllocator() catch |err| { @@ -299,9 +296,9 @@ pub fn addCron(self: *Self, schedule: []const u8, name: []const u8, hook: *const j.name = name; j.exec = hook; - self.mu.lock(); + self.mu.lock(utils.io) catch {}; try self.jobs.append(j); - self.mu.unlock(); + self.mu.unlock(utils.io); const msg = utils.combine( self.container.allocator, @@ -413,6 +410,8 @@ fn mockContainer(allocator: std.mem.Allocator) root.container { .services = undefined, .pubsub = null, .Kakfa = null, + .Nats = null, + .pubSub = null, }; } diff --git a/src/cronz/job.zig b/src/cronz/job.zig index f628a38..96659af 100644 --- a/src/cronz/job.zig +++ b/src/cronz/job.zig @@ -37,10 +37,7 @@ pub const Job: type = struct { const ctx = context.?; - var timer = std.time.Timer.start() catch |err| { - ctx.any(err); - return; - }; + const start = utils.nowMonotonic(); root.cronz.current_job_name = self.name; self.exec(ctx) catch |err| { @@ -49,7 +46,7 @@ pub const Job: type = struct { }; root.cronz.current_job_name = null; - const elapsed: f32 = @floatFromInt(timer.lap() / 1000000); + const elapsed: f32 = utils.elapsedMs(start); const msg = utils.combine( ctx.allocator, @@ -144,7 +141,7 @@ test "job compare returns true when all fields match" { try j.month.put(3, true); try j.dayOfWeek.put(1, true); - const now = DateTime.nowUTC(); + const now = DateTime.nowUTC(utils.io); const result = j.compare(now); _ = result; } @@ -168,7 +165,7 @@ test "job compare returns false when field mismatches" { try j.month.put(1, true); try j.dayOfWeek.put(0, true); - const now = DateTime.nowUTC(); + const now = DateTime.nowUTC(utils.io); const second = now.second; if (!j.sec.contains(second)) { try std.testing.expect(j.compare(now) == false); @@ -187,7 +184,7 @@ test "job getTick returns current time components" { j.dayOfWeek.deinit(); } - const now = DateTime.nowUTC(); + const now = DateTime.nowUTC(utils.io); const t = j.getTick(now); try std.testing.expect(t.sec <= 59); try std.testing.expect(t.min <= 59); @@ -208,6 +205,6 @@ test "job compare returns false for empty job" { j.dayOfWeek.deinit(); } - const now = DateTime.nowUTC(); + const now = DateTime.nowUTC(utils.io); try std.testing.expect(j.compare(now) == false); } diff --git a/src/datasource/SQL.zig b/src/datasource/SQL.zig index b0772e1..2d7432a 100644 --- a/src/datasource/SQL.zig +++ b/src/datasource/SQL.zig @@ -1,5 +1,6 @@ const std = @import("std"); const root = @import("../zero.zig"); +const utils = root.utils; const SQL = @This(); const Self = @This(); @@ -8,6 +9,7 @@ const Results = root.pgz.Result; const QueryRow = root.pgz.QueryRow; const context = root.Context; const sqlStats = root.metricz.AppSQLStatsLabel; +const Mapper = root.pgz.Mapper; sql: *pgz.Pool, log: *root.logger, @@ -55,93 +57,166 @@ pub fn recordMetrics(self: *Self, duration: f32, query: []const u8, queryType: [ } pub fn queryRow(self: *Self, comptime query: []const u8, args: anytype) !?QueryRow { - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); - const rows = try self.sql.row(query, args); + const conn = try self.sql.acquire(); + defer self.sql.release(conn); - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + const rows = (try conn.row(query, args)) orelse unreachable; + + const duration: f32 = utils.elapsedMs(start); self.recordMetrics(duration, query, "select"); return rows; } pub fn queryRowContext(self: *Self, _: *context, comptime query: []const u8, args: anytype) !?QueryRow { - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); + + const conn = try self.sql.acquire(); + defer self.sql.release(conn); - const results = try self.sql.row(query, args); + const rows = conn.row(query, args) catch |err| { + if (err == error.PG) { + if (conn.err) |pge| { + self.log.err(pge.message); + } + } + return err; + }; - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + const duration: f32 = utils.elapsedMs(start); self.recordMetrics(duration, query, "select"); - return results; + return rows; } pub fn queryRows(self: *Self, comptime query: []const u8, args: anytype) !*Results { - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); + + const conn = try self.sql.acquire(); + defer self.sql.release(conn); - const results = try self.sql.query(query, args); + const rows = conn.query(query, args) catch |err| { + if (err == error.PG) { + if (conn.err) |pge| { + self.log.err(pge.message); + } + } + return err; + }; - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + const duration: f32 = utils.elapsedMs(start); self.recordMetrics(duration, query, "select"); - return results; + return rows; } pub fn queryRowsContext(self: *Self, _: *context, comptime query: []const u8, args: anytype) !*Results { - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); + + const conn = try self.sql.acquire(); + defer self.sql.release(conn); - const results = try self.sql.query(query, args); + const rows = conn.row(query, args) catch |err| { + if (err == error.PG) { + if (conn.err) |pge| { + self.log.err(pge.message); + } + } + return err; + }; - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + const duration: f32 = utils.elapsedMs(start); self.recordMetrics(duration, query, "select"); - return results; + return rows; } pub fn exec(self: *Self, comptime query: []const u8, args: anytype) !?i64 { - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); - const id = try self.sql.exec(query, args); + const conn = try self.sql.acquire(); + defer self.sql.release(conn); - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + const id = conn.exec(query, args) catch |err| { + if (err == error.PG) { + if (conn.err) |pge| { + self.log.err(pge.message); + } + } + return err; + }; + + const duration: f32 = utils.elapsedMs(start); self.recordMetrics(duration, query, "insert"); return id; } pub fn execWithContext(self: *Self, _: *context, comptime query: []const u8, args: anytype) !?i64 { - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); + + const conn = try self.sql.acquire(); + defer self.sql.release(conn); - const id = try self.sql.exec(query, args); + const id = conn.exec(query, args) catch |err| { + if (err == error.PG) { + if (conn.err) |pge| { + self.log.err(pge.message); + } + } + return err; + }; - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + const duration: f32 = utils.elapsedMs(start); self.recordMetrics(duration, query, "insert"); return id; } pub fn select(self: *Self, comptime _type: anytype, comptime query: []const u8, args: anytype) !?_type { - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); + + const conn = try self.sql.acquire(); + defer self.sql.release(conn); - const row = self.sql.row(query, args); - defer row.deinit() catch {}; + const row = try conn.queryOpts(query, args, .{ .column_names = true }); + defer row.deinit(); - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + var result: _type = undefined; + while (try row.next()) |_row| { + result = try _row.to(_type, .{}); + } + + const duration: f32 = utils.elapsedMs(start); self.recordMetrics(duration, query, "select"); - const result = try row.to(_type, .{}); return result; } -pub fn selectSlice(self: *Self, comptime _type: anytype, comptime query: []const u8, args: anytype) !*Results { - var timer = try std.time.Timer.start(); +pub fn selectSlice( + self: *Self, + comptime _type: anytype, + list: *std.array_list.Managed(_type), + comptime query: []const u8, + args: anytype, +) !i64 { + const start = utils.nowMonotonic(); + + const conn = try self.sql.acquire(); + defer self.sql.release(conn); + + const rows = try conn.queryOpts(query, args, .{ .column_names = true }); + defer rows.deinit(); - const row = self.sql.queryOpts(query, args); - defer row.deinit() catch {}; + var res = rows.mapper(_type, .{ .dupe = true }); + while (try res.next()) |T| { + try list.append(T); + } - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + const duration: f32 = utils.elapsedMs(start); self.recordMetrics(duration, query, "select"); - const results = try row.mapper(_type, .{}); - return results; + return 0; } diff --git a/src/datasource/SQLite.zig b/src/datasource/SQLite.zig index 6f6d23a..e6a016d 100644 --- a/src/datasource/SQLite.zig +++ b/src/datasource/SQLite.zig @@ -2,112 +2,76 @@ const std = @import("std"); const root = @import("../zero.zig"); const SQLite = @This(); -const Self = @This(); -const sqlitez = root.sqlitez; -db: sqlitez.Db, +allocator: std.mem.Allocator, log: *root.logger, metricz: *root.metricz, -allocator: std.mem.Allocator, +db: root.sqlitez.Db, pub fn init( allocator: std.mem.Allocator, - dbPath: []const u8, + db_path: []const u8, create: bool, write: bool, - threading_mode: sqlitez.ThreadingMode, + threading_mode: root.sqlitez.ThreadingMode, l: *root.logger, m: *root.metricz, ) !*SQLite { + const db_path_z = try allocator.dupeZ(u8, db_path); + const source = try allocator.create(SQLite); errdefer allocator.destroy(source); - const nullTermPath = try allocator.dupeZ(u8, dbPath); - - const options = sqlitez.InitOptions{ - .mode = .{ .File = nullTermPath }, - .open_flags = .{ .write = write, .create = create }, - .threading_mode = threading_mode, - }; - source.* = SQLite{ - .db = try sqlitez.Db.init(options), + .allocator = allocator, .log = l, .metricz = m, - .allocator = allocator, + .db = undefined, }; + source.db = try root.sqlitez.Db.init(.{ + .mode = .{ .File = db_path_z }, + .open_flags = .{ .write = write, .create = create }, + .threading_mode = threading_mode, + }); + return source; } pub fn queryRow(self: *SQLite, comptime Type: type, comptime query: []const u8, args: anytype) !?Type { - var timer = try std.time.Timer.start(); - - const result = try self.db.one(Type, query, .{}, args); - - const duration: f32 = @floatFromInt(timer.lap() / 1000000); - self.recordMetrics(duration, query, "select"); - - return result; + var stmt = try self.db.prepareDynamic(query); + defer stmt.deinit(); + return try stmt.one(Type, .{}, args); } pub fn queryRowContext(self: *SQLite, comptime Type: type, alloc: std.mem.Allocator, comptime query: []const u8, args: anytype) !?Type { - var timer = try std.time.Timer.start(); - - const result = try self.db.oneAlloc(Type, alloc, query, .{}, args); - - const duration: f32 = @floatFromInt(timer.lap() / 1000000); - self.recordMetrics(duration, query, "select"); - - return result; + var stmt = try self.db.prepareDynamic(query); + defer stmt.deinit(); + return try stmt.oneAlloc(Type, alloc, .{}, args); } pub fn queryRows(self: *SQLite, comptime Type: type, alloc: std.mem.Allocator, comptime query: []const u8, args: anytype) ![]Type { - var timer = try std.time.Timer.start(); - - var stmt = try self.db.prepare(query); + var stmt = try self.db.prepareDynamic(query); defer stmt.deinit(); - - const result = try stmt.all(Type, alloc, .{}, args); - - const duration: f32 = @floatFromInt(timer.lap() / 1000000); - self.recordMetrics(duration, query, "select"); - - return result; + return try stmt.all(Type, alloc, .{}, args); } pub fn queryRowsContext(self: *SQLite, comptime Type: type, alloc: std.mem.Allocator, comptime query: []const u8, args: anytype) ![]Type { - var timer = try std.time.Timer.start(); - - var stmt = try self.db.prepare(query); + var stmt = try self.db.prepareDynamic(query); defer stmt.deinit(); - - const result = try stmt.all(Type, alloc, .{}, args); - - const duration: f32 = @floatFromInt(timer.lap() / 1000000); - self.recordMetrics(duration, query, "select"); - - return result; + return try stmt.all(Type, alloc, .{}, args); } pub fn exec(self: *SQLite, comptime query: []const u8, args: anytype) !void { - var timer = try std.time.Timer.start(); - - const options = sqlitez.QueryOptions{}; - try self.db.exec(query, options, args); - - const duration: f32 = @floatFromInt(timer.lap() / 1000000); - self.recordMetrics(duration, query, "exec"); + var stmt = try self.db.prepareDynamic(query); + defer stmt.deinit(); + return try stmt.exec(.{}, args); } pub fn execContext(self: *SQLite, comptime query: []const u8, args: anytype) !void { - var timer = try std.time.Timer.start(); - - const options = sqlitez.QueryOptions{}; - try self.db.exec(query, options, args); - - const duration: f32 = @floatFromInt(timer.lap() / 1000000); - self.recordMetrics(duration, query, "exec"); + var stmt = try self.db.prepareDynamic(query); + defer stmt.deinit(); + return try stmt.exec(.{}, args); } pub fn rowsAffected(self: *SQLite) usize { @@ -117,17 +81,3 @@ pub fn rowsAffected(self: *SQLite) usize { pub fn lastInsertRowID(self: *SQLite) i64 { return self.db.getLastInsertRowID(); } - -fn recordMetrics(self: *SQLite, duration: f32, query: []const u8, queryType: []const u8) void { - _ = query; - _ = queryType; - self.metricz.sqlResponse( - .{ - .hostname = "", - .database = "", - .query = "", - .operation = "", - }, - duration, - ) catch unreachable; -} diff --git a/src/handler.zig b/src/handler.zig index e702f2e..14e9a97 100644 --- a/src/handler.zig +++ b/src/handler.zig @@ -19,7 +19,6 @@ pub const Handler = struct { _res: *httpz.Response = undefined, container: *root.container = undefined, ctx: *Context = undefined, - timer: std.time.Timer = undefined, wsClient: wsHandler = undefined, pub const WebsocketHandler = wsHandler; @@ -29,12 +28,16 @@ pub const Handler = struct { } pub fn ws(self: *Handler, action: Responder.Do(*Context), req: *httpz.Request, res: *httpz.Response) !void { - var ctx = try Context.init(req.arena, self.container, req, res); - defer req.arena.destroy(&ctx); - + // The websocket connection outlives this request, so the Context must be + // heap-allocated with a persistent allocator. Using req.arena (and a + // stack variable) left a dangling pointer that crashed on the first + // message (garbage allocator vtable during logging). + const ctx = try self.container.allocator.create(Context); + ctx.* = try Context.init(self.container.allocator, self.container, req, res); ctx.action = action; - if (try httpz.upgradeWebsocket(wsHandler, req, res, &ctx) == false) { + if (try httpz.upgradeWebsocket(wsHandler, req, res, ctx) == false) { + ctx.deinit(); res.setStatus(.internal_server_error); res.body = "invalid websocket"; return; @@ -51,12 +54,12 @@ pub const Handler = struct { var ctx = try Context.init(req.arena, self.container, req, res); defer req.arena.destroy(&ctx); - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); try action(&ctx); // does not include middleware executions - const duration: f32 = @floatFromInt(timer.lap() / 1000000); + const duration: f32 = utils.elapsedMs(start); try self.metric(duration, @tagName(req.method), res.status, req.url.path); diff --git a/src/httpServer.zig b/src/httpServer.zig index c41bc23..1275c24 100644 --- a/src/httpServer.zig +++ b/src/httpServer.zig @@ -54,9 +54,10 @@ pub fn create(allocator: std.mem.Allocator, container: *root.container) !*server }; hzs.http = try httpz.Server(*root.handler.Handler).init( + utils.io, hzs.container.allocator, .{ - .port = hzs.port, + .address = httpz.Config.Address.all(hzs.port), }, &hzs.handler, ); @@ -165,7 +166,7 @@ fn loadAuthProviderConfig(self: *Self) anyerror!?*authProvider { const refreshAt = try std.fmt.parseInt(i16, refreshInterval, 10); provider = try authProvider.create(self.container, .OAuth); - provider.?.mutex = .{}; + provider.?.mutex = .init; provider.?.pathUrl = jwksUrl; provider.?.refreshInterval = refreshAt; provider.?.pubKeys = std.StringHashMap(PubKey).init(self.container.allocator); diff --git a/src/logger.zig b/src/logger.zig index 2a735e5..1158289 100644 --- a/src/logger.zig +++ b/src/logger.zig @@ -4,10 +4,7 @@ const Self = @This(); const root = @import("zero.zig"); const utils = root.utils; -var stdout: *std.Io.Writer = undefined; -var stdout_buffer: [512]u8 = undefined; -var stdout_writer: std.fs.File.Writer = undefined; -var mutex: std.Thread.Mutex = .{}; +var mutex: std.Io.Mutex = .init; allocator: std.mem.Allocator, logLevel: u8 = undefined, @@ -18,19 +15,18 @@ pub fn custom( comptime format: []const u8, args: anytype, ) void { - mutex.lock(); - defer mutex.unlock(); - nosuspend stdout.print(format, args) catch return; - nosuspend stdout.flush() catch return; + mutex.lock(utils.io) catch {}; + defer mutex.unlock(utils.io); + var buf: [2048]u8 = undefined; + const msg = std.fmt.bufPrint(&buf, format, args) catch "log format error"; + const out = std.Io.File.stdout(); + out.writeStreamingAll(utils.io, msg) catch return; } pub fn create(allocator: std.mem.Allocator) !*logger { const l: *logger = try allocator.create(logger); errdefer allocator.destroy(l); - stdout_writer = std.fs.File.stdout().writer(&stdout_buffer); - stdout = &stdout_writer.interface; - l.allocator = allocator; l.logLevel = 1; diff --git a/src/metricz.zig b/src/metricz.zig index 3b91389..06ce56f 100644 --- a/src/metricz.zig +++ b/src/metricz.zig @@ -1,11 +1,11 @@ const std = @import("std"); -const metrics = @import("metriks"); const root = @import("zero.zig"); const builtin = @import("builtin"); const Allocator = std.mem.Allocator; const Self = @This(); const metricz = @This(); const pgz = root.pgz; +const metrics = root.httpz.metriks; const Context = root.Context; const Process = root.process; const utils = root.utils; @@ -200,11 +200,12 @@ pub fn SubscriberSuccess(self: *Self, labels: PubSubSubscriberSuccessLabel) !voi } pub fn initialize(allocator: Allocator, comptime _: metrics.RegistryOpts) !*metricz { + metrics.setIo(utils.io); const m = try allocator.create(metricz); errdefer allocator.destroy(m); m.Info = try metrics.CounterVec(u32, AppInfoLabel).Impl - .init(allocator, "app_info", .{ .help = "Info for app_name, app_version and framework_version." }); + .init(allocator, utils.io, "app_info", .{ .help = "Info for app_name, app_version and framework_version." }); m.Threads = try metrics.GaugeVec(u64, AppThreadsourceLabel).Impl .init(allocator, "app_threads", .{ .help = "Info of overall app threads count." }); @@ -216,28 +217,28 @@ pub fn initialize(allocator: Allocator, comptime _: metrics.RegistryOpts) !*metr .init(allocator, "app_memory_total", .{ .help = "Info of overall app memory total usage." }); m.ResponseBucket = try metrics.HistogramVec(f64, AppHttpResponseLatencyLabel, &.{ 0.001, 0.003, 0.005, 0.01, 0.02, 0.03, 0.05, 0.1, 0.2, 0.3, 0.5, 0.75, 1, 2, 3, 5, 10, 30 }).Impl - .init(allocator, "app_http_response", .{ .help = "Response time of HTTP requests in seconds." }); + .init(allocator, utils.io, "app_http_response", .{ .help = "Response time of HTTP requests in seconds." }); m.ResponseBucketHits = try metrics.CounterVec(u64, AppHttpResponseHitLabel).Impl - .init(allocator, "app_http_response_hits", .{ .help = "Response counts of HTTP requests." }); + .init(allocator, utils.io, "app_http_response_hits", .{ .help = "Response counts of HTTP requests." }); m.ServiceResponseBucket = try metrics.HistogramVec(f64, ServiceResponseLabel, &.{ 0.001, 0.003, 0.005, 0.01, 0.02, 0.03, 0.05, 0.1, 0.2, 0.3, 0.5, 0.75, 1, 2, 3, 5, 10, 30 }).Impl - .init(allocator, "app_http_service_response", .{ .help = "Response time of external service requests in seconds." }); + .init(allocator, utils.io, "app_http_service_response", .{ .help = "Response time of external service requests in seconds." }); m.SQLBucket = try metrics.HistogramVec(f64, AppSQLStatsLabel, &.{ 0.001, 0.003, 0.005, 0.01, 0.02, 0.03, 0.05, 0.1, 0.2, 0.3, 0.5, 0.75, 1, 2, 3, 5, 10, 30 }).Impl - .init(allocator, "app_sql_response", .{ .help = "Response time of sql query execution in seconds." }); + .init(allocator, utils.io, "app_sql_response", .{ .help = "Response time of sql query execution in seconds." }); m.PubSubPublisherTotal = try metrics.CounterVec(u64, PubSubPublisherTotalLabel).Impl - .init(allocator, "app_pubsub_publish_total_count", .{ .help = "Total pubsub publisher counter per topic" }); + .init(allocator, utils.io, "app_pubsub_publish_total_count", .{ .help = "Total pubsub publisher counter per topic" }); m.PubSubPublisherSuccess = try metrics.CounterVec(u64, PubSubPublisherSuccessLabel).Impl - .init(allocator, "app_pubsub_publish_success_count", .{ .help = "Successful pubsub publisher counter per topic" }); + .init(allocator, utils.io, "app_pubsub_publish_success_count", .{ .help = "Successful pubsub publisher counter per topic" }); m.PubSubSubscriberTotal = try metrics.CounterVec(u64, PubSubSubscriberTotalLabel).Impl - .init(allocator, "app_pubsub_subscriber_total_count", .{ .help = "Total pubsub subscriber counter per topic per consumer group" }); + .init(allocator, utils.io, "app_pubsub_subscriber_total_count", .{ .help = "Total pubsub subscriber counter per topic per consumer group" }); m.PubSubSubscriberSuccess = try metrics.CounterVec(u64, PubSubSubscriberSuccessLabel).Impl - .init(allocator, "app_pubsub_subscriber_success_count", .{ .help = "Successful pubsub subscriber counter per topic per consumer group" }); + .init(allocator, utils.io, "app_pubsub_subscriber_success_count", .{ .help = "Successful pubsub subscriber counter per topic per consumer group" }); return m; } diff --git a/src/migration/SQL.zig b/src/migration/SQL.zig index b326ee3..da34fbb 100644 --- a/src/migration/SQL.zig +++ b/src/migration/SQL.zig @@ -8,6 +8,13 @@ const migrate = root.migrate; const utils = root.utils; const dateTime = root.zdt.Datetime; +const zeroTable = struct { + epoch: i64, + execution: []const u8, + start_time: []const u8, + duration: i64, +}; + const migrationTablePostgres = \\ CREATE TABLE IF NOT EXISTS zero_migrations ( \\ epoch BIGINT NOT NULL, @@ -20,7 +27,7 @@ const migrationTablePostgres = const migrationTableSQLite = \\ CREATE TABLE IF NOT EXISTS zero_migrations ( - \\ epoch INTEGER NOT NULL, + \\ epoch BIGINT NOT NULL, \\ execution TEXT NOT NULL, \\ start_time TEXT NOT NULL, \\ duration INTEGER, @@ -29,11 +36,11 @@ const migrationTableSQLite = ; const lastMigrationRecord = - \\"SELECT COALESCE(MAX(epoch), 0) FROM zero_migrations;" + \\SELECT epoch, execution, start_time, duration FROM zero_migrations order by epoch desc limit 1 ; const insertMigrationRecordPostgres = - \\"INSERT INTO zero_migrations (epoch, execution, start_time, duration) VALUES ($1, $2, $3, $4);" + \\INSERT INTO zero_migrations (epoch, execution, start_time, duration) VALUES ($1, $2, $3, $4) ; pub fn checkAndCreateMigrationTable(ctx: *Context) !void { @@ -56,13 +63,19 @@ pub fn checkAndCreateMigrationTable(ctx: *Context) !void { pub fn lastMigration(ctx: *Context) !i64 { const dialect = ctx.container.config.get("DB_DIALECT"); + if (std.mem.eql(u8, "postgres", dialect)) { - const result = try ctx.SQL.queryRow(lastMigrationRecord, .{}); + const result: ?zeroTable = try ctx.SQL.select( + zeroTable, + lastMigrationRecord, + .{}, + ); + return result.?.epoch; + } else if (std.mem.eql(u8, "sqlite", dialect)) { + const result = try ctx.SQLite.queryRow(i64, lastMigrationRecord, .{}); if (result) |r| { - return r.get(i64, 0); + return r; } - } else if (std.mem.eql(u8, "sqlite", dialect)) { - return ctx.SQLite.lastInsertRowID(); } return 0; diff --git a/src/migration/migration.zig b/src/migration/migration.zig index 43c944b..6ea096a 100644 --- a/src/migration/migration.zig +++ b/src/migration/migration.zig @@ -9,6 +9,7 @@ const SQL = root.SQL; const util = root.utils; const migrate = root.migrate; const zdt = root.zdt; +const utils = root.utils; const sqlMigrator = @import("./SQL.zig"); @@ -49,7 +50,11 @@ pub fn run(self: *Self) anyerror!void { const lastMigration = try sqlMigrator.lastMigration(ctx); for (self.keys.items) |key| { - const keyAsString = try util.toStringFromInt(ctx.allocator, "{d}", key); + const keyAsString = try util.toStringFromInt( + ctx.allocator, + "{d}", + key, + ); const value = self.map.get(keyAsString); @@ -59,7 +64,7 @@ pub fn run(self: *Self) anyerror!void { continue; } - var timer = try std.time.Timer.start(); + const start = util.nowReal(); m.run(ctx) catch |err| switch (err) { else => { @@ -68,7 +73,7 @@ pub fn run(self: *Self) anyerror!void { }, }; - const duration: u64 = timer.lap() / 1000000; + const duration: u64 = @as(u64, @intCast(@divTrunc(start.nanoseconds, 1_000_000))); _ = try sqlMigrator.insertMigration(ctx, m, duration); diff --git a/src/mw/authProvider.zig b/src/mw/authProvider.zig index 117baae..a354a5d 100644 --- a/src/mw/authProvider.zig +++ b/src/mw/authProvider.zig @@ -73,7 +73,7 @@ container: *root.container, keys: std.StringHashMap([]const u8) = undefined, pubKeys: std.StringHashMap(publiKey) = undefined, refreshThread: std.Thread = undefined, -mutex: std.Thread.Mutex = undefined, +mutex: std.Io.Mutex = undefined, refreshInterval: i16 = 60, // seconds pathUrl: []const u8 = undefined, @@ -230,14 +230,14 @@ pub fn validateOAuthToken(self: *Self, allocator: std.mem.Allocator, authHeader: }; defer claims.deinit(); - var validator = jwt.Validator.init(&jwtTokenizer) catch |err| switch (err) { + var validator = jwt.Validator.init(allocator, &jwtTokenizer) catch |err| switch (err) { else => { return AuthError.TokenInvalidClaims; }, }; defer validator.deinit(); - const now = std.time.timestamp(); + const now = @as(i64, @intCast(@divTrunc(utils.nowReal().nanoseconds, 1_000_000_000))); // validator.hasBeenIssuedBy(publicKey.) // iss // validator.isRelatedTo("sub") // sub // validator.isIdentifiedBy("jti rrr") // jti @@ -331,9 +331,9 @@ pub fn refreshKeys(ctx: *Context) !void { defer parsed.deinit(); for (parsed.value.keys) |key| { - ctx.container.authProvider.mutex.lock(); + ctx.container.authProvider.mutex.lock(utils.io) catch {}; try ctx.container.authProvider.pubKeys.put(key.kid, key); - ctx.container.authProvider.mutex.unlock(); + ctx.container.authProvider.mutex.unlock(utils.io); } ctx.info("oatuh keys refreshed"); diff --git a/src/mw/tracz.zig b/src/mw/tracz.zig index dd9cc7e..dc4ff94 100644 --- a/src/mw/tracz.zig +++ b/src/mw/tracz.zig @@ -4,6 +4,7 @@ const root = @import("../zero.zig"); const tracz = @This(); const zul = root.zul; +const utils = root.utils; allocator: std.mem.Allocator, @@ -14,7 +15,7 @@ pub fn init(c: Config) !tracz { } pub fn execute(_: *const tracz, req: *httpz.Request, res: *httpz.Response, executor: anytype) !void { - const uuid = zul.UUID.v4(); + const uuid = zul.UUID.v4(utils.io); var buffer: []u8 = undefined; buffer = try req.arena.alloc(u8, 36); diff --git a/src/pubsub/interface.zig b/src/pubsub/interface.zig new file mode 100644 index 0000000..85cbd00 --- /dev/null +++ b/src/pubsub/interface.zig @@ -0,0 +1,38 @@ +const std = @import("std"); +const root = @import("../zero.zig"); + +/// Unified inbound message. A tagged union over the per-backend message +/// types so subscribe hooks can read the payload regardless of backend. +pub const Message = union(enum) { + mqtt: *root.mqMessage, + kafka: *root.kafkaMessage, + nats: *root.natsMessage, +}; + +/// Unified pub/sub interface (type-erased VTable). +/// +/// Wraps any configured backend (MQTT, Kafka, NATS) behind a stable +/// function-pointer table. Handler code uses `ctx.pubsub.Publish(...)` / +/// `ctx.pubsub.subscribe(...)` without knowing or reading the backend. +pub const Interface = struct { + ptr: *anyopaque, + vtable: *const VTable, + + pub const VTable = struct { + publish: *const fn (*anyopaque, []const u8, []const u8) anyerror!void, + subscribe: *const fn (*anyopaque, []const u8, *const fn (*root.Context) anyerror!void) anyerror!void, + }; + + pub fn Publish(self: Interface, subject: []const u8, payload: []const u8) !void { + return self.vtable.publish(self.ptr, subject, payload); + } + + pub fn subscribe(self: Interface, subject: []const u8, hook: *const fn (*root.Context) anyerror!void) !void { + return self.vtable.subscribe(self.ptr, subject, hook); + } + + /// Alias for `subscribe`, matching the app-level `addPubSubSubscription` naming. + pub fn addSubscriber(self: Interface, subject: []const u8, hook: *const fn (*root.Context) anyerror!void) !void { + return self.subscribe(subject, hook); + } +}; diff --git a/src/pubsub/kafka/kafka.zig b/src/pubsub/kafka/kafka.zig index 76ad670..2007ba2 100644 --- a/src/pubsub/kafka/kafka.zig +++ b/src/pubsub/kafka/kafka.zig @@ -28,7 +28,7 @@ const _res: *httpz.Response = undefined; thread: std.Thread = undefined, container: *root.container = undefined, rootContext: *root.Context = undefined, -mu: std.Thread.Mutex = undefined, +mu: std.Io.Mutex = undefined, signal: Atomic(bool) = undefined, config: ?*kafkaConfig, topic: ?*kafkaTopic, @@ -47,7 +47,7 @@ pub fn create( const c = try container.allocator.create(Kafka); errdefer container.allocator.destroy(c); - c.mu = .{}; + c.mu = .init; c.signal = Atomic(bool).init(true); c.container = container; c.subscriber = std.array_list.Managed(kafkaSubscriber).init(container.allocator); @@ -174,6 +174,35 @@ pub fn publish(self: *Self, ctx: *Context, topic: *kafkaTopic, key: []const u8, self.container.metricz.publisherTotal(.{ .topic = self.getTopicName(topic) }) catch unreachable; } +/// Convenience for the unified `PubSub` interface: publish to a subject +/// using a throwaway context (Kafka's `publish` requires a `*Context`). +pub fn publishOnSubject(self: *Self, subject: []const u8, payload: []const u8) !void { + const ca = self.prepareChildAllocator() catch |err| { + self.container.log.any(err); + return; + }; + defer self.destroryChildAllocator(ca); + + var ctx = Context.init( + ca.allocator(), + self.container, + _req, + _res, + ) catch |err| { + self.container.log.any(err); + return; + }; + const context = &ctx; + + const topic = self.getTopicHandler(context, subject) catch |err| { + self.container.log.any(err); + return; + }; + defer rdkafka.rd_kafka_topic_destroy(topic); + + try self.publish(context, topic, "", payload); +} + pub inline fn wait(self: Self, comptime timeout_ms: u16) void { while (rdkafka.rd_kafka_outq_len(self._producer) > 0) { _ = rdkafka.rd_kafka_poll(self._producer, timeout_ms); @@ -225,7 +254,7 @@ pub fn readPayload(self: *Self, subscriber: kafkaSubscriber) !void { const context = &ctx; // transform packet to client.response using std.json.parse. - context.message2 = &msg; + context.message = .{ .kafka = &msg }; try subscriber.exec(context); @@ -238,7 +267,7 @@ pub fn readPayload(self: *Self, subscriber: kafkaSubscriber) !void { fn subscriptions(self: *Self) !void { for (self.subscriber.items) |s| { - std.Thread.sleep(std.time.ns_per_ms * 100); + std.Io.sleep(utils.io, std.Io.Duration.fromMilliseconds(100), .awake) catch {}; const err_code: c_int = rdkafka.rd_kafka_subscribe(self.client, s.topics); if (err_code != rdkafka.RD_KAFKA_RESP_ERR_NO_ERROR) { const msg = try utils.combine( @@ -320,9 +349,9 @@ pub fn addSubscriber(self: *Self, topic: []const u8, hook: *const fn (*root.Cont .exec = hook, }; - self.mu.lock(); + self.mu.lock(utils.io) catch {}; try self.subscriber.append(s); - self.mu.unlock(); + self.mu.unlock(utils.io); const msg = utils.combine( self.container.allocator, @@ -340,3 +369,19 @@ inline fn getTopicName(_: *Self, topic: *kafkaTopic) []const u8 { const name: []const u8 = std.mem.span(rdkafka.rd_kafka_topic_name(topic)); return name; } + +/// Type-erased VTable conforming to `pubsubInterface.Interface.VTable`. +pub const vtable = root.pubsubInterface.Interface.VTable{ + .publish = struct { + fn call(ptr: *anyopaque, subject: []const u8, payload: []const u8) anyerror!void { + const self: *Kafka = @ptrCast(@alignCast(ptr)); + try self.publishOnSubject(subject, payload); + } + }.call, + .subscribe = struct { + fn call(ptr: *anyopaque, subject: []const u8, hook: *const fn (*root.Context) anyerror!void) anyerror!void { + const self: *Kafka = @ptrCast(@alignCast(ptr)); + try self.addSubscriber(subject, hook); + } + }.call, +}; diff --git a/src/pubsub/mqtt/MQTT.zig b/src/pubsub/mqtt/MQTT.zig index c144b73..eeb83d5 100644 --- a/src/pubsub/mqtt/MQTT.zig +++ b/src/pubsub/mqtt/MQTT.zig @@ -24,9 +24,9 @@ thread: std.Thread = undefined, container: *root.container = undefined, rootContext: *root.Context = undefined, subscriber: std.array_list.Managed(mqSubscriber) = undefined, -mu: std.Thread.Mutex = undefined, +mu: std.Io.Mutex = undefined, signal: Atomic(bool) = undefined, -mqtt: root.mqttz.posix.Client = undefined, +mqtt: root.mqttz.posix.Client311 = undefined, mqttClient: ?[]const u8 = undefined, isPubSubSet: bool = false, @@ -34,12 +34,12 @@ pub fn create(container: *root.container, config: *const mqConfig) !*MQTT { const c = try container.allocator.create(MQTT); errdefer container.allocator.destroy(c); - c.mu = .{}; + c.mu = .init; c.signal = Atomic(bool).init(true); c.container = container; c.subscriber = std.array_list.Managed(mqSubscriber).init(container.allocator); - const m = try root.mqttz.posix.Client.init(.{ + const m = try root.mqttz.posix.Client311.init(utils.io, .{ .port = config.port, .ip = config.ip, .host = config.hostname, @@ -66,8 +66,10 @@ pub fn create(container: *root.container, config: *const mqConfig) !*MQTT { c.mqttClient = cack.assigned_client_identifier; - msg = try utils.combine(container.allocator, "MQTT client id {s}", .{cack.assigned_client_identifier.?}); - container.log.info(msg); + if (cack.assigned_client_identifier) |id| { + msg = try utils.combine(container.allocator, "MQTT client id {s}", .{id}); + container.log.info(msg); + } }, else => { const msg = try utils.combine(container.allocator, "could not connect to MQTT at '{s}:{d}'", .{ config.hostname, config.port }); @@ -114,7 +116,7 @@ fn destroryChildAllocator(self: *Self, ca: *arena) void { pub fn readPackets(self: *Self, subscriber: mqSubscriber) !void { while (self.signal.load(.monotonic)) { - std.Thread.sleep(std.time.ns_per_s); + std.Io.sleep(utils.io, std.Io.Duration.fromSeconds(1), .awake) catch {}; const packet = try self.mqtt.readPacket(.{ .timeout = 1000 }) orelse { continue; }; @@ -143,7 +145,7 @@ pub fn readPackets(self: *Self, subscriber: mqSubscriber) !void { }; // transform packet to client.response using std.json.parse. - context.message = &message; + context.message = .{ .mqtt = &message }; try subscriber.exec(context); }, @@ -184,7 +186,7 @@ fn subscriptions(self: *Self) !void { }, }; - std.Thread.sleep(std.time.ns_per_ms * 100); + std.Io.sleep(utils.io, std.Io.Duration.fromMilliseconds(100), .awake) catch {}; const thread = Thread.spawn(.{}, Self.readPackets, .{ self, client }) catch |err| { self.container.log.any(err); return; @@ -207,9 +209,9 @@ pub fn addSubscriber(self: *Self, topic: []const u8, hook: *const fn (*root.Cont .exec = hook, }; - self.mu.lock(); + self.mu.lock(utils.io) catch {}; try self.subscriber.append(s); - self.mu.unlock(); + self.mu.unlock(utils.io); const msg = utils.combine( self.container.allocator, @@ -222,3 +224,19 @@ pub fn addSubscriber(self: *Self, topic: []const u8, hook: *const fn (*root.Cont self.container.log.info(msg); } + +/// Type-erased VTable conforming to `pubsubInterface.Interface.VTable`. +pub const vtable = root.pubsubInterface.Interface.VTable{ + .publish = struct { + fn call(ptr: *anyopaque, subject: []const u8, payload: []const u8) anyerror!void { + const self: *MQTT = @ptrCast(@alignCast(ptr)); + _ = try self.Publish(subject, payload); + } + }.call, + .subscribe = struct { + fn call(ptr: *anyopaque, subject: []const u8, hook: *const fn (*root.Context) anyerror!void) anyerror!void { + const self: *MQTT = @ptrCast(@alignCast(ptr)); + try self.addSubscriber(subject, hook); + } + }.call, +}; diff --git a/src/pubsub/nats/NATS.zig b/src/pubsub/nats/NATS.zig new file mode 100644 index 0000000..3f0ab44 --- /dev/null +++ b/src/pubsub/nats/NATS.zig @@ -0,0 +1,256 @@ +const std = @import("std"); +const root = @import("../../zero.zig"); +pub const NATS = @This(); +const Self = @This(); + +const nats = root.natslib; +const natsConfig = root.natsConfig; +const natsMessage = root.natsMessage; +const natsSubscriber = root.natsSubscriber; + +const time = std.time; +const Thread = std.Thread; +const Atomic = std.atomic.Value; +const arena: type = std.heap.ArenaAllocator; + +const utils = root.utils; +const Context = root.Context; +const constants = root.constants; +const httpz = root.httpz; + +const _req: *httpz.Request = undefined; +const _res: *httpz.Response = undefined; + +allocator: std.mem.Allocator = undefined, +thread: std.Thread = undefined, +container: *root.container = undefined, +client: *nats.Client = undefined, +js: ?nats.jetstream.JetStream = null, +stream: ?nats.jetstream.PullSubscription = null, +subscriber: std.array_list.Managed(natsSubscriber) = undefined, +mu: std.Io.Mutex = undefined, +signal: Atomic(bool) = undefined, +isPubSubSet: bool = false, + +pub fn create(container: *root.container, config: *const natsConfig) !*NATS { + const c = try container.allocator.create(NATS); + errdefer container.allocator.destroy(c); + + c.mu = .init; + c.signal = Atomic(bool).init(true); + c.container = container; + c.subscriber = std.array_list.Managed(natsSubscriber).init(container.allocator); + c.allocator = container.allocator; + + var opts = nats.Options{}; + if (config.creds_file.len > 0) { + opts.creds_file = config.creds_file; + } + + const client = try nats.Client.connect(container.allocator, utils.io, config.url, opts); + c.client = client; + + if (config.hasStream()) { + c.js = try nats.jetstream.JetStream.init(client, .{}); + + var subjects_buf: [8][]const u8 = undefined; + var it = std.mem.splitScalar(u8, config.subjects, ','); + var count: usize = 0; + while (it.next()) |s| { + const trimmed = std.mem.trim(u8, s, " \t"); + if (trimmed.len == 0) continue; + if (count >= subjects_buf.len) break; + subjects_buf[count] = trimmed; + count += 1; + } + const subjects = subjects_buf[0..count]; + + _ = c.js.?.createStream(.{ .name = config.stream, .subjects = subjects }) catch |err| { + // a stream with the same name may already exist; treat that as ok. + if (err != error.StreamExists) { + container.log.any(err); + } + }; + + _ = c.js.?.createOrUpdateConsumer(config.stream, .{ + .durable_name = config.consumer, + .ack_policy = .all, + }) catch |err| { + container.log.any(err); + return err; + }; + + var ps = nats.jetstream.PullSubscription{ .js = &c.js.?, .stream = config.stream }; + try ps.setConsumer(config.consumer); + c.stream = ps; + } + + c.isPubSubSet = true; + + const msg = utils.combine( + container.allocator, + "connected to NATS at '{s}'", + .{config.url}, + ) catch |err| { + container.log.any(err); + return err; + }; + + container.log.info(msg); + + return c; +} + +pub fn destroy(self: *Self) void { + self.signal.store(false, .release); + if (self.stream) |*ps| { + ps.deinit(); + } + self.client.deinit(); + if (self.subscriber.count() > 0) { + self.thread.join(); + } +} + +pub fn Publish(self: *Self, subject: []const u8, payload: []const u8) !void { + return try self.client.publish(subject, payload); +} + +fn prepareChildAllocator(self: *Self) !*arena { + const ca: *arena = try self.container.allocator.create(arena); + errdefer self.container.allocator.destroy(ca); + + ca.* = arena.init(self.container.allocator); + errdefer ca.deinit(); + + return ca; +} + +fn destroryChildAllocator(self: *Self, ca: *arena) void { + const caPtr: *arena = @ptrCast(@alignCast(ca.allocator().ptr)); + caPtr.deinit(); + + self.container.allocator.destroy(caPtr); +} + +fn dispatch(self: *Self, subject: []const u8, payload: []const u8, hook: *const fn (*root.Context) anyerror!void) void { + const ca = self.prepareChildAllocator() catch |err| { + self.container.log.any(err); + return; + }; + defer self.destroryChildAllocator(ca); + + var ctx = Context.init( + ca.allocator(), + self.container, + _req, + _res, + ) catch |err| { + self.container.log.any(err); + return; + }; + const context = &ctx; + + var message = natsMessage{ + .context = context, + .subject = subject, + .payload = payload, + }; + context.message = .{ .nats = &message }; + + hook(context) catch |err| { + self.container.log.any(err); + }; +} + +fn readJetStream(self: *Self, sub: natsSubscriber) !void { + while (self.signal.load(.monotonic)) { + var result = self.stream.?.fetch(.{ + .max_messages = 1, + .timeout_ms = self.container.natsPullWaitMs(), + }) catch |err| { + if (err == error.NoHeartbeat) { + continue; + } + self.container.log.any(err); + return; + }; + defer result.deinit(); + + if (result.count() == 0) continue; + + var msg = result.messages[0]; + const subject = msg.subject(); + const payload = msg.data(); + self.dispatch(subject, payload, sub.exec); + msg.ack() catch {}; + // NOTE: do not call msg.deinit() here — result.deinit() (deferred above) + // owns and frees all JsMsg buffers. Calling it again double-frees. + } +} + +fn readCore(self: *Self, sub: natsSubscriber) !void { + const s = try self.client.subscribeSync(sub.topic); + while (self.signal.load(.monotonic)) { + std.Io.sleep(utils.io, std.Io.Duration.fromMilliseconds(100), .awake) catch {}; + const msg = s.tryNextMsg() orelse continue; + self.dispatch(msg.subject, msg.data, sub.exec); + msg.deinit(); + } +} + +fn subscriptions(self: *Self) !void { + for (self.subscriber.items) |client| { + if (self.stream != null) { + try self.readJetStream(client); + } else { + try self.readCore(client); + } + } +} + +pub fn startSubscription(self: *Self) !void { + self.thread = Thread.spawn(.{}, Self.subscriptions, .{self}) catch |err| { + self.container.log.any(err); + return; + }; +} + +pub fn addSubscriber(self: *Self, topic: []const u8, hook: *const fn (*root.Context) anyerror!void) !void { + const s = natsSubscriber{ + .topic = topic, + .name = topic, + .exec = hook, + }; + + self.mu.lock(utils.io) catch {}; + try self.subscriber.append(s); + self.mu.unlock(utils.io); + + const msg = utils.combine( + self.container.allocator, + "topic:{s} nats subscriber added", + .{s.topic}, + ) catch |err| { + self.container.log.any(err); + return; + }; + + self.container.log.info(msg); +} + +/// Type-erased VTable conforming to `pubsubInterface.Interface.VTable`. +pub const vtable = root.pubsubInterface.Interface.VTable{ + .publish = struct { + fn call(ptr: *anyopaque, subject: []const u8, payload: []const u8) anyerror!void { + const self: *NATS = @ptrCast(@alignCast(ptr)); + try self.Publish(subject, payload); + } + }.call, + .subscribe = struct { + fn call(ptr: *anyopaque, subject: []const u8, hook: *const fn (*root.Context) anyerror!void) anyerror!void { + const self: *NATS = @ptrCast(@alignCast(ptr)); + try self.addSubscriber(subject, hook); + } + }.call, +}; diff --git a/src/pubsub/nats/config.zig b/src/pubsub/nats/config.zig new file mode 100644 index 0000000..7271e9e --- /dev/null +++ b/src/pubsub/nats/config.zig @@ -0,0 +1,15 @@ +const std = @import("std"); + +pub const natsConfig = struct { + url: []const u8 = undefined, + stream: []const u8 = undefined, + subjects: []const u8 = undefined, + max_wait_ms: u32 = undefined, + max_pull_wait_ms: u32 = undefined, + consumer: []const u8 = undefined, + creds_file: []const u8 = undefined, + + pub fn hasStream(self: *const natsConfig) bool { + return self.stream.len > 0; + } +}; diff --git a/src/pubsub/nats/message.zig b/src/pubsub/nats/message.zig new file mode 100644 index 0000000..def9e0b --- /dev/null +++ b/src/pubsub/nats/message.zig @@ -0,0 +1,8 @@ +const std = @import("std"); +const root = @import("../../zero.zig"); + +pub const natsMessage = struct { + context: *root.Context, + subject: []const u8, + payload: []const u8, +}; diff --git a/src/pubsub/nats/subscriber.zig b/src/pubsub/nats/subscriber.zig new file mode 100644 index 0000000..9e618f0 --- /dev/null +++ b/src/pubsub/nats/subscriber.zig @@ -0,0 +1,8 @@ +const std = @import("std"); +const root = @import("../../zero.zig"); + +pub const natsSubscriber = struct { + topic: []const u8, + name: []const u8, + exec: *const fn (*root.Context) anyerror!void, +}; diff --git a/src/pubsub/pubsub.zig b/src/pubsub/pubsub.zig deleted file mode 100644 index 9b3246e..0000000 --- a/src/pubsub/pubsub.zig +++ /dev/null @@ -1,4 +0,0 @@ -/// handles pubsub interface definition -const std = @import("std"); - -const pubsub = struct {}; diff --git a/src/service/client.zig b/src/service/client.zig index ad97a02..8cb22fa 100644 --- a/src/service/client.zig +++ b/src/service/client.zig @@ -32,7 +32,7 @@ pub fn create( const c = try ct.allocator.create(Client); // errdefer ct.allocator.destroy(c); - c.client = zul.http.Client.init(ct.allocator); + c.client = zul.http.Client.init(utils.io, ct.allocator); c.name = service_name; c.container = ct; c.url = _url; @@ -200,11 +200,11 @@ fn createAndSendRequest( req.body(body); } - var timer = try std.time.Timer.start(); + const start = utils.nowMonotonic(); var res = try req.getResponse(.{}); - const elapsed: f32 = @floatFromInt(timer.lap() / 1000000); + const elapsed: f32 = utils.elapsedMs(start); switch (res.status) { //expand more 404 => { diff --git a/src/tests.zig b/src/tests.zig index 3b1b07e..5b8c9c2 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -25,6 +25,9 @@ pub const kafkaConfig = @import("pubsub/kafka/config.zig"); pub const kafkaSubscriber = @import("pubsub/kafka/subscriber.zig"); pub const mqttConfig = @import("pubsub/mqtt/config.zig"); pub const mqttSubscriber = @import("pubsub/mqtt/subscriber.zig"); +pub const natsConfig = @import("pubsub/nats/config.zig"); +pub const natsSubscriber = @import("pubsub/nats/subscriber.zig"); +pub const pubsub = @import("pubsub/interface.zig"); comptime { _ = zero; @@ -52,4 +55,7 @@ comptime { _ = kafkaSubscriber; _ = mqttConfig; _ = mqttSubscriber; + _ = natsConfig; + _ = natsSubscriber; + _ = pubsub; } diff --git a/src/utils.zig b/src/utils.zig index cde9688..e1ab8c2 100644 --- a/src/utils.zig +++ b/src/utils.zig @@ -1,10 +1,35 @@ const std = @import("std"); +const builtin = @import("builtin"); const utils = @This(); const Self = @This(); const root = @import("zero.zig"); const dateTime = root.zdt.Datetime; +/// Global I/O reactor. Set once at startup (see `setIo`) and used by any +/// code that needs the clock or file descriptors under Zig 0.16's `std.Io`. +pub var io: std.Io = if (builtin.is_test) std.testing.io else undefined; + +pub fn setIo(i: std.Io) void { + io = i; +} + +pub fn nowMonotonic() std.Io.Timestamp { + return std.Io.Timestamp.now(io, .awake); +} + +pub fn nowReal() std.Io.Timestamp { + return std.Io.Timestamp.now(io, .real); +} + +pub fn elapsedNanos(start: std.Io.Timestamp) i96 { + return std.Io.Timestamp.durationTo(start, nowMonotonic()).nanoseconds; +} + +pub fn elapsedMs(start: std.Io.Timestamp) f32 { + return @floatFromInt(@as(u64, @intCast(@divTrunc(elapsedNanos(start), 1_000_000)))); +} + pub fn combine(allocator: std.mem.Allocator, comptime format: []const u8, value: anytype) ![]const u8 { var buffer: []u8 = undefined; buffer = try allocator.alloc(u8, 256); @@ -27,7 +52,7 @@ pub fn toStringFromInt(allocator: std.mem.Allocator, comptime format: []const u8 } pub fn timestampz(allocator: std.mem.Allocator) ![]const u8 { - const now = @as(u64, @intCast(std.time.timestamp())); + const now = @as(u64, @intCast(@divTrunc(nowReal().nanoseconds, 1_000_000_000))); const epoch_seconds = std.time.epoch.EpochSeconds{ .secs = now }; const time = epoch_seconds.getDaySeconds(); const hour = time.getHoursIntoDay(); @@ -43,7 +68,7 @@ pub fn sqlTimestampz(allocator: std.mem.Allocator) ![]const u8 { var buffer: []u8 = undefined; buffer = try allocator.alloc(u8, 100); - const now = dateTime.nowUTC(); + const now = dateTime.nowUTC(utils.io); const yr = @as(u64, @intCast(now.year)); //2000-01-01T07:24:22 diff --git a/src/websocket.zig b/src/websocket.zig index 82b7238..a7b73dd 100644 --- a/src/websocket.zig +++ b/src/websocket.zig @@ -40,3 +40,9 @@ pub fn afterInit(self: *WebSocket) !void { try self.conn.write("connected!"); try self.context.action(self.context); } + +// Called by httpz when the underlying connection closes; frees the +// heap-allocated Context created in Handler.ws. +pub fn close(self: *WebSocket) void { + self.context.deinit(); +} diff --git a/src/zero.zig b/src/zero.zig index 0f551f7..c0e661e 100644 --- a/src/zero.zig +++ b/src/zero.zig @@ -5,13 +5,14 @@ pub const constants = @import("constants.zig"); pub const zul = @import("zul"); pub const pgz = @import("pg"); pub const httpz = @import("httpz"); -pub const metriks = @import("metriks"); +// pub const metriks = @import("metricz"); pub const rediz = @import("rediz"); pub const dotenv = @import("dotenv"); pub const zdt = @import("zdt"); pub const regexp = @import("regexp"); pub const mqttz = @import("mqttz"); pub const jwt = @import("jwt"); +pub const natslib = @import("nats"); pub const rdkafka = @import("cimport.zig").librdkafka; pub const sqlitez = @import("sqlite"); @@ -56,6 +57,13 @@ pub const kafka = @import("pubsub/kafka/kafka.zig"); pub const kafkaSubscriber = @import("pubsub/kafka/subscriber.zig"); pub const kafkaMessage = @import("pubsub/kafka/message.zig").Message; +pub const natsConfig = @import("pubsub/nats/config.zig").natsConfig; +pub const natsSubscriber = @import("pubsub/nats/subscriber.zig").natsSubscriber; +pub const natsMessage = @import("pubsub/nats/message.zig").natsMessage; +pub const nats = @import("pubsub/nats/NATS.zig").NATS; +pub const pubsubInterface = @import("pubsub/interface.zig"); +pub const PubSub = pubsubInterface.Interface; + pub const WSHandler = @import("websocket.zig"); pub const WSMiddleware = @import("mw/ws.zig"); pub const WSClient = httpz.websocket.Conn; @@ -81,15 +89,14 @@ pub const App = @import("app.zig"); pub const std_options: std.Options = .{ .logFn = logger.custom, + .panicFn = panic, }; -fn panic(_: []const u8, _: ?*std.builtin.StackTrace, _: ?usize) noreturn { - var it = std.debug.StackIterator.init(@returnAddress(), null); - var ix: usize = 0; +fn panic(msg: []const u8, return_address: ?usize) noreturn { + _ = msg; std.log.err("=== Stack Trace ==============", .{}); - while (it.next()) |frame| : (ix += 1) { - std.log.err("#{d:0>2}: 0x{X:0>16}", .{ ix, frame }); - } + std.debug.dumpCurrentStackTrace(.{ .first_address = return_address }); + std.process.exit(1); } pub fn main() !void {} diff --git a/src/zsutil/cpu.zig b/src/zsutil/cpu.zig index 0dc637b..9cf5555 100644 --- a/src/zsutil/cpu.zig +++ b/src/zsutil/cpu.zig @@ -1,5 +1,6 @@ const std = @import("std"); const root = @import("../zero.zig"); +const utils = root.utils; const Context = root.Context; /// The path to the CPU information file. @@ -35,12 +36,12 @@ pub const CpuInfo = struct { /// /// Returns a `CpuInfo` struct containing the CPU information. pub fn info(ctx: *Context) !CpuInfo { - const file = try std.fs.openFileAbsolute("/proc/cpuinfo", .{}); - defer file.close(); + const file = try std.Io.Dir.openFileAbsolute(utils.io, "/proc/cpuinfo", .{}); + defer file.close(utils.io); var buffer: [1024]u8 = undefined; - const bytes_read = try file.readAll(&buffer); + const bytes_read = try file.readPositionalAll(utils.io, &buffer, 0); const contents = buffer[0..bytes_read]; var cpuinfo = CpuInfo{}; @@ -85,7 +86,7 @@ fn setValue(allocator: std.mem.Allocator, comptime T: type, value: *T, line: []c /// Returns the percentage of CPU usage as a `f32` value. pub fn percentageUsed() !f32 { const prev_stats = try usage(); - std.Thread.sleep(update_interval); + std.Io.sleep(utils.io, std.Io.Duration.fromNanoseconds(update_interval), .awake) catch {}; const curr_stats = try usage(); return calculateCpuUsage(prev_stats, curr_stats); } @@ -96,11 +97,11 @@ pub fn percentageUsed() !f32 { /// /// Returns a `CpuUsage` struct with the current CPU usage statistics, or an error if the data is invalid. pub fn usage() !CpuUsage { - const file = try std.fs.openFileAbsolute(stat_file, .{}); - defer file.close(); + const file = try std.Io.Dir.openFileAbsolute(utils.io, stat_file, .{}); + defer file.close(utils.io); var buffer: [256]u8 = undefined; - const bytes_read = try file.readAll(&buffer); + const bytes_read = try file.readPositionalAll(utils.io, &buffer, 0); const data = buffer[0..bytes_read]; var lines = std.mem.splitSequence(u8, data, "\n"); diff --git a/src/zsutil/host.zig b/src/zsutil/host.zig index db229d8..47cac39 100644 --- a/src/zsutil/host.zig +++ b/src/zsutil/host.zig @@ -1,6 +1,7 @@ const std = @import("std"); const testing = std.testing; const root = @import("../zero.zig"); +const utils = root.utils; const Context = root.Context; /// Retrieves the current process statistics. @@ -10,11 +11,11 @@ const Context = root.Context; /// /// Returns a `ProcessStatus` struct with the current memory usage statistics. pub fn usage(ctx: *Context) !Host { - const file = try std.fs.openFileAbsolute("/etc/os-release", .{}); - defer file.close(); + const file = try std.Io.Dir.openFileAbsolute(utils.io, "/etc/os-release", .{}); + defer file.close(utils.io); var buffer: [1024]u8 = undefined; - var bytes_read = try file.readAll(&buffer); + var bytes_read = try file.readPositionalAll(utils.io, &buffer, 0); var contents = buffer[0..bytes_read]; var lines = std.mem.splitSequence(u8, contents, "\n"); @@ -28,11 +29,11 @@ pub fn usage(ctx: *Context) !Host { try setValue(ctx.allocator, []const u8, &host.versionFull, line, "DEBIAN_VERSION_FULL="); } - const file2 = try std.fs.openFileAbsolute("/etc/hostname", .{}); - defer file2.close(); + const file2 = try std.Io.Dir.openFileAbsolute(utils.io, "/etc/hostname", .{}); + defer file2.close(utils.io); buffer = undefined; - bytes_read = try file2.readAll(&buffer); + bytes_read = try file2.readPositionalAll(utils.io, &buffer, 0); contents = buffer[0..bytes_read]; try setValue(ctx.allocator, []const u8, &host.hostname, contents, ""); diff --git a/src/zsutil/memory.zig b/src/zsutil/memory.zig index e8cb940..af68ddf 100644 --- a/src/zsutil/memory.zig +++ b/src/zsutil/memory.zig @@ -1,5 +1,6 @@ const std = @import("std"); const testing = std.testing; +const utils = @import("../utils.zig"); /// Retrieves the current memory usage statistics. /// @@ -7,11 +8,11 @@ const testing = std.testing; /// /// Returns a `MemUsage` struct with the current memory usage statistics. pub fn usage() !MemUsage { - const file = try std.fs.openFileAbsolute("/proc/meminfo", .{}); - defer file.close(); + const file = try std.Io.Dir.openFileAbsolute(utils.io, "/proc/meminfo", .{}); + defer file.close(utils.io); var buffer: [1024]u8 = undefined; - const bytes_read = try file.readAll(&buffer); + const bytes_read = try file.readPositionalAll(utils.io, &buffer, 0); const contents = buffer[0..bytes_read]; diff --git a/src/zsutil/process.zig b/src/zsutil/process.zig index aa5546d..b9d6279 100644 --- a/src/zsutil/process.zig +++ b/src/zsutil/process.zig @@ -1,5 +1,7 @@ const std = @import("std"); const testing = std.testing; +const root = @import("../zero.zig"); +const utils = root.utils; /// Retrieves the current process statistics. /// @@ -8,11 +10,11 @@ const testing = std.testing; /// /// Returns a `ProcessStatus` struct with the current memory usage statistics. pub fn usage(allocator: std.mem.Allocator, path: []const u8) !ProcessStatus { - const file = try std.fs.openFileAbsolute(path, .{}); - defer file.close(); + const file = try std.Io.Dir.openFileAbsolute(utils.io, path, .{}); + defer file.close(utils.io); var buffer: [1024]u8 = undefined; - const bytes_read = try file.readAll(&buffer); + const bytes_read = try file.readPositionalAll(utils.io, &buffer, 0); const contents = buffer[0..bytes_read];