deps: update nats.c to 3.9.3

Support for earlier zig versions has been dropped.

.gitmodules

@@ -1,3 +0,0 @@
-[submodule "deps/nats.c"]
-	path = deps/nats.c
-	url = https://github.com/nats-io/nats.c.git

README.md

@@ -1,6 +1,6 @@
 # NATS - Zig Client
 
-This is a Zig client library for the [NATS messaging system](https://nats.io). It's currently a thin wrapper over [NATS.c](https://github.com/nats-io/nats.c).
+This is a Zig client library for the [NATS messaging system](https://nats.io). It's currently a thin wrapper over [NATS.c](https://github.com/nats-io/nats.c), which is included and automatically built as part of the package.
 
 There are three main goals:
 
@@ -8,14 +8,58 @@ There are three main goals:
   2. Provide a native-feeling Zig client API.
   3. Support cross-compilation to the platforms that Zig supports.
 
-Right now, in service of goal 3, the underlying C library is built without certain features (notably, without TLS support and without streaming support) because those features require wrangling some complex transitive dependencies (OpenSSL and Protocol Buffers, respectively). Solving this limitation is somewhere on the roadmap, but it's not high priority.
+`nats.c` is compiled against a copy of LibreSSL that has been wrapped with the zig build system. This appears to work, but it notably is not specifically OpenSSL, so there may be corner cases around encrypted connections.
+
+# Status
+
+All basic `nats.c` APIs are wrapped. The JetStream APIs are not currently wrapped, and the streaming API is not wrapped. It is unlikely I will wrap these as I do not require them for my primary use case. Contributions on this front are welcome. People who are brave or desperate can use these APIs unwrapped through the exposed `nats.nats_c` object.
+
+In theory, all wrapped APIs are referenced in unit tests so that they are at least checked to compile correctly. The unit tests do not do much in the way of behavioral testing, under the assumption that the underlying C library is well tested. However, there may be some gaps in the test coverage around less-common APIs.
+
+The standard workflows around publishing and subscribing to messages seem to work well and feel (in my opinion) sufficiently Zig-like. Some of the APIs use getter/setter functions more heavily than I think a native Zig implementation would, due to the fact that the underlying C library is designed with a very clean opaque handle API style.
+
+
+# Zig Version Support
+
+Since the language is still under active development, any written Zig code is a moving target. The master branch targets zig `0.14`.
+
+# Using
+
+These bindings are ready-to-use with the Zig package manager. This means you will need to create a `build.zig.zon` and modify your `build.zig` to use the dependency.
+
+```sh
+# bootstrap your zig project if you haven't already
+zig init
+# add the nats-client dependency
+zig fetch --save git+https://github.com/epicyclic-dev/nats-client.git
+```
+
+You can then use `nats_client` in your `build.zig` with:
+
+```zig
+const nats_dep = b.dependency("nats_client", .{
+    .target = target,
+    .optimize = optimize,
+    .@"enable-libsodium" = true, // Use libsodium for optimized implementations of some signing routines
+    .@"enable-tls" = true, // enable SSL/TLS support
+    .@"force-host-verify" = true, // force hostname verification for TLS connections
+    .@"enable-streaming" = true, // build with support for NATS streaming extensions
+});
+your_exe.root_module.addImport("nats", nats_dep.artifact("nats"));
+```
 
 # Building
 
-Currently, a demonstration executable can be built in the standard fashion, i.e. by running `zig build`.
+Some basic example executables can be built using `zig build examples`. These examples expect you to be running a copy of `nats-server` listening for unencrypted connections on `localhost:4222` (the default NATS port).
+
+# Testing
+
+Unit tests can be run using `zig build test`. The unit tests expect an executable named `nats-server` to be in your PATH in order to run properly.
 
 # License
 
+Unless noted otherwise (check file headers), all source code is licensed under the Apache License, Version 2.0 (which is also the `nats.c` license).
+
 ```
 Licensed under the Apache License, Version 2.0 (the "License");
 You may obtain a copy of the License at

build.zig

@@ -1,49 +1,83 @@
 // This file is licensed under the CC0 1.0 license.
 // See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
-
 const std = @import("std");
-const nats_build = @import("./nats-c.build.zig");
 
 pub fn build(b: *std.Build) void {
     const target = b.standardTargetOptions(.{});
     const optimize = b.standardOptimizeOption(.{});
 
-    // const nats = b.addModule("nats", .{
-    //     .source_file = .{ .path = "source/nats.zig" },
-    // });
+    const enable_libsodium = b.option(bool, "enable-libsodium", "Build with libsodium for higher-performance signing (default: true)") orelse true;
+    const enable_tls = b.option(bool, "enable-tls", "Build TLS support (default: true)") orelse true;
+    const tls_verify = b.option(bool, "force-host-verify", "Force hostname verification for TLS connections (default: true)") orelse true;
+    const enable_streaming = b.option(bool, "enable-streaming", "Build with streaming support (default: true)") orelse true;
 
-    const nats = b.addExecutable(.{
-        .name = "nats_test",
-        .root_source_file = .{ .path = "src/nats.zig" },
+    const nats = b.addModule("nats", .{
+        .root_source_file = b.path("src/nats.zig"),
+    });
+
+    const nats_c = b.dependency("nats_c", .{
+        .target = target,
+        .optimize = optimize,
+        .@"enable-libsodium" = enable_libsodium,
+        .@"enable-tls" = enable_tls,
+        .@"force-host-verify" = tls_verify,
+        .@"enable-streaming" = enable_streaming,
+    });
+    nats.linkLibrary(nats_c.artifact("nats"));
+
+    const tests = b.addTest(.{
+        .name = "nats-zig-unit-tests",
+        .root_source_file = b.path("tests/main.zig"),
         .target = target,
         .optimize = optimize,
     });
 
-    const tinfo = nats.target_info.target;
+    tests.root_module.addImport("nats", nats);
+    tests.linkLibrary(nats_c.artifact("nats"));
 
-    const nats_c = nats_build.nats_c_lib(
-        b,
-        .{ .name = "nats-c", .target = target, .optimize = optimize },
-    );
-    switch (tinfo.os.tag) {
-        .windows => {
-            if (tinfo.abi != .msvc) {
-                nats_c.addCSourceFiles(&.{"src/win-crosshack.c"}, &.{"-fno-sanitize=undefined"});
-            }
-        },
-        else => {},
-    }
-
-    nats.linkLibrary(nats_c);
-    b.installArtifact(nats);
-
-    const main_tests = b.addTest(.{
-        .root_source_file = .{ .path = "src/nats.zig" },
-        .target = target,
-        .optimize = optimize,
-    });
-
-    const run_main_tests = b.addRunArtifact(main_tests);
+    const run_main_tests = b.addRunArtifact(tests);
     const test_step = b.step("test", "Run tests");
+    test_step.dependOn(&b.addInstallArtifact(tests, .{}).step);
     test_step.dependOn(&run_main_tests.step);
+
+    add_examples(b, .{
+        .target = target,
+        .optimize = optimize,
+        .nats_module = nats,
+    });
+}
+
+const ExampleOptions = struct {
+    target: std.Build.ResolvedTarget,
+    optimize: std.builtin.OptimizeMode,
+    nats_module: *std.Build.Module,
+};
+
+const Example = struct {
+    name: []const u8,
+    file: []const u8,
+};
+
+const examples = [_]Example{
+    .{ .name = "request_reply", .file = "examples/request_reply.zig" },
+    .{ .name = "headers", .file = "examples/headers.zig" },
+    .{ .name = "pub_bytes", .file = "examples/pub_bytes.zig" },
+};
+
+pub fn add_examples(b: *std.Build, options: ExampleOptions) void {
+    const example_step = b.step("examples", "build examples");
+
+    inline for (examples) |example| {
+        const ex_exe = b.addExecutable(.{
+            .name = example.name,
+            .root_source_file = b.path(example.file),
+            .target = options.target,
+            .optimize = .Debug,
+        });
+
+        ex_exe.root_module.addImport("nats", options.nats_module);
+
+        const install = b.addInstallArtifact(ex_exe, .{});
+        example_step.dependOn(&install.step);
+    }
 }

build.zig.zon

@@ -1,5 +1,18 @@
 .{
-    .name = "nats-client",
-    .version = "0.0.1",
-    .dependencies = .{},
+    .name = .nats_client,
+    .fingerprint = 0x7B8B5B9E2C2BD086,
+    .version = "0.1.0",
+    .minimum_zig_version = "0.14.0",
+    .paths = .{
+        "src",
+        "build.zig",
+        "build.zig.zon",
+        "LICENSE",
+    },
+    .dependencies = .{
+        .nats_c = .{
+            .url = "git+https://github.com/allyourcodebase/nats.c.git?ref=3.9.3#81a546c524f7c3db16606b85b5d729b393d8667e",
+            .hash = "nats_c-3.9.3-0V8jnNdSAACkcRB0PdrHECfsTXSiySA6OeAOx1e83iaM",
+        },
+    },
 }

examples/headers.zig

@@ -0,0 +1,71 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+const std = @import("std");
+const nats = @import("nats");
+
+pub fn main() !void {
+    const connection = try nats.Connection.connectTo(nats.default_server_url);
+    defer connection.destroy();
+
+    const message = try nats.Message.create("subject", null, "message");
+    defer message.destroy();
+
+    try message.setHeaderValue("foo", "foo-value");
+    try message.setHeaderValue("bar", "bar-value");
+    try message.addHeaderValue("foo", "bar-value");
+    try message.setHeaderValue("baz", "baz-value");
+    try message.addHeaderValue("qux", "qux-value");
+
+    try message.deleteHeader("baz");
+
+    {
+        var iter = try message.getHeaderIterator();
+        defer iter.destroy();
+
+        while (iter.next()) |header| {
+            var val_iter = try header.valueIterator();
+            defer val_iter.destroy();
+
+            std.debug.print("key '{s}' got: ", .{header.key});
+            while (val_iter.next()) |value| {
+                std.debug.print("'{s}'{s}", .{ value, if (val_iter.peek()) |_| ", " else "" });
+            }
+            std.debug.print("\n", .{});
+        }
+    }
+
+    const subscription = try connection.subscribeSync("subject");
+    defer subscription.destroy();
+
+    try connection.publishMessage(message);
+    const received = try subscription.nextMessage(1000);
+    defer received.destroy();
+
+    {
+        var iter = try received.getHeaderValueIterator("foo");
+        defer iter.destroy();
+
+        std.debug.print("For key 'foo' got: ", .{});
+        while (iter.next()) |value| {
+            std.debug.print("'{s}'{s}", .{ value, if (iter.peek()) |_| ", " else "" });
+        }
+        std.debug.print("\n", .{});
+    }
+
+    _ = received.getHeaderValue("key-does-not-exist") catch |err| switch (err) {
+        nats.Error.NotFound => {},
+        else => {
+            std.debug.print("Should not have found that key!", .{});
+            return err;
+        },
+    };
+
+    received.deleteHeader("key-does-not-exist") catch |err| switch (err) {
+        nats.Error.NotFound => {},
+        else => {
+            std.debug.print("Should not have found that key!", .{});
+            return err;
+        },
+    };
+}

examples/pub_bytes.zig

@@ -0,0 +1,13 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+const std = @import("std");
+const nats = @import("nats");
+
+pub fn main() !void {
+    const connection = try nats.Connection.connectTo(nats.default_server_url);
+    defer connection.destroy();
+
+    const data = [_]u8{ 104, 101, 108, 108, 111, 33 };
+    try connection.publish("subject", &data);
+}

examples/request_reply.zig

@@ -0,0 +1,50 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+const std = @import("std");
+const nats = @import("nats");
+
+fn onMessage(
+    userdata: *u32,
+    connection: *nats.Connection,
+    subscription: *nats.Subscription,
+    message: *nats.Message,
+) void {
+    _ = subscription;
+
+    std.debug.print("Subject \"{s}\" received message: \"{s}\"\n", .{
+        message.getSubject(),
+        message.getData() orelse "[null]",
+    });
+
+    if (message.getReply()) |reply| {
+        connection.publish(reply, "salutations") catch @panic("HELP");
+    }
+
+    userdata.* += 1;
+}
+
+pub fn main() !void {
+    const connection = try nats.Connection.connectTo(nats.default_server_url);
+    defer connection.destroy();
+
+    var count: u32 = 0;
+    const subscription = try connection.subscribe(*u32, "channel", onMessage, &count);
+    defer subscription.destroy();
+
+    while (count < 10) : (nats.sleep(100)) {
+        const reply = try connection.request("channel", "greetings", 1000);
+        defer reply.destroy();
+
+        std.debug.print("Reply \"{s}\" got message: {s}\n", .{
+            reply.getSubject(),
+            reply.getData() orelse "[null]",
+        });
+    }
+
+    const stats = try connection.getStats();
+    std.debug.print(
+        "Server stats => {{\n\tmessages_in: {d} ({d} B),\n\tmessages_out: {d} ({d} B),\n\treconnects: {d}\n}}\n",
+        .{ stats.messages_in, stats.bytes_in, stats.messages_out, stats.bytes_out, stats.reconnects },
+    );
+}

nats-c.build.zig

@@ -1,127 +0,0 @@
-// This file is licensed under the CC0 1.0 license.
-// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
-
-const std = @import("std");
-
-const NatsCOptions = struct {
-    name: []const u8,
-    target: std.zig.CrossTarget,
-    optimize: std.builtin.OptimizeMode,
-};
-
-pub fn nats_c_lib(
-    b: *std.Build,
-    options: NatsCOptions,
-) *std.Build.Step.Compile {
-    const lib = b.addStaticLibrary(.{
-        .name = options.name,
-        .target = options.target,
-        .optimize = options.optimize,
-    });
-
-    lib.disable_sanitize_c = true;
-
-    lib.linkLibC();
-    lib.addCSourceFiles(&common_sources, &.{"-fno-sanitize=undefined"});
-    lib.addIncludePath(.{ .path = nats_src_prefix ++ "include" });
-    // if building with streaming support
-    // lib.addIncludePath(.{ .path = nats_src_prefix ++ "stan" });
-    // lib.addCSourceFiles(&streaming_sources, &.{"-fno-sanitize=undefined"});
-
-    switch (lib.target_info.target.os.tag) {
-        .windows => {
-            lib.addCSourceFiles(&win_sources, &.{"-fno-sanitize=undefined"});
-            lib.defineCMacro("_WIN32", null);
-            lib.linkSystemLibrary("Ws2_32");
-        },
-        .macos => {
-            lib.addCSourceFiles(&unix_sources, &.{"-fno-sanitize=undefined"});
-            lib.defineCMacro("DARWIN", null);
-        },
-        else => {
-            lib.addCSourceFiles(&unix_sources, &.{"-fno-sanitize=undefined"});
-            lib.defineCMacro("_GNU_SOURCE", null);
-            lib.defineCMacro("LINUX", null);
-            // may need to link pthread and rt. Not sure if those are inluded with linkLibC
-            lib.linkSystemLibrary("pthread");
-            lib.linkSystemLibrary("rt");
-        },
-    }
-
-    lib.defineCMacro("_REENTRANT", null);
-
-    inline for (install_headers) |header| {
-        lib.installHeader(nats_src_prefix ++ header, "nats/" ++ header);
-    }
-
-    b.installArtifact(lib);
-
-    return lib;
-}
-
-pub fn build(b: *std.Build) void {
-    const target = b.standardTargetOptions(.{});
-    const optimize = b.standardOptimizeOption(.{});
-
-    _ = nats_c_lib(b, .{ .name = "nats-c", .target = target, .optimize = optimize });
-}
-
-const nats_src_prefix = "deps/nats.c/src/";
-
-const install_headers = [_][]const u8{
-    "nats.h",
-    "status.h",
-    "version.h",
-};
-
-const common_sources = [_][]const u8{
-    nats_src_prefix ++ "asynccb.c",
-    nats_src_prefix ++ "comsock.c",
-    nats_src_prefix ++ "crypto.c",
-    nats_src_prefix ++ "js.c",
-    nats_src_prefix ++ "kv.c",
-    nats_src_prefix ++ "nats.c",
-    nats_src_prefix ++ "nkeys.c",
-    nats_src_prefix ++ "opts.c",
-    nats_src_prefix ++ "pub.c",
-    nats_src_prefix ++ "stats.c",
-    nats_src_prefix ++ "sub.c",
-    nats_src_prefix ++ "url.c",
-    nats_src_prefix ++ "buf.c",
-    nats_src_prefix ++ "conn.c",
-    nats_src_prefix ++ "hash.c",
-    nats_src_prefix ++ "jsm.c",
-    nats_src_prefix ++ "msg.c",
-    nats_src_prefix ++ "natstime.c",
-    nats_src_prefix ++ "nuid.c",
-    nats_src_prefix ++ "parser.c",
-    nats_src_prefix ++ "srvpool.c",
-    nats_src_prefix ++ "status.c",
-    nats_src_prefix ++ "timer.c",
-    nats_src_prefix ++ "util.c",
-};
-
-const unix_sources = [_][]const u8{
-    nats_src_prefix ++ "unix/cond.c",
-    nats_src_prefix ++ "unix/mutex.c",
-    nats_src_prefix ++ "unix/sock.c",
-    nats_src_prefix ++ "unix/thread.c",
-};
-
-const win_sources = [_][]const u8{
-    nats_src_prefix ++ "win/cond.c",
-    nats_src_prefix ++ "win/mutex.c",
-    nats_src_prefix ++ "win/sock.c",
-    nats_src_prefix ++ "win/strings.c",
-    nats_src_prefix ++ "win/thread.c",
-};
-
-const streaming_sources = [_][]const u8{
-    nats_src_prefix ++ "stan/conn.c",
-    nats_src_prefix ++ "stan/copts.c",
-    nats_src_prefix ++ "stan/msg.c",
-    nats_src_prefix ++ "stan/protocol.pb-c.c",
-    nats_src_prefix ++ "stan/pub.c",
-    nats_src_prefix ++ "stan/sopts.c",
-    nats_src_prefix ++ "stan/sub.c",
-};

src/error.zig

@@ -0,0 +1,323 @@
+// Copyright 2023 torque@epicyclic.dev
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+const std = @import("std");
+
+const nats_c = @import("./nats_c.zig").nats_c;
+
+// pub const AllocError = Error || std.mem.Allocator.Error;
+
+pub const ErrorInfo = struct {
+    code: ?Error,
+    desc: [:0]const u8,
+};
+
+pub fn getLastError() ErrorInfo {
+    var status: c_uint = 0;
+    const desc = nats_c.nats_GetLastError(&status);
+
+    return .{
+        .code = Status.fromInt(status).toError(),
+        .desc = std.mem.sliceTo(desc, 0),
+    };
+}
+
+pub fn getLastErrorStack(buffer: *[]u8) Error!void {
+    const status = Status.fromInt(nats_c.nats_GetLastErrorStack(buffer.ptr, buffer.len));
+    return status.raise();
+}
+
+// NATS_EXTERN void nats_PrintLastErrorStack(FILE *file);
+
+pub const Status = enum(c_int) {
+    okay = nats_c.NATS_OK,
+    generic_error = nats_c.NATS_ERR,
+    protocol_error = nats_c.NATS_PROTOCOL_ERROR,
+    io_error = nats_c.NATS_IO_ERROR,
+    line_too_long = nats_c.NATS_LINE_TOO_LONG,
+    connection_closed = nats_c.NATS_CONNECTION_CLOSED,
+    no_server = nats_c.NATS_NO_SERVER,
+    stale_connection = nats_c.NATS_STALE_CONNECTION,
+    secure_connection_wanted = nats_c.NATS_SECURE_CONNECTION_WANTED,
+    secure_connection_required = nats_c.NATS_SECURE_CONNECTION_REQUIRED,
+    connection_disconnected = nats_c.NATS_CONNECTION_DISCONNECTED,
+    connection_auth_failed = nats_c.NATS_CONNECTION_AUTH_FAILED,
+    not_permitted = nats_c.NATS_NOT_PERMITTED,
+    not_found = nats_c.NATS_NOT_FOUND,
+    address_missing = nats_c.NATS_ADDRESS_MISSING,
+    invalid_subject = nats_c.NATS_INVALID_SUBJECT,
+    invalid_arg = nats_c.NATS_INVALID_ARG,
+    invalid_subscription = nats_c.NATS_INVALID_SUBSCRIPTION,
+    invalid_timeout = nats_c.NATS_INVALID_TIMEOUT,
+    illegal_state = nats_c.NATS_ILLEGAL_STATE,
+    slow_consumer = nats_c.NATS_SLOW_CONSUMER,
+    max_payload = nats_c.NATS_MAX_PAYLOAD,
+    max_delivered_msgs = nats_c.NATS_MAX_DELIVERED_MSGS,
+    insufficient_buffer = nats_c.NATS_INSUFFICIENT_BUFFER,
+    no_memory = nats_c.NATS_NO_MEMORY,
+    sys_error = nats_c.NATS_SYS_ERROR,
+    timeout = nats_c.NATS_TIMEOUT,
+    failed_to_initialize = nats_c.NATS_FAILED_TO_INITIALIZE,
+    not_initialized = nats_c.NATS_NOT_INITIALIZED,
+    ssl_error = nats_c.NATS_SSL_ERROR,
+    no_server_support = nats_c.NATS_NO_SERVER_SUPPORT,
+    not_yet_connected = nats_c.NATS_NOT_YET_CONNECTED,
+    draining = nats_c.NATS_DRAINING,
+    invalid_queue_name = nats_c.NATS_INVALID_QUEUE_NAME,
+    no_responders = nats_c.NATS_NO_RESPONDERS,
+    mismatch = nats_c.NATS_MISMATCH,
+    missed_heartbeat = nats_c.NATS_MISSED_HEARTBEAT,
+    _,
+
+    // this is a weird quirk of translate-c. All the enum members are translated as independent
+    // constants of type c_int, but the enum type itself is translated as c_uint.
+    pub fn fromInt(int: c_uint) Status {
+        return @enumFromInt(int);
+    }
+
+    pub fn toInt(self: Status) c_uint {
+        return @intFromEnum(self);
+    }
+
+    pub fn description(self: Status) [:0]const u8 {
+        return std.mem.sliceTo(nats_c.natsStatus_GetText(self), 0);
+    }
+
+    pub fn raise(self: Status) Error!void {
+        return switch (self) {
+            .okay => void{},
+            .generic_error => Error.GenericError,
+            .protocol_error => Error.ProtocolError,
+            .io_error => Error.IoError,
+            .line_too_long => Error.LineTooLong,
+            .connection_closed => Error.ConnectionClosed,
+            .no_server => Error.NoServer,
+            .stale_connection => Error.StaleConnection,
+            .secure_connection_wanted => Error.SecureConnectionWanted,
+            .secure_connection_required => Error.SecureConnectionRequired,
+            .connection_disconnected => Error.ConnectionDisconnected,
+            .connection_auth_failed => Error.ConnectionAuthFailed,
+            .not_permitted => Error.NotPermitted,
+            .not_found => Error.NotFound,
+            .address_missing => Error.AddressMissing,
+            .invalid_subject => Error.InvalidSubject,
+            .invalid_arg => Error.InvalidArg,
+            .invalid_subscription => Error.InvalidSubscription,
+            .invalid_timeout => Error.InvalidTimeout,
+            .illegal_state => Error.IllegalState,
+            .slow_consumer => Error.SlowConsumer,
+            .max_payload => Error.MaxPayload,
+            .max_delivered_msgs => Error.MaxDeliveredMsgs,
+            .insufficient_buffer => Error.InsufficientBuffer,
+            .no_memory => Error.NoMemory,
+            .sys_error => Error.SysError,
+            .timeout => Error.Timeout,
+            .failed_to_initialize => Error.FailedToInitialize,
+            .not_initialized => Error.NotInitialized,
+            .ssl_error => Error.SslError,
+            .no_server_support => Error.NoServerSupport,
+            .not_yet_connected => Error.NotYetConnected,
+            .draining => Error.Draining,
+            .invalid_queue_name => Error.InvalidQueueName,
+            .no_responders => Error.NoResponders,
+            .mismatch => Error.Mismatch,
+            .missed_heartbeat => Error.MissedHeartbeat,
+            _ => Error.UnknownError,
+        };
+    }
+
+    pub fn toError(self: Status) ?Error {
+        return switch (self) {
+            .okay => null,
+            .generic_error => Error.GenericError,
+            .protocol_error => Error.ProtocolError,
+            .io_error => Error.IoError,
+            .line_too_long => Error.LineTooLong,
+            .connection_closed => Error.ConnectionClosed,
+            .no_server => Error.NoServer,
+            .stale_connection => Error.StaleConnection,
+            .secure_connection_wanted => Error.SecureConnectionWanted,
+            .secure_connection_required => Error.SecureConnectionRequired,
+            .connection_disconnected => Error.ConnectionDisconnected,
+            .connection_auth_failed => Error.ConnectionAuthFailed,
+            .not_permitted => Error.NotPermitted,
+            .not_found => Error.NotFound,
+            .address_missing => Error.AddressMissing,
+            .invalid_subject => Error.InvalidSubject,
+            .invalid_arg => Error.InvalidArg,
+            .invalid_subscription => Error.InvalidSubscription,
+            .invalid_timeout => Error.InvalidTimeout,
+            .illegal_state => Error.IllegalState,
+            .slow_consumer => Error.SlowConsumer,
+            .max_payload => Error.MaxPayload,
+            .max_delivered_msgs => Error.MaxDeliveredMsgs,
+            .insufficient_buffer => Error.InsufficientBuffer,
+            .no_memory => Error.NoMemory,
+            .sys_error => Error.SysError,
+            .timeout => Error.Timeout,
+            .failed_to_initialize => Error.FailedToInitialize,
+            .not_initialized => Error.NotInitialized,
+            .ssl_error => Error.SslError,
+            .no_server_support => Error.NoServerSupport,
+            .not_yet_connected => Error.NotYetConnected,
+            .draining => Error.Draining,
+            .invalid_queue_name => Error.InvalidQueueName,
+            .no_responders => Error.NoResponders,
+            .mismatch => Error.Mismatch,
+            .missed_heartbeat => Error.MissedHeartbeat,
+            _ => Error.UnknownError,
+        };
+    }
+
+    pub fn fromError(err: ?anyerror) Status {
+        return if (err) |e|
+            switch (e) {
+                Error.ProtocolError => .protocol_error,
+                Error.IoError => .io_error,
+                Error.LineTooLong => .line_too_long,
+                Error.ConnectionClosed => .connection_closed,
+                Error.NoServer => .no_server,
+                Error.StaleConnection => .stale_connection,
+                Error.SecureConnectionWanted => .secure_connection_wanted,
+                Error.SecureConnectionRequired => .secure_connection_required,
+                Error.ConnectionDisconnected => .connection_disconnected,
+                Error.ConnectionAuthFailed => .connection_auth_failed,
+                Error.NotPermitted => .not_permitted,
+                Error.NotFound => .not_found,
+                Error.AddressMissing => .address_missing,
+                Error.InvalidSubject => .invalid_subject,
+                Error.InvalidArg => .invalid_arg,
+                Error.InvalidSubscription => .invalid_subscription,
+                Error.InvalidTimeout => .invalid_timeout,
+                Error.IllegalState => .illegal_state,
+                Error.SlowConsumer => .slow_consumer,
+                Error.MaxPayload => .max_payload,
+                Error.MaxDeliveredMsgs => .max_delivered_msgs,
+                Error.InsufficientBuffer => .insufficient_buffer,
+                Error.NoMemory => .no_memory,
+                Error.SysError => .sys_error,
+                Error.Timeout => .timeout,
+                Error.FailedToInitialize => .failed_to_initialize,
+                Error.NotInitialized => .not_initialized,
+                Error.SslError => .ssl_error,
+                Error.NoServerSupport => .no_server_support,
+                Error.NotYetConnected => .not_yet_connected,
+                Error.Draining => .draining,
+                Error.InvalidQueueName => .invalid_queue_name,
+                Error.NoResponders => .no_responders,
+                Error.Mismatch => .mismatch,
+                Error.MissedHeartbeat => .missed_heartbeat,
+                else => .generic_error,
+            }
+        else
+            .okay;
+    }
+};
+
+pub const Error = error{
+    /// Generic error
+    GenericError,
+    /// Error when parsing a protocol message, or not getting the expected message.
+    ProtocolError,
+    /// IO Error (network communication).
+    IoError,
+    /// The protocol message read from the socket does not fit in the read buffer.
+    LineTooLong,
+
+    /// Operation on this connection failed because the connection is closed.
+    ConnectionClosed,
+    /// Unable to connect, the server could not be reached or is not running.
+    NoServer,
+    /// The server closed our connection because it did not receive PINGs at the expected interval.
+    StaleConnection,
+    /// The client is configured to use TLS, but the server is not.
+    SecureConnectionWanted,
+    /// The server expects a TLS connection.
+    SecureConnectionRequired,
+    /// The connection was disconnected. Depending on the configuration, the connection may reconnect.
+    ConnectionDisconnected,
+
+    /// The connection failed due to authentication error.
+    ConnectionAuthFailed,
+    /// The action is not permitted.
+    NotPermitted,
+    /// An action could not complete because something was not found. So far, this is an internal error.
+    NotFound,
+
+    /// Incorrect URL. For instance no host specified in the URL.
+    AddressMissing,
+
+    /// Invalid subject, for instance NULL or empty string.
+    InvalidSubject,
+    /// An invalid argument is passed to a function. For instance passing NULL to an API that does not accept this value.
+    InvalidArg,
+    /// The call to a subscription function fails because the subscription has previously been closed.
+    InvalidSubscription,
+    /// Timeout must be positive numbers.
+    InvalidTimeout,
+
+    /// An unexpected state, for instance calling natsSubscription_NextMsg on an asynchronous subscriber.
+    IllegalState,
+
+    /// The maximum number of messages waiting to be delivered has been reached. Messages are dropped.
+    SlowConsumer,
+
+    /// Attempt to send a payload larger than the maximum allowed by the NATS Server.
+    MaxPayload,
+    /// Attempt to receive more messages than allowed, for instance because of #natsSubscription_AutoUnsubscribe().
+    MaxDeliveredMsgs,
+
+    /// A buffer is not large enough to accommodate the data.
+    InsufficientBuffer,
+
+    /// An operation could not complete because of insufficient memory.
+    NoMemory,
+
+    /// Some system function returned an error.
+    SysError,
+
+    /// An operation timed-out. For instance #natsSubscription_NextMsg().
+    Timeout,
+
+    /// The library failed to initialize.
+    FailedToInitialize,
+    /// The library is not yet initialized.
+    NotInitialized,
+
+    /// An SSL error occurred when trying to establish a connection.
+    SslError,
+
+    /// The server does not support this action.
+    NoServerSupport,
+
+    /// A connection could not be immediately established and #natsOptions_SetRetryOnFailedConnect() specified a connected callback. The connect is retried asynchronously.
+    NotYetConnected,
+
+    /// A connection and/or subscription entered the draining mode. Some operations will fail when in that mode.
+    Draining,
+
+    /// An invalid queue name was passed when creating a queue subscription.
+    InvalidQueueName,
+
+    /// No responders were running when the server received the request.
+    NoResponders,
+
+    /// For JetStream subscriptions, it means that a consumer sequence mismatch was discovered.
+    Mismatch,
+    /// For JetStream subscriptions, it means that the library detected that server heartbeats have been missed.
+    MissedHeartbeat,
+
+    /// The C API has returned an error that the zig layer does not know about.
+    UnknownError,
+};

src/message.zig

@@ -0,0 +1,220 @@
+// Copyright 2023 torque@epicyclic.dev
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+const std = @import("std");
+
+const nats_c = @import("./nats_c.zig").nats_c;
+
+const err_ = @import("./error.zig");
+const Error = err_.Error;
+const Status = err_.Status;
+
+pub const Message = opaque {
+    pub fn create(subject: [:0]const u8, reply: ?[:0]const u8, data: ?[]const u8) Error!*Message {
+        var self: *Message = undefined;
+        const status = Status.fromInt(nats_c.natsMsg_Create(
+            @ptrCast(&self),
+            subject.ptr,
+            if (reply) |r| r.ptr else null,
+            if (data) |d| d.ptr else null,
+            if (data) |d| @intCast(d.len) else 0,
+        ));
+
+        return status.toError() orelse self;
+    }
+
+    pub fn destroy(self: *Message) void {
+        nats_c.natsMsg_Destroy(@ptrCast(self));
+    }
+
+    pub fn getSubject(self: *Message) [:0]const u8 {
+        const subject = nats_c.natsMsg_GetSubject(@ptrCast(self)) orelse unreachable;
+        return std.mem.sliceTo(subject, 0);
+    }
+
+    pub fn getReply(self: *Message) ?[:0]const u8 {
+        const reply = nats_c.natsMsg_GetReply(@ptrCast(self)) orelse return null;
+        return std.mem.sliceTo(reply, 0);
+    }
+
+    pub fn getData(self: *Message) ?[:0]const u8 {
+        const data = nats_c.natsMsg_GetData(@ptrCast(self)) orelse return null;
+        return data[0..self.getDataLength() :0];
+    }
+
+    pub fn getDataLength(self: *Message) usize {
+        return @intCast(nats_c.natsMsg_GetDataLength(@ptrCast(self)));
+    }
+
+    pub fn setHeaderValue(self: *Message, key: [:0]const u8, value: [:0]const u8) Error!void {
+        const status = Status.fromInt(nats_c.natsMsgHeader_Set(@ptrCast(self), key.ptr, value.ptr));
+        return status.raise();
+    }
+
+    pub fn addHeaderValue(self: *Message, key: [:0]const u8, value: [:0]const u8) Error!void {
+        const status = Status.fromInt(nats_c.natsMsgHeader_Add(@ptrCast(self), key.ptr, value.ptr));
+        return status.raise();
+    }
+
+    pub fn getHeaderValue(self: *Message, key: [:0]const u8) Error![:0]const u8 {
+        var value: [*c]const u8 = null;
+        const status = Status.fromInt(nats_c.natsMsgHeader_Get(@ptrCast(self), key.ptr, &value));
+
+        return status.toError() orelse std.mem.sliceTo(value.?, 0);
+    }
+
+    pub fn getHeaderValueIterator(self: *Message, key: [:0]const u8) Error!HeaderValueIterator {
+        return .{ .values = try self.getAllHeaderValues(key) };
+    }
+
+    pub fn getHeaderIterator(self: *Message) Error!HeaderIterator {
+        return .{
+            .message = self,
+            .keys = try self.getAllHeaderKeys(),
+        };
+    }
+
+    pub fn deleteHeader(self: *Message, key: [:0]const u8) Error!void {
+        const status = Status.fromInt(nats_c.natsMsgHeader_Delete(@ptrCast(self), key.ptr));
+        return status.raise();
+    }
+
+    pub fn isNoResponders(self: *Message) bool {
+        return nats_c.natsMsg_IsNoResponders(@ptrCast(self));
+    }
+
+    // prefer using message.getHeaderValueIterator
+    pub fn getAllHeaderValues(self: *Message, key: [:0]const u8) Error![][*:0]const u8 {
+        var values: [*][*:0]const u8 = undefined;
+        var count: c_int = 0;
+
+        const status = Status.fromInt(
+            nats_c.natsMsgHeader_Values(@ptrCast(self), key.ptr, @ptrCast(&values), &count),
+        );
+
+        // the user must use std.mem.spanTo on each item they want to read to get a
+        // slice, since we can't do that automatically without having to allocate.
+        return status.toError() orelse values[0..@intCast(count)];
+    }
+
+    // prefer using message.getHeaderIterator
+    pub fn getAllHeaderKeys(self: *Message) Error![][*:0]const u8 {
+        var keys: [*][*:0]const u8 = undefined;
+        var count: c_int = 0;
+
+        const status = Status.fromInt(nats_c.natsMsgHeader_Keys(@ptrCast(self), @ptrCast(&keys), &count));
+
+        // the user must use std.mem.spanTo on each item they want to read to get a
+        // slice, since we can't do that automatically without having to allocate.
+        // the returned slice
+        return status.toError() orelse keys[0..@intCast(count)];
+    }
+
+    pub const HeaderValueIterator = struct {
+        values: [][*:0]const u8,
+        index: usize = 0,
+
+        pub fn destroy(self: HeaderValueIterator) void {
+            std.heap.raw_c_allocator.free(self.values);
+        }
+
+        pub const deinit = HeaderValueIterator.destroy;
+
+        pub fn next(self: *HeaderValueIterator) ?[:0]const u8 {
+            if (self.index >= self.values.len) return null;
+            defer self.index += 1;
+
+            return std.mem.sliceTo(self.values[self.index], 0);
+        }
+
+        pub fn peek(self: *HeaderValueIterator) ?[:0]const u8 {
+            if (self.index >= self.values.len) return null;
+            return std.mem.sliceTo(self.values[self.index], 0);
+        }
+    };
+
+    pub const HeaderIterator = struct {
+        message: *Message,
+        keys: [][*:0]const u8,
+        index: usize = 0,
+
+        pub const ValueResolver = struct {
+            message: *Message,
+            key: [:0]const u8,
+
+            pub fn value(self: ValueResolver) Error![:0]const u8 {
+                // TODO: if we didn't care about the lifecycle of self.message, we
+                // could do catch unreachable here and make this error-free
+                return try self.message.getHeaderValue(self.key);
+            }
+
+            pub fn valueIterator(self: ValueResolver) Error!HeaderValueIterator {
+                return try self.message.getHeaderValueIterator(self.key);
+            }
+        };
+
+        pub fn destroy(self: *HeaderIterator) void {
+            std.heap.raw_c_allocator.free(self.keys);
+        }
+
+        pub const deinit = HeaderIterator.destroy;
+
+        pub fn next(self: *HeaderIterator) ?ValueResolver {
+            if (self.index >= self.keys.len) return null;
+            defer self.index += 1;
+
+            return .{
+                .message = self.message,
+                .key = std.mem.sliceTo(self.keys[self.index], 0),
+            };
+        }
+
+        pub fn peek(self: *HeaderIterator) ?ValueResolver {
+            if (self.index >= self.keys.len) return null;
+            return .{
+                .message = self.message,
+                .key = std.mem.sliceTo(self.keys[self.index], 0),
+            };
+        }
+
+        pub fn nextKey(self: *HeaderIterator) ?[:0]const u8 {
+            if (self.index >= self.keys.len) return null;
+            defer self.index += 1;
+            return std.mem.sliceTo(self.keys[self.index], 0);
+        }
+
+        pub fn peekKey(self: *HeaderIterator) ?[:0]const u8 {
+            if (self.index >= self.keys.len) return null;
+            return std.mem.sliceTo(self.keys[self.index], 0);
+        }
+    };
+};
+
+// TODO: not implementing jetstream API right now
+// NATS_EXTERN natsStatus natsMsg_Ack(natsMsg *msg, jsOptions *opts);
+// NATS_EXTERN natsStatus natsMsg_AckSync(natsMsg *msg, jsOptions *opts, jsErrCode *errCode);
+// NATS_EXTERN natsStatus natsMsg_Nak(natsMsg *msg, jsOptions *opts);
+// NATS_EXTERN natsStatus natsMsg_NakWithDelay(natsMsg *msg, int64_t delay, jsOptions *opts);
+// NATS_EXTERN natsStatus natsMsg_InProgress(natsMsg *msg, jsOptions *opts);
+// NATS_EXTERN natsStatus natsMsg_Term(natsMsg *msg, jsOptions *opts);
+// NATS_EXTERN uint64_t natsMsg_GetSequence(natsMsg *msg);
+// NATS_EXTERN int64_t natsMsg_GetTime(natsMsg *msg);
+
+// TODO: not implementing streaming API right now
+// NATS_EXTERN uint64_t stanMsg_GetSequence(const stanMsg *msg);
+// NATS_EXTERN int64_t stanMsg_GetTimestamp(const stanMsg *msg);
+// NATS_EXTERN bool stanMsg_IsRedelivered(const stanMsg *msg);
+// NATS_EXTERN const char* stanMsg_GetData(const stanMsg *msg);
+// NATS_EXTERN int stanMsg_GetDataLength(const stanMsg *msg);
+// NATS_EXTERN void stanMsg_Destroy(stanMsg *msg);

src/nats.zig

@@ -14,238 +14,130 @@
 
 const std = @import("std");
 
-pub const nats_c = @cImport({
-    @cInclude("nats/nats.h");
-});
+pub const nats_c = @import("./nats_c.zig").nats_c;
 
-fn onMessage(
-    conn: ?*nats_c.natsConnection,
-    sub: ?*nats_c.natsSubscription,
-    message: ?*nats_c.natsMsg,
-    userdata: ?*anyopaque,
-) callconv(.C) void {
-    _ = sub;
-    defer nats_c.natsMsg_Destroy(message);
+const err_ = @import("./error.zig");
+const con_ = @import("./connection.zig");
+const sub_ = @import("./subscription.zig");
+const msg_ = @import("./message.zig");
+const sta_ = @import("./statistics.zig");
 
-    const msgData = nats_c.natsMsg_GetData(message)[0..@intCast(nats_c.natsMsg_GetDataLength(message))];
-    std.debug.print("Received message: {s} - {s}\n", .{ nats_c.natsMsg_GetSubject(message), msgData });
+pub const default_server_url = con_.default_server_url;
+pub const Connection = con_.Connection;
+pub const ConnectionOptions = con_.ConnectionOptions;
+pub const JwtResponseOrError = con_.JwtResponseOrError;
+pub const SignatureResponseOrError = con_.SignatureResponseOrError;
 
-    if (@as(?[*]const u8, nats_c.natsMsg_GetReply(message))) |reply| {
-        _ = nats_c.natsConnection_PublishString(conn, reply, "salutations");
-    }
+pub const Subscription = sub_.Subscription;
 
-    if (@as(?*bool, @ptrCast(userdata))) |signal| {
-        signal.* = true;
-    }
+pub const Message = msg_.Message;
+
+pub const Statistics = sta_.Statistics;
+pub const StatsCounts = sta_.StatsCounts;
+
+pub const ErrorInfo = err_.ErrorInfo;
+pub const getLastError = err_.getLastError;
+pub const getLastErrorStack = err_.getLastErrorStack;
+pub const Status = err_.Status;
+pub const Error = err_.Error;
+
+pub fn getVersion() [:0]const u8 {
+    const verString = nats_c.nats_GetVersion();
+    return std.mem.sliceTo(verString, 0);
 }
 
-pub fn main() void {
-    var conn: ?*nats_c.natsConnection = null;
-    defer nats_c.natsConnection_Destroy(conn);
-
-    if (nats_c.natsConnection_ConnectTo(&conn, nats_c.NATS_DEFAULT_URL) != nats_c.NATS_OK) {
-        std.debug.print("oh no {s}\n", .{nats_c.NATS_DEFAULT_URL});
-        return;
-    }
-
-    var sub: ?*nats_c.natsSubscription = null;
-    defer nats_c.natsSubscription_Destroy(sub);
-    var done = false;
-    if (nats_c.natsConnection_Subscribe(&sub, conn, "channel", onMessage, &done) != nats_c.NATS_OK) {
-        std.debug.print("whops\n", .{});
-        return;
-    }
-
-    while (!done) {
-        var reply: ?*nats_c.natsMsg = null;
-        defer nats_c.natsMsg_Destroy(reply);
-
-        if (nats_c.natsConnection_RequestString(&reply, conn, "channel", "whatsup", 1000) != nats_c.NATS_OK) {
-            std.debug.print("geez\n", .{});
-            return;
-        } else if (reply) |message| {
-            const msgData = nats_c.natsMsg_GetData(message)[0..@intCast(nats_c.natsMsg_GetDataLength(message))];
-            std.debug.print("Got reply: {s}\n", .{msgData});
-        }
-    }
+pub fn getVersionNumber() u32 {
+    return nats_c.nats_GetVersionNumber();
 }
 
-// NATS_EXTERN natsStatus nats_Open(int64_t lockSpinCount);
-// NATS_EXTERN const char* nats_GetVersion(void);
-// NATS_EXTERN uint32_t nats_GetVersionNumber(void);
+pub fn checkCompatibility() bool {
+    return nats_c.nats_CheckCompatibilityImpl(
+        nats_c.NATS_VERSION_REQUIRED_NUMBER,
+        nats_c.NATS_VERSION_NUMBER,
+        nats_c.NATS_VERSION_STRING,
+    );
+}
 
-// #define nats_CheckCompatibility() nats_CheckCompatibilityImpl(NATS_VERSION_REQUIRED_NUMBER, NATS_VERSION_NUMBER, NATS_VERSION_STRING)
-// NATS_EXTERN bool nats_CheckCompatibilityImpl(uint32_t reqVerNumber, uint32_t verNumber, const char *verString);
+pub fn now() i64 {
+    return nats_c.nats_Now();
+}
 
-// NATS_EXTERN int64_t nats_Now(void);
-// NATS_EXTERN int64_t nats_NowInNanoSeconds(void);
-// NATS_EXTERN void nats_Sleep(int64_t sleepTime);
-// NATS_EXTERN const char* nats_GetLastError(natsStatus *status);
-// NATS_EXTERN natsStatus nats_GetLastErrorStack(char *buffer, size_t bufLen);
-// NATS_EXTERN void nats_PrintLastErrorStack(FILE *file);
-// NATS_EXTERN natsStatus nats_SetMessageDeliveryPoolSize(int max);
-// NATS_EXTERN void nats_ReleaseThreadMemory(void);
+pub fn nowInNanoseconds() i64 {
+    return nats_c.nats_NowInNanoSeconds();
+}
 
-// NATS_EXTERN natsStatus nats_Sign(const char *encodedSeed, const char *input, unsigned char **signature, int *signatureLength);
+pub fn sleep(sleep_time: i64) void {
+    return nats_c.nats_Sleep(sleep_time);
+}
 
-// NATS_EXTERN void nats_Close(void);
-// NATS_EXTERN natsStatus nats_CloseAndWait(int64_t timeout);
+pub fn setMessageDeliveryPoolSize(max: c_int) Error!void {
+    const status = Status.fromInt(nats_c.nats_SetMessageDeliveryPoolSize(max));
+    return status.raise();
+}
 
-// NATS_EXTERN const char* natsStatus_GetText(natsStatus s);
+pub fn releaseThreadMemory() void {
+    return nats_c.nats_ReleaseThreadMemory();
+}
 
-// NATS_EXTERN natsStatus natsStatistics_Create(natsStatistics **newStats);
-// NATS_EXTERN natsStatus natsStatistics_GetCounts(const natsStatistics *stats, uint64_t *inMsgs, uint64_t *inBytes, uint64_t *outMsgs, uint64_t *outBytes, uint64_t *reconnects);
-// NATS_EXTERN void natsStatistics_Destroy(natsStatistics *stats);
+pub const default_spin_count: i64 = -1;
 
-// NATS_EXTERN natsStatus natsOptions_Create(natsOptions **newOpts);
-// NATS_EXTERN natsStatus natsOptions_SetURL(natsOptions *opts, const char *url);
-// NATS_EXTERN natsStatus natsOptions_SetServers(natsOptions *opts, const char** servers, int serversCount);
-// NATS_EXTERN natsStatus natsOptions_SetUserInfo(natsOptions *opts, const char *user, const char *password);
-// NATS_EXTERN natsStatus natsOptions_SetToken(natsOptions *opts, const char *token);
-// NATS_EXTERN natsStatus natsOptions_SetTokenHandler(natsOptions *opts, natsTokenHandler tokenCb, void *closure);
-// NATS_EXTERN natsStatus natsOptions_SetNoRandomize(natsOptions *opts, bool noRandomize);
-// NATS_EXTERN natsStatus natsOptions_SetTimeout(natsOptions *opts, int64_t timeout);
-// NATS_EXTERN natsStatus natsOptions_SetName(natsOptions *opts, const char *name);
-// NATS_EXTERN natsStatus natsOptions_SetSecure(natsOptions *opts, bool secure);
-// NATS_EXTERN natsStatus natsOptions_LoadCATrustedCertificates(natsOptions *opts, const char *fileName);
-// NATS_EXTERN natsStatus natsOptions_SetCATrustedCertificates(natsOptions *opts, const char *certificates);
-// NATS_EXTERN natsStatus natsOptions_LoadCertificatesChain(natsOptions *opts, const char *certsFileName, const char *keyFileName);
-// NATS_EXTERN natsStatus natsOptions_SetCertificatesChain(natsOptions *opts, const char *cert, const char *key);
-// NATS_EXTERN natsStatus natsOptions_SetCiphers(natsOptions *opts, const char *ciphers);
-// NATS_EXTERN natsStatus natsOptions_SetCipherSuites(natsOptions *opts, const char *ciphers);
-// NATS_EXTERN natsStatus natsOptions_SetExpectedHostname(natsOptions *opts, const char *hostname);
-// NATS_EXTERN natsStatus natsOptions_SkipServerVerification(natsOptions *opts, bool skip);
-// NATS_EXTERN natsStatus natsOptions_SetVerbose(natsOptions *opts, bool verbose);
-// NATS_EXTERN natsStatus natsOptions_SetPedantic(natsOptions *opts, bool pedantic);
-// NATS_EXTERN natsStatus natsOptions_SetPingInterval(natsOptions *opts, int64_t interval);
-// NATS_EXTERN natsStatus natsOptions_SetMaxPingsOut(natsOptions *opts, int maxPingsOut);
-// NATS_EXTERN natsStatus natsOptions_SetIOBufSize(natsOptions *opts, int ioBufSize);
-// NATS_EXTERN natsStatus natsOptions_SetAllowReconnect(natsOptions *opts, bool allow);
-// NATS_EXTERN natsStatus natsOptions_SetMaxReconnect(natsOptions *opts, int maxReconnect);
-// NATS_EXTERN natsStatus natsOptions_SetReconnectWait(natsOptions *opts, int64_t reconnectWait);
-// NATS_EXTERN natsStatus natsOptions_SetReconnectJitter(natsOptions *opts, int64_t jitter, int64_t jitterTLS);
-// NATS_EXTERN natsStatus natsOptions_SetCustomReconnectDelay(natsOptions *opts, natsCustomReconnectDelayHandler cb, void *closure);
-// NATS_EXTERN natsStatus natsOptions_SetReconnectBufSize(natsOptions *opts, int reconnectBufSize);
-// NATS_EXTERN natsStatus natsOptions_SetMaxPendingMsgs(natsOptions *opts, int maxPending);
-// NATS_EXTERN natsStatus natsOptions_SetErrorHandler(natsOptions *opts, natsErrHandler errHandler, void *closure);
-// NATS_EXTERN natsStatus natsOptions_SetClosedCB(natsOptions *opts, natsConnectionHandler closedCb, void *closure);
-// NATS_EXTERN natsStatus natsOptions_SetDisconnectedCB(natsOptions *opts, natsConnectionHandler disconnectedCb, void *closure);
-// NATS_EXTERN natsStatus natsOptions_SetReconnectedCB(natsOptions *opts, natsConnectionHandler reconnectedCb, void *closure);
-// NATS_EXTERN natsStatus natsOptions_SetDiscoveredServersCB(natsOptions *opts, natsConnectionHandler discoveredServersCb, void *closure);
-// NATS_EXTERN natsStatus natsOptions_SetIgnoreDiscoveredServers(natsOptions *opts, bool ignore);
-// NATS_EXTERN natsStatus natsOptions_SetLameDuckModeCB(natsOptions *opts, natsConnectionHandler lameDuckCb, void *closure);
-// NATS_EXTERN natsStatus natsOptions_SetEventLoop(natsOptions *opts, void *loop, natsEvLoop_Attach attachCb, natsEvLoop_ReadAddRemove readCb, natsEvLoop_WriteAddRemove writeCb, natsEvLoop_Detach detachCb);
-// NATS_EXTERN natsStatus natsOptions_UseGlobalMessageDelivery(natsOptions *opts, bool global);
-// NATS_EXTERN natsStatus natsOptions_IPResolutionOrder(natsOptions *opts, int order);
-// NATS_EXTERN natsStatus natsOptions_SetSendAsap(natsOptions *opts, bool sendAsap);
-// NATS_EXTERN natsStatus natsOptions_UseOldRequestStyle(natsOptions *opts, bool useOldStyle);
-// NATS_EXTERN natsStatus natsOptions_SetFailRequestsOnDisconnect(natsOptions *opts, bool failRequests);
-// NATS_EXTERN natsStatus natsOptions_SetNoEcho(natsOptions *opts, bool noEcho);
-// NATS_EXTERN natsStatus natsOptions_SetRetryOnFailedConnect(natsOptions *opts, bool retry, natsConnectionHandler connectedCb, void* closure);
-// NATS_EXTERN natsStatus natsOptions_SetUserCredentialsCallbacks(natsOptions *opts, natsUserJWTHandler ujwtCB, void *ujwtClosure, natsSignatureHandler sigCB, void *sigClosure);
-// NATS_EXTERN natsStatus natsOptions_SetUserCredentialsFromFiles(natsOptions *opts, const char *userOrChainedFile, const char *seedFile);
-// NATS_EXTERN natsStatus natsOptions_SetUserCredentialsFromMemory(natsOptions *opts, const char *jwtAndSeedContent);
-// NATS_EXTERN natsStatus natsOptions_SetNKey(natsOptions             *opts, const char              *pubKey, natsSignatureHandler    sigCB, void                    *sigClosure);
-// NATS_EXTERN natsStatus natsOptions_SetNKeyFromSeed(natsOptions *opts, const char  *pubKey, const char  *seedFile);
-// NATS_EXTERN natsStatus natsOptions_SetWriteDeadline(natsOptions *opts, int64_t deadline);
-// NATS_EXTERN natsStatus natsOptions_DisableNoResponders(natsOptions *opts, bool disabled);
-// NATS_EXTERN natsStatus natsOptions_SetCustomInboxPrefix(natsOptions *opts, const char *inboxPrefix);
-// NATS_EXTERN natsStatus natsOptions_SetMessageBufferPadding(natsOptions *opts, int paddingSize);
-// NATS_EXTERN void natsOptions_Destroy(natsOptions *opts);
+pub fn init(lock_spin_count: i64) Error!void {
+    const status = Status.fromInt(nats_c.nats_Open(lock_spin_count));
+    return status.raise();
+}
 
-// NATS_EXTERN natsStatus natsInbox_Create(natsInbox **newInbox);
-// NATS_EXTERN void natsInbox_Destroy(natsInbox *inbox);
-// NATS_EXTERN void natsMsgList_Destroy(natsMsgList *list);
+pub fn deinit() void {
+    return nats_c.nats_Close();
+}
 
-// NATS_EXTERN natsStatus natsMsg_Create(natsMsg **newMsg, const char *subj, const char *reply, const char *data, int dataLen);
-// NATS_EXTERN const char* natsMsg_GetSubject(const natsMsg *msg);
-// NATS_EXTERN const char* natsMsg_GetReply(const natsMsg *msg);
-// NATS_EXTERN const char* natsMsg_GetData(const natsMsg *msg);
-// NATS_EXTERN int natsMsg_GetDataLength(const natsMsg *msg);
-// NATS_EXTERN natsStatus natsMsgHeader_Set(natsMsg *msg, const char *key, const char *value);
-// NATS_EXTERN natsStatus natsMsgHeader_Add(natsMsg *msg, const char *key, const char *value);
-// NATS_EXTERN natsStatus natsMsgHeader_Get(natsMsg *msg, const char *key, const char **value);
-// NATS_EXTERN natsStatus natsMsgHeader_Values(natsMsg *msg, const char *key, const char* **values, int *count);
-// NATS_EXTERN natsStatus natsMsgHeader_Keys(natsMsg *msg, const char* **keys, int *count);
-// NATS_EXTERN natsStatus natsMsgHeader_Delete(natsMsg *msg, const char *key);
-// NATS_EXTERN bool natsMsg_IsNoResponders(natsMsg *msg);
-// NATS_EXTERN void natsMsg_Destroy(natsMsg *msg);
-// NATS_EXTERN uint64_t stanMsg_GetSequence(const stanMsg *msg);
-// NATS_EXTERN int64_t stanMsg_GetTimestamp(const stanMsg *msg);
-// NATS_EXTERN bool stanMsg_IsRedelivered(const stanMsg *msg);
-// NATS_EXTERN const char* stanMsg_GetData(const stanMsg *msg);
-// NATS_EXTERN int stanMsg_GetDataLength(const stanMsg *msg);
-// NATS_EXTERN void stanMsg_Destroy(stanMsg *msg);
+pub fn deinitWait(timeout: i64) Error!void {
+    const status = Status.fromInt(nats_c.nats_CloseAndWait(timeout));
+    return status.raise();
+}
 
-// NATS_EXTERN natsStatus natsConnection_Connect(natsConnection **nc, natsOptions *options);
-// NATS_EXTERN void natsConnection_ProcessReadEvent(natsConnection *nc);
-// NATS_EXTERN void natsConnection_ProcessWriteEvent(natsConnection *nc);
-// NATS_EXTERN natsStatus natsConnection_ConnectTo(natsConnection **nc, const char *urls);
-// NATS_EXTERN bool natsConnection_IsClosed(natsConnection *nc);
-// NATS_EXTERN bool natsConnection_IsReconnecting(natsConnection *nc);
-// NATS_EXTERN natsConnStatus natsConnection_Status(natsConnection *nc);
-// NATS_EXTERN int natsConnection_Buffered(natsConnection *nc);
-// NATS_EXTERN natsStatus natsConnection_Flush(natsConnection *nc);
-// NATS_EXTERN natsStatus natsConnection_FlushTimeout(natsConnection *nc, int64_t timeout);
-// NATS_EXTERN int64_t natsConnection_GetMaxPayload(natsConnection *nc);
-// NATS_EXTERN natsStatus natsConnection_GetStats(natsConnection *nc, natsStatistics *stats);
-// NATS_EXTERN natsStatus natsConnection_GetConnectedUrl(natsConnection *nc, char *buffer, size_t bufferSize);
-// NATS_EXTERN natsStatus natsConnection_GetConnectedServerId(natsConnection *nc, char *buffer, size_t bufferSize);
-// NATS_EXTERN natsStatus natsConnection_GetServers(natsConnection *nc, char ***servers, int *count);
-// NATS_EXTERN natsStatus natsConnection_GetDiscoveredServers(natsConnection *nc, char ***servers, int *count);
-// NATS_EXTERN natsStatus natsConnection_GetLastError(natsConnection *nc, const char **lastError);
-// NATS_EXTERN natsStatus natsConnection_GetClientID(natsConnection *nc, uint64_t *cid);
-// NATS_EXTERN natsStatus natsConnection_Drain(natsConnection *nc);
-// NATS_EXTERN natsStatus natsConnection_DrainTimeout(natsConnection *nc, int64_t timeout);
-// NATS_EXTERN natsStatus natsConnection_Sign(natsConnection *nc, const unsigned char *message, int messageLen, unsigned char sig[64]);
-// NATS_EXTERN natsStatus natsConnection_GetClientIP(natsConnection *nc, char **ip);
-// NATS_EXTERN natsStatus natsConnection_GetRTT(natsConnection *nc, int64_t *rtt);
-// NATS_EXTERN natsStatus natsConnection_HasHeaderSupport(natsConnection *nc);
-// NATS_EXTERN void natsConnection_Close(natsConnection *nc);
-// NATS_EXTERN void natsConnection_Destroy(natsConnection *nc);
-// NATS_EXTERN natsStatus natsConnection_Publish(natsConnection *nc, const char *subj, const void *data, int dataLen);
-// NATS_EXTERN natsStatus natsConnection_PublishString(natsConnection *nc, const char *subj, const char *str);
-// NATS_EXTERN natsStatus natsConnection_PublishMsg(natsConnection *nc, natsMsg *msg);
-// NATS_EXTERN natsStatus natsConnection_PublishRequest(natsConnection *nc, const char *subj, const char *reply, const void *data, int dataLen);
-// NATS_EXTERN natsStatus natsConnection_PublishRequestString(natsConnection *nc, const char *subj, const char *reply, const char *str);
-// NATS_EXTERN natsStatus natsConnection_Request(natsMsg **replyMsg, natsConnection *nc, const char *subj, const void *data, int dataLen, int64_t timeout);
-// NATS_EXTERN natsStatus natsConnection_RequestString(natsMsg **replyMsg, natsConnection *nc, const char *subj, const char *str, int64_t timeout);
-// NATS_EXTERN natsStatus natsConnection_RequestMsg(natsMsg **replyMsg, natsConnection *nc,natsMsg *requestMsg, int64_t timeout);
-// NATS_EXTERN natsStatus natsConnection_Subscribe(natsSubscription **sub, natsConnection *nc, const char *subject, natsMsgHandler cb, void *cbClosure);
-// NATS_EXTERN natsStatus natsConnection_SubscribeTimeout(natsSubscription **sub, natsConnection *nc, const char *subject, int64_t timeout, natsMsgHandler cb, void *cbClosure);
-// NATS_EXTERN natsStatus natsConnection_SubscribeSync(natsSubscription **sub, natsConnection *nc, const char *subject);
-// NATS_EXTERN natsStatus natsConnection_QueueSubscribe(natsSubscription **sub, natsConnection *nc, const char *subject, const char *queueGroup, natsMsgHandler cb, void *cbClosure);
-// NATS_EXTERN natsStatus natsConnection_QueueSubscribeTimeout(natsSubscription **sub, natsConnection *nc, const char *subject, const char *queueGroup, int64_t timeout, natsMsgHandler cb, void *cbClosure);
-// NATS_EXTERN natsStatus natsConnection_QueueSubscribeSync(natsSubscription **sub, natsConnection *nc, const char *subject, const char *queueGroup);
+// the result of this requires manual deallocation unless it is used to provide the
+// signature out-parameter in the natsSignatureHandler callback. Calling it outside of
+// that context seems unlikely, but we should probably provide a deinit function so the
+// user doesn't have to dig around for libc free to deallocate it.
+pub fn sign(encoded_seed: [:0]const u8, input: [:0]const u8) Error![]const u8 {
+    var result: [*c]u8 = undefined;
+    var length: c_int = 0;
+    const status = Status.fromInt(nats_c.nats_Sign(
+        encoded_seed.ptr,
+        input.ptr,
+        &result,
+        &length,
+    ));
 
-// NATS_EXTERN natsStatus natsSubscription_NoDeliveryDelay(natsSubscription *sub);
-// NATS_EXTERN natsStatus natsSubscription_NextMsg(natsMsg **nextMsg, natsSubscription *sub, int64_t timeout);
-// NATS_EXTERN natsStatus natsSubscription_Unsubscribe(natsSubscription *sub);
-// NATS_EXTERN natsStatus natsSubscription_AutoUnsubscribe(natsSubscription *sub, int max);
-// NATS_EXTERN natsStatus natsSubscription_QueuedMsgs(natsSubscription *sub, uint64_t *queuedMsgs);
-// NATS_EXTERN int64_t natsSubscription_GetID(natsSubscription* sub);
-// NATS_EXTERN const char* natsSubscription_GetSubject(natsSubscription* sub);
-// NATS_EXTERN natsStatus natsSubscription_SetPendingLimits(natsSubscription *sub, int msgLimit, int bytesLimit);
-// NATS_EXTERN natsStatus natsSubscription_GetPendingLimits(natsSubscription *sub, int *msgLimit, int *bytesLimit);
-// NATS_EXTERN natsStatus natsSubscription_GetPending(natsSubscription *sub, int *msgs, int *bytes);
-// NATS_EXTERN natsStatus natsSubscription_GetDelivered(natsSubscription *sub, int64_t *msgs);
-// NATS_EXTERN natsStatus natsSubscription_GetDropped(natsSubscription *sub, int64_t *msgs);
-// NATS_EXTERN natsStatus natsSubscription_GetMaxPending(natsSubscription *sub, int *msgs, int *bytes);
-// NATS_EXTERN natsStatus natsSubscription_ClearMaxPending(natsSubscription *sub);
-// NATS_EXTERN natsStatus natsSubscription_GetStats(natsSubscription *sub, int *pendingMsgs, int *pendingBytes, int *maxPendingMsgs, int *maxPendingBytes, int64_t *deliveredMsgs, int64_t *droppedMsgs);
-// NATS_EXTERN bool natsSubscription_IsValid(natsSubscription *sub);
-// NATS_EXTERN natsStatus natsSubscription_Drain(natsSubscription *sub);
-// NATS_EXTERN natsStatus natsSubscription_DrainTimeout(natsSubscription *sub, int64_t timeout);
-// NATS_EXTERN natsStatus natsSubscription_WaitForDrainCompletion(natsSubscription *sub, int64_t timeout);
-// NATS_EXTERN natsStatus natsSubscription_DrainCompletionStatus(natsSubscription *sub);
-// NATS_EXTERN natsStatus natsSubscription_SetOnCompleteCB(natsSubscription *sub, natsOnCompleteCB cb, void *closure);
-// NATS_EXTERN void natsSubscription_Destroy(natsSubscription *sub);
+    return status.toError() orelse result[0..@intCast(length)];
+}
 
-// NATS_EXTERN natsStatus natsMsg_Ack(natsMsg *msg, jsOptions *opts);
-// NATS_EXTERN natsStatus natsMsg_AckSync(natsMsg *msg, jsOptions *opts, jsErrCode *errCode);
-// NATS_EXTERN natsStatus natsMsg_Nak(natsMsg *msg, jsOptions *opts);
-// NATS_EXTERN natsStatus natsMsg_NakWithDelay(natsMsg *msg, int64_t delay, jsOptions *opts);
-// NATS_EXTERN natsStatus natsMsg_InProgress(natsMsg *msg, jsOptions *opts);
-// NATS_EXTERN natsStatus natsMsg_Term(natsMsg *msg, jsOptions *opts);
-// NATS_EXTERN uint64_t natsMsg_GetSequence(natsMsg *msg);
-// NATS_EXTERN int64_t natsMsg_GetTime(natsMsg *msg);
+// Note: an "Inbox" is actually just a string. This API creates a random (unique)
+// string suitable for passing as the `reply` field to Message.create or
+// Connection.publishRequest. The string is owned by the caller and should be freed
+// using `destroyInbox`.
+pub fn createInbox() Error![:0]u8 {
+    var self: [*c]u8 = undefined;
+    const status = Status.fromInt(nats_c.natsInbox_Create(@ptrCast(&self)));
+
+    return status.toError() orelse std.mem.sliceTo(self, 0);
+}
+
+pub fn destroyInbox(inbox: [:0]u8) void {
+    nats_c.natsInbox_Destroy(@ptrCast(inbox.ptr));
+}
+
+// I think this is also a jetstream API. This function sure does not seem at all useful
+// by itself. Note: for some reason, most of the jetstream data structures are all
+// public, instead of following the opaque handle style that the rest of the library
+// does.
+
+// typedef struct natsMsgList {
+//         natsMsg         **Msgs;
+//         int             Count;
+// } natsMsgList;
+pub const MessageList = opaque {
+    pub fn destroy(self: *MessageList) void {
+        nats_c.natsMsgList_Destroy(@ptrCast(self));
+    }
+};

src/nats_c.zig

@@ -0,0 +1,3 @@
+pub const nats_c = @cImport({
+    @cInclude("nats/nats.h");
+});

src/statistics.zig

@@ -0,0 +1,54 @@
+// Copyright 2023 torque@epicyclic.dev
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+const std = @import("std");
+
+const nats_c = @import("./nats_c.zig").nats_c;
+
+const err_ = @import("./error.zig");
+const Status = err_.Status;
+const Error = err_.Error;
+
+pub const StatsCounts = struct {
+    messages_in: u64 = 0,
+    bytes_in: u64 = 0,
+    messages_out: u64 = 0,
+    bytes_out: u64 = 0,
+    reconnects: u64 = 0,
+};
+
+pub const Statistics = opaque {
+    pub fn create() Error!*Statistics {
+        var stats: *Statistics = undefined;
+        const status = Status.fromInt(nats_c.natsStatistics_Create(@ptrCast(&stats)));
+        return status.toError() orelse stats;
+    }
+
+    pub fn destroy(self: *Statistics) void {
+        nats_c.natsStatistics_Destroy(@ptrCast(self));
+    }
+
+    pub fn getCounts(self: *Statistics) Error!StatsCounts {
+        var counts: StatsCounts = .{};
+        const status = Status.fromInt(nats_c.natsStatistics_GetCounts(
+            @ptrCast(self),
+            &counts.messages_in,
+            &counts.bytes_in,
+            &counts.messages_out,
+            &counts.bytes_out,
+            &counts.reconnects,
+        ));
+        return status.toError() orelse counts;
+    }
+};

src/subscription.zig

@@ -0,0 +1,216 @@
+// Copyright 2023 torque@epicyclic.dev
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+const std = @import("std");
+
+const nats_c = @import("./nats_c.zig").nats_c;
+
+const Connection = @import("./connection.zig").Connection;
+
+const Message = @import("./message.zig").Message;
+
+const err_ = @import("./error.zig");
+const Error = err_.Error;
+const Status = err_.Status;
+
+const thunkhelper = @import("./thunk.zig");
+
+pub const Subscription = opaque {
+    pub const MessageCount = struct {
+        messages: c_int = 0,
+        bytes: c_int = 0,
+    };
+
+    pub const SubscriptionStats = struct {
+        pending: MessageCount = .{},
+        max_pending: MessageCount = .{},
+        delivered_messages: i64 = 0,
+        dropped_messages: i64 = 0,
+    };
+
+    pub fn isValid(self: *Subscription) bool {
+        return nats_c.natsSubscription_IsValid(@ptrCast(self));
+    }
+
+    pub fn destroy(self: *Subscription) void {
+        nats_c.natsSubscription_Destroy(@ptrCast(self));
+    }
+
+    pub fn unsubscribe(self: *Subscription) Error!void {
+        return Status.fromInt(nats_c.natsSubscription_Unsubscribe(@ptrCast(self))).raise();
+    }
+
+    pub fn autoUnsubscribe(self: *Subscription, max: c_int) Error!void {
+        return Status.fromInt(nats_c.natsSubscription_AutoUnsubscribe(@ptrCast(self), max)).raise();
+    }
+
+    pub fn nextMessage(self: *Subscription, timeout: i64) Error!*Message {
+        var message: *Message = undefined;
+        const status = Status.fromInt(nats_c.natsSubscription_NextMsg(
+            @ptrCast(&message),
+            @ptrCast(self),
+            timeout,
+        ));
+
+        return status.toError() orelse message;
+    }
+
+    pub fn queuedMessageCount(self: *Subscription) Error!u64 {
+        var count: u64 = 0;
+        const status = Status.fromInt(nats_c.natsSubscription_QueuedMsgs(@ptrCast(self), &count));
+        return status.toError() orelse count;
+    }
+
+    pub fn getId(self: *Subscription) i64 {
+        // TODO: invalid/closed subscriptions return 0. Should we convert that into an
+        // error? could return error.InvalidSubscription
+        return nats_c.natsSubscription_GetID(@ptrCast(self));
+    }
+
+    pub fn getSubject(self: *Subscription) ?[:0]const u8 {
+        // invalid/closed subscriptions return null. should we convert that into an
+        // error? could return error.InvalidSubscription
+        const result = nats_c.natsSubscription_GetSubject(@ptrCast(self)) orelse return null;
+        return std.mem.sliceTo(result, 0);
+    }
+
+    pub fn setPendingLimits(self: *Subscription, limit: MessageCount) Error!void {
+        return Status.fromInt(
+            nats_c.natsSubscription_SetPendingLimits(@ptrCast(self), limit.messages, limit.bytes),
+        ).raise();
+    }
+
+    pub fn getPendingLimits(self: *Subscription) Error!MessageCount {
+        var result: MessageCount = .{};
+        const status = Status.fromInt(
+            nats_c.natsSubscription_GetPendingLimits(@ptrCast(self), &result.messages, &result.bytes),
+        );
+
+        return status.toError() orelse result;
+    }
+
+    pub fn getPending(self: *Subscription) Error!MessageCount {
+        var result: MessageCount = .{};
+        const status = Status.fromInt(
+            nats_c.natsSubscription_GetPending(@ptrCast(self), &result.messages, &result.bytes),
+        );
+
+        return status.toError() orelse result;
+    }
+
+    pub fn getMaxPending(self: *Subscription) Error!MessageCount {
+        var result: MessageCount = .{};
+        const status = Status.fromInt(
+            nats_c.natsSubscription_GetMaxPending(@ptrCast(self), &result.messages, &result.bytes),
+        );
+
+        return status.toError() orelse result;
+    }
+
+    pub fn clearMaxPending(self: *Subscription) Error!void {
+        return Status.fromInt(nats_c.natsSubscription_ClearMaxPending(@ptrCast(self))).raise();
+    }
+
+    pub fn getDelivered(self: *Subscription) Error!i64 {
+        var result: i64 = 0;
+        const status = Status.fromInt(nats_c.natsSubscription_GetDelivered(@ptrCast(self), &result));
+
+        return status.toError() orelse result;
+    }
+
+    pub fn getDropped(self: *Subscription) Error!i64 {
+        var result: i64 = 0;
+        const status = Status.fromInt(nats_c.natsSubscription_GetDropped(@ptrCast(self), &result));
+
+        return status.toError() orelse result;
+    }
+
+    pub fn getStats(self: *Subscription) Error!SubscriptionStats {
+        var result: SubscriptionStats = .{};
+        const status = Status.fromInt(nats_c.natsSubscription_GetStats(
+            @ptrCast(self),
+            &result.pending.messages,
+            &result.pending.bytes,
+            &result.max_pending.messages,
+            &result.max_pending.bytes,
+            &result.delivered_messages,
+            &result.dropped_messages,
+        ));
+
+        return status.toError() orelse result;
+    }
+
+    pub fn drain(self: *Subscription) Error!void {
+        return Status.fromInt(nats_c.natsSubscription_Drain(@ptrCast(self))).raise();
+    }
+
+    pub fn drainTimeout(self: *Subscription, timeout: i64) Error!void {
+        return Status.fromInt(nats_c.natsSubscription_DrainTimeout(@ptrCast(self), timeout)).raise();
+    }
+
+    pub fn waitForDrainCompletion(self: *Subscription, timeout: i64) Error!void {
+        return Status.fromInt(nats_c.natsSubscription_WaitForDrainCompletion(@ptrCast(self), timeout)).raise();
+    }
+
+    pub fn drainCompletionStatus(self: *Subscription) ?Error {
+        return Status.fromInt(nats_c.natsSubscription_DrainCompletionStatus(@ptrCast(self))).toError();
+    }
+
+    pub fn setCompletionCallback(
+        self: *Subscription,
+        comptime T: type,
+        comptime callback: *const thunkhelper.SimpleCallbackThunkSignature(T),
+        userdata: T,
+    ) Error!void {
+        return Status.fromInt(nats_c.natsSubscription_SetOnCompleteCB(
+            @ptrCast(self),
+            thunkhelper.makeSimpleCallbackThunk(T, callback),
+            thunkhelper.opaqueFromUserdata(userdata),
+        )).raise();
+    }
+};
+
+const SubscriptionCallback = fn (
+    ?*nats_c.natsConnection,
+    ?*nats_c.natsSubscription,
+    ?*nats_c.natsMsg,
+    ?*anyopaque,
+) callconv(.C) void;
+
+pub fn SubscriptionCallbackSignature(comptime UDT: type) type {
+    return fn (UDT, *Connection, *Subscription, *Message) void;
+}
+
+pub fn makeSubscriptionCallbackThunk(
+    comptime UDT: type,
+    comptime callback: *const SubscriptionCallbackSignature(UDT),
+) *const SubscriptionCallback {
+    return struct {
+        fn thunk(
+            conn: ?*nats_c.natsConnection,
+            sub: ?*nats_c.natsSubscription,
+            msg: ?*nats_c.natsMsg,
+            userdata: ?*anyopaque,
+        ) callconv(.C) void {
+            const message: *Message = if (msg) |m| @ptrCast(m) else unreachable;
+            defer message.destroy();
+
+            const connection: *Connection = if (conn) |c| @ptrCast(c) else unreachable;
+            const subscription: *Subscription = if (sub) |s| @ptrCast(s) else unreachable;
+
+            const data = thunkhelper.userdataFromOpaque(UDT, userdata);
+            callback(data, connection, subscription, message);
+        }
+    }.thunk;
+}

src/thunk.zig

@@ -0,0 +1,86 @@
+// Copyright 2023 torque@epicyclic.dev
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+const std = @import("std");
+
+const nats_c = @import("./nats_c.zig").nats_c;
+
+const optional = if (@hasField(std.builtin.Type, "optional")) .optional else .Optional;
+const pointer = if (@hasField(std.builtin.Type, "pointer")) .pointer else .Pointer;
+const void_type = if (@hasField(std.builtin.Type, "void")) .void else .Void;
+const null_type = if (@hasField(std.builtin.Type, "null")) .null else .Null;
+
+pub fn opaqueFromUserdata(userdata: anytype) ?*anyopaque {
+    checkUserDataType(@TypeOf(userdata));
+    return switch (@typeInfo(@TypeOf(userdata))) {
+        optional, pointer => @constCast(@ptrCast(userdata)),
+        void_type => null,
+        else => @compileError("Unsupported userdata type " ++ @typeName(@TypeOf(userdata))),
+    };
+}
+
+pub fn userdataFromOpaque(comptime UDT: type, userdata: ?*anyopaque) UDT {
+    comptime checkUserDataType(UDT);
+    return if (UDT == void)
+        void{}
+    else if (@typeInfo(UDT) == optional)
+        @alignCast(@ptrCast(userdata))
+    else
+        @alignCast(@ptrCast(userdata.?));
+}
+
+pub fn checkUserDataType(comptime T: type) void {
+    switch (@typeInfo(T)) {
+        optional => |info| switch (@typeInfo(info.child)) {
+            optional => @compileError(
+                "nats callbacks can only accept void or an (optional) single, many," ++
+                    " or c pointer as userdata. \"" ++
+                    @typeName(T) ++ "\" has more than one optional specifier.",
+            ),
+            else => checkUserDataType(info.child),
+        },
+        pointer => |info| switch (info.size) {
+            .slice => @compileError(
+                "nats callbacks can only accept void or an (optional) single, many," ++
+                    " or c pointer as userdata, not slices. \"" ++
+                    @typeName(T) ++ "\" appears to be a slice.",
+            ),
+            else => {},
+        },
+        void_type => {},
+        else => @compileError(
+            "nats callbacks can only accept void or an (optional) single, many," ++
+                " or c pointer as userdata. \"" ++
+                @typeName(T) ++ "\" is not a pointer type.",
+        ),
+    }
+}
+
+const SimpleCallback = fn (?*anyopaque) callconv(.C) void;
+
+pub fn SimpleCallbackThunkSignature(comptime UDT: type) type {
+    return fn (UDT) void;
+}
+
+pub fn makeSimpleCallbackThunk(
+    comptime UDT: type,
+    comptime callback: *const SimpleCallbackThunkSignature(UDT),
+) *const SimpleCallback {
+    comptime checkUserDataType(UDT);
+    return struct {
+        fn thunk(userdata: ?*anyopaque) callconv(.C) void {
+            callback(userdataFromOpaque(UDT, userdata));
+        }
+    }.thunk;
+}

src/win-crosshack.c

@@ -1,44 +0,0 @@
-// These functions are taken from MinGW-w64.
-// In theory, they have been released into the public domain.
-
-#include <winsock2.h>
-#include <ws2tcpip.h>
-
-#define WIN32_LEAN_AND_MEAN
-#include <windows.h>
-#undef WIN32_LEAN_AND_MEAN
-
-// this is supposed to be inlined but apparently there is a flag causing it to not be
-// inlined? zig mingw may be too old
-PVOID WINAPI RtlSecureZeroMemory(PVOID ptr,SIZE_T cnt)
-{
-  volatile char *vptr = (volatile char *)ptr;
-#ifdef __x86_64
-  __stosb ((PBYTE)((DWORD64)vptr),0,cnt);
-#else
-  while (cnt != 0)
-    {
-      *vptr++ = 0;
-      cnt--;
-    }
-#endif /* __x86_64 */
-  return ptr;
-}
-
-// zig doesn't compile the parts of mingw that contain this for some reason
-WCHAR *gai_strerrorW(int ecode)
-{
-    DWORD dwMsgLen __attribute__((unused));
-    static WCHAR buff[GAI_STRERROR_BUFFER_SIZE + 1];
-    dwMsgLen = FormatMessageW(FORMAT_MESSAGE_FROM_SYSTEM|FORMAT_MESSAGE_IGNORE_INSERTS|FORMAT_MESSAGE_MAX_WIDTH_MASK,
-                  NULL, ecode, MAKELANGID(LANG_NEUTRAL,SUBLANG_DEFAULT), (LPWSTR)buff,
-                  GAI_STRERROR_BUFFER_SIZE, NULL);
-    return buff;
-}
-
-char *gai_strerrorA(int ecode)
-{
-    static char buff[GAI_STRERROR_BUFFER_SIZE + 1];
-    wcstombs(buff, gai_strerrorW(ecode), GAI_STRERROR_BUFFER_SIZE + 1);
-    return buff;
-}

tests/connection.zig

@@ -0,0 +1,271 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+const std = @import("std");
+
+const nats = @import("nats");
+
+const util = @import("./util.zig");
+
+const rsa_key = @embedFile("./data/client-rsa.key");
+const rsa_cert = @embedFile("./data/client-rsa.cert");
+const ecc_key = @embedFile("./data/client-ecc.key");
+const ecc_cert = @embedFile("./data/client-ecc.cert");
+
+test "nats.Connection.connectTo" {
+    {
+        var server = try util.TestServer.launch(.{});
+        defer server.stop();
+
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+
+        const connection = try nats.Connection.connectTo(server.url);
+        defer connection.destroy();
+    }
+
+    {
+        var server = try util.TestServer.launch(.{
+            .auth = .{ .token = "test_token" },
+        });
+        defer server.stop();
+
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+
+        const connection = try nats.Connection.connectTo(server.url);
+        defer connection.destroy();
+    }
+
+    {
+        var server = try util.TestServer.launch(.{ .auth = .{
+            .password = .{ .user = "user", .pass = "password" },
+        } });
+        defer server.stop();
+
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+
+        const connection = try nats.Connection.connectTo(server.url);
+        defer connection.destroy();
+        connection.close();
+    }
+}
+
+test "nats.Connection" {
+    var server = try util.TestServer.launch(.{});
+    defer server.stop();
+
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const connection = try nats.Connection.connectTo(server.url);
+    defer connection.destroy();
+
+    _ = connection.isClosed();
+    _ = connection.isReconnecting();
+    _ = connection.getStatus();
+    _ = connection.bytesBuffered();
+    try connection.flush();
+    try connection.flushTimeout(100);
+    _ = connection.getMaxPayload();
+    _ = try connection.getStats();
+    {
+        // id is 56 bytes plus terminating zero
+        var buf = [_]u8{0} ** 57;
+        _ = try connection.getConnectedUrl(&buf);
+        _ = try connection.getConnectedServerId(&buf);
+    }
+    {
+        var servers = try connection.getServers();
+        defer servers.deinit();
+
+        var discovered = try connection.getDiscoveredServers();
+        defer discovered.deinit();
+    }
+
+    _ = connection.getLastError();
+    _ = try connection.getClientId();
+    // our connection does not have a JWT, so this call will always fail
+    _ = connection.sign("greetings") catch {};
+    _ = try connection.getLocalIpAndPort();
+    _ = connection.getRtt() catch {};
+    _ = connection.hasHeaderSupport();
+    // this closes the connection, but it does not block until the connection is closed,
+    // which can result in nondeterministic behavior for calls after this one.
+    try connection.drain();
+    // this will return error.ConnectionClosed if the connection is already closed, so
+    // don't expect this to be error free.
+    connection.drainTimeout(1000) catch {};
+}
+
+fn callbacks(comptime UDT: type) type {
+    return struct {
+        fn reconnectDelayHandler(userdata: UDT, connection: *nats.Connection, attempts: c_int) i64 {
+            _ = userdata;
+            _ = connection;
+            _ = attempts;
+
+            return 0;
+        }
+
+        fn errorHandler(
+            userdata: UDT,
+            connection: *nats.Connection,
+            subscription: *nats.Subscription,
+            status: nats.Status,
+        ) void {
+            _ = userdata;
+            _ = connection;
+            _ = subscription;
+            _ = status;
+        }
+
+        fn connectionHandler(userdata: UDT, connection: *nats.Connection) void {
+            _ = userdata;
+            _ = connection;
+        }
+
+        fn jwtHandler(userdata: UDT) nats.JwtResponseOrError {
+            _ = userdata;
+            // return .{ .jwt = std.heap.raw_c_allocator.dupeZ(u8, "abcdef") catch @panic("no!") };
+            return .{ .error_message = std.heap.raw_c_allocator.dupeZ(u8, "dang") catch @panic("no!") };
+        }
+
+        fn signatureHandler(userdata: UDT, nonce: [:0]const u8) nats.SignatureResponseOrError {
+            _ = userdata;
+            _ = nonce;
+            // return .{ .signature = std.heap.raw_c_allocator.dupe(u8, "01230123") catch @panic("no!") };
+            return .{ .error_message = std.heap.raw_c_allocator.dupeZ(u8, "whoops") catch @panic("no!") };
+        }
+    };
+}
+
+test "nats.ConnectionOptions" {
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const options = try nats.ConnectionOptions.create();
+    defer options.destroy();
+
+    const userdata: u32 = 0;
+
+    try options.setUrl(nats.default_server_url);
+    const servers = [_][*:0]const u8{ "nats://127.0.0.1:4442", "nats://127.0.0.1:4443" };
+    try options.setServers(&servers);
+    try options.setCredentials("user", "password");
+    try options.setToken("test_token");
+    try options.setNoRandomize(false);
+    try options.setTimeout(1000);
+    try options.setName("name");
+
+    try options.setVerbose(true);
+    try options.setPedantic(true);
+    try options.setPingInterval(1000);
+    try options.setMaxPingsOut(100);
+    try options.setIoBufSize(1024);
+    try options.setAllowReconnect(false);
+    try options.setMaxReconnect(10);
+    try options.setReconnectWait(500);
+    try options.setReconnectJitter(100, 200);
+    try options.setCustomReconnectDelay(*const u32, callbacks(*const u32).reconnectDelayHandler, &userdata);
+    try options.setCustomReconnectDelay(void, callbacks(void).reconnectDelayHandler, {});
+    try options.setCustomReconnectDelay(?*const u32, callbacks(?*const u32).reconnectDelayHandler, null);
+    try options.setReconnectBufSize(1024);
+    try options.setMaxPendingMessages(50);
+    try options.setErrorHandler(*const u32, callbacks(*const u32).errorHandler, &userdata);
+    try options.setErrorHandler(void, callbacks(void).errorHandler, {});
+    try options.setErrorHandler(?*const u32, callbacks(?*const u32).errorHandler, null);
+    try options.setClosedCallback(*const u32, callbacks(*const u32).connectionHandler, &userdata);
+    try options.setClosedCallback(void, callbacks(void).connectionHandler, {});
+    try options.setClosedCallback(?*const u32, callbacks(?*const u32).connectionHandler, null);
+    try options.setDisconnectedCallback(*const u32, callbacks(*const u32).connectionHandler, &userdata);
+    try options.setDisconnectedCallback(void, callbacks(void).connectionHandler, {});
+    try options.setDisconnectedCallback(?*const u32, callbacks(?*const u32).connectionHandler, null);
+    try options.setDiscoveredServersCallback(*const u32, callbacks(*const u32).connectionHandler, &userdata);
+    try options.setDiscoveredServersCallback(void, callbacks(void).connectionHandler, {});
+    try options.setDiscoveredServersCallback(?*const u32, callbacks(?*const u32).connectionHandler, null);
+    try options.setLameDuckModeCallback(*const u32, callbacks(*const u32).connectionHandler, &userdata);
+    try options.setLameDuckModeCallback(void, callbacks(void).connectionHandler, {});
+    try options.setLameDuckModeCallback(?*const u32, callbacks(?*const u32).connectionHandler, null);
+    try options.ignoreDiscoveredServers(true);
+    try options.useGlobalMessageDelivery(false);
+    try options.ipResolutionOrder(.ipv4_first);
+    try options.setSendAsap(true);
+    try options.useOldRequestStyle(false);
+    try options.setFailRequestsOnDisconnect(true);
+    try options.setNoEcho(true);
+    try options.setRetryOnFailedConnect(*const u32, callbacks(*const u32).connectionHandler, true, &userdata);
+    try options.setRetryOnFailedConnect(void, callbacks(void).connectionHandler, true, {});
+    try options.setRetryOnFailedConnect(?*const u32, callbacks(?*const u32).connectionHandler, true, null);
+    try options.setUserCredentialsCallbacks(*const u32, *const u32, callbacks(*const u32).jwtHandler, callbacks(*const u32).signatureHandler, &userdata, &userdata);
+    try options.setUserCredentialsCallbacks(void, void, callbacks(void).jwtHandler, callbacks(void).signatureHandler, {}, {});
+    try options.setWriteDeadline(5);
+    try options.disableNoResponders(true);
+    try options.setCustomInboxPrefix("_FOOBOX");
+    try options.setMessageBufferPadding(123);
+}
+
+fn tokenHandler(userdata: *u32) [:0]const u8 {
+    _ = userdata;
+    return "token";
+}
+
+test "nats.ConnectionOptions (crypto edition)" {
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const options = try nats.ConnectionOptions.create();
+    defer options.destroy();
+    var userdata: u32 = 0;
+
+    try options.setTokenHandler(*u32, tokenHandler, &userdata);
+    try options.setSecure(false);
+    try options.setCertificatesChain(rsa_cert, rsa_key);
+    try options.setCiphers("-ALL:HIGH");
+    try options.setCipherSuites("TLS_AES_256_GCM_SHA384:TLS_CHACHA20_POLY1305_SHA256:TLS_AES_128_GCM_SHA256");
+    try options.setExpectedHostname("test.nats.zig");
+    try options.skipServerVerification(true);
+}
+
+test "nats.ConnectionOptions (crypto connect)" {
+    {
+        var server = try util.TestServer.launch(.{ .tls = .rsa });
+        defer server.stop();
+
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+
+        const options = try nats.ConnectionOptions.create();
+        defer options.destroy();
+
+        try options.setSecure(true);
+        try options.skipServerVerification(true);
+        try options.setCertificatesChain(rsa_cert, rsa_key);
+
+        const connection = try nats.Connection.connect(options);
+        defer connection.destroy();
+
+        try connection.publish("foo", "bar");
+    }
+
+    {
+        var server = try util.TestServer.launch(.{ .tls = .ecc });
+        defer server.stop();
+
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+
+        const options = try nats.ConnectionOptions.create();
+        defer options.destroy();
+
+        try options.setSecure(true);
+        try options.skipServerVerification(true);
+        try options.setCertificatesChain(ecc_cert, ecc_key);
+
+        const connection = try nats.Connection.connect(options);
+        defer connection.destroy();
+
+        try connection.publish("foo", "bar");
+    }
+}

tests/data/client-ecc.cert

@@ -0,0 +1,16 @@
+-----BEGIN CERTIFICATE-----
+MIIClTCCAhygAwIBAgIUNUxHIA2G0sEDgjWcx1svHk9wK1AwCgYIKoZIzj0EAwIw
+gYAxCzAJBgNVBAYTAlhYMRIwEAYDVQQIDAlTdGF0ZU5hbWUxETAPBgNVBAcMCENp
+dHlOYW1lMRQwEgYDVQQKDAtDb21wYW55TmFtZTEbMBkGA1UECwwSQ29tcGFueVNl
+Y3Rpb25OYW1lMRcwFQYDVQQDDA50ZXN0cy5uYXRzLnppZzAgFw0yMzA5MDIyMjEz
+MzJaGA8yMTIzMDgwOTIyMTMzMlowgYAxCzAJBgNVBAYTAlhYMRIwEAYDVQQIDAlT
+dGF0ZU5hbWUxETAPBgNVBAcMCENpdHlOYW1lMRQwEgYDVQQKDAtDb21wYW55TmFt
+ZTEbMBkGA1UECwwSQ29tcGFueVNlY3Rpb25OYW1lMRcwFQYDVQQDDA50ZXN0cy5u
+YXRzLnppZzB2MBAGByqGSM49AgEGBSuBBAAiA2IABPcAhGdzooeoL0SY1qJmY3SY
+cyngBaon8ICyaQLRvOKEIhym25jB90xj4J2IDy3pg/5564G43NOOBZ/T4ClTv3XF
++2E71w+31HVGVKW4l+natAQt72oXn+HcjSnwrplYz6NTMFEwHQYDVR0OBBYEFFkN
+/08ID2jBHkYMUzm7o8S8Ch86MB8GA1UdIwQYMBaAFFkN/08ID2jBHkYMUzm7o8S8
+Ch86MA8GA1UdEwEB/wQFMAMBAf8wCgYIKoZIzj0EAwIDZwAwZAIwWauKBQMCf6XT
+K1y2TvNZpqehk+MsQ8aDcNoUJ+iALJM7Y89XpibdZ4hvGPsQK/cgAjAg8SXHVWS5
+yeFNtZdnRowEuQhtk5rRaj983wMSLbMXbz9Oxm0MY7edcS4MWP7l7nI=
+-----END CERTIFICATE-----

tests/data/client-ecc.key

@@ -0,0 +1,6 @@
+-----BEGIN PRIVATE KEY-----
+MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDDfkvPMjrVTAU06KSLK
+vxKDF3+vdWDQUxQBIJ3F1qrSMJNzfuguXiYv8DMTbNvCNB+hZANiAAT3AIRnc6KH
+qC9EmNaiZmN0mHMp4AWqJ/CAsmkC0bzihCIcptuYwfdMY+CdiA8t6YP+eeuBuNzT
+jgWf0+ApU791xfthO9cPt9R1RlSluJfp2rQELe9qF5/h3I0p8K6ZWM8=
+-----END PRIVATE KEY-----

tests/data/client-rsa.cert

@@ -0,0 +1,34 @@
+-----BEGIN CERTIFICATE-----
+MIIF5TCCA82gAwIBAgIUSiVHqM6yI2CBF8ZuakGuaXdAd0EwDQYJKoZIhvcNAQEL
+BQAwgYAxCzAJBgNVBAYTAlhYMRIwEAYDVQQIDAlTdGF0ZU5hbWUxETAPBgNVBAcM
+CENpdHlOYW1lMRQwEgYDVQQKDAtDb21wYW55TmFtZTEbMBkGA1UECwwSQ29tcGFu
+eVNlY3Rpb25OYW1lMRcwFQYDVQQDDA50ZXN0cy5uYXRzLnppZzAgFw0yMzA5MDIy
+MjE0MDZaGA8yMTIzMDgwOTIyMTQwNlowgYAxCzAJBgNVBAYTAlhYMRIwEAYDVQQI
+DAlTdGF0ZU5hbWUxETAPBgNVBAcMCENpdHlOYW1lMRQwEgYDVQQKDAtDb21wYW55
+TmFtZTEbMBkGA1UECwwSQ29tcGFueVNlY3Rpb25OYW1lMRcwFQYDVQQDDA50ZXN0
+cy5uYXRzLnppZzCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAOF2MLuH
+xvEUbc2PXQKhTL7flDvl/KWGxikE4HBQ96vQ3bHAzVuLezTI9D0qLTOvY1SVBy0C
+En4e5ZklufTHciOg/a1B76IC2EFPEnhY0oO7pLfLzA43MU7biNvb1BGPW/MEihOZ
+2ZmGnK1CiwynFKTkRJdafzK8OrQV9nmz+WNX0UqbDH6Gjp36V6VMcz+gHi01hqK+
+v1lbAuUATGNEoZSVpGSoJx4+Ogyob9f4irsJX7GpO0dsKHUwd51pAm1pkpOH1rIU
+CM+biSYaLt5fxFkSUJc9wqa+odE997hu5Ch6EEkdwg/4vYnigBMZxwFB3EVinXI0
+5ywrYMwO8ozCrKmIDVG/FhNlzspooKVkZFKl8/T/1n8rDElsLc8No7clD0xFQxU8
+3qyw05yaYm3khDMWTaD/B2w3AOoifZ9oWr6Ra2NN57yQx4sQdDy8s0WY4rVkv/nX
+9VsBSeJwmmCfnnpse1Tq4Wfo7Jjz1Akl4n0GAQxkxR6gAkCfnkJrdkMKaOPQe+ep
+Fs+5g2VrIeWQgzs5L4rte1qtMokK963IvqDSnsbPM6m9tcMQb8r+A0YbEOAK6D0D
+nnKcguSKNkPP8hwaqJ9XNDXSN65I+DsgX0ngfHCh/grkSPJI2oEmx4DeaA8hgRzh
+zbGpazOcVa4ahyUCxZKZi9c/+Aw3gir0fpQbAgMBAAGjUzBRMB0GA1UdDgQWBBRI
+3sPebMsmYX0ourw+0SA8BqcXwTAfBgNVHSMEGDAWgBRI3sPebMsmYX0ourw+0SA8
+BqcXwTAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3DQEBCwUAA4ICAQCkGzTMZdGT
+dUQBbAqTzHvoc1Vy5OxadgJ4UGly9s7GpYg8brsevFmHyGHl+J9dBsnhmmNGxs1t
+rU5lvnMWLOyvg7iLyx/4rkNAR03wyRgx1jK+4xIxMBj7YDB0WK6AtL3OUiqvTW4h
+INoEIylyrIIOf7oHS9UkzDdZ3ZghH+EADjE6ksM/PuCRK0oTz9z7LUSMlcI5UCeb
+P/V6N2QHzz2n+/BvTo/bpjHrwyUZ+eZbUC7rCfZX7Pi/pTGS2bps16lO2Us6tzIn
+9If5h+MHHGOZXQtOZPYDFi3nLo/E+bBzt6Lvrt0Vm5DjXKzplrq+UMPbsEtGue7v
+xSfcbPsXf44n+kZSFtFTjZ4zqpP05/H3R5eTayRUhVLkJlxDZbPiFjOSEupUmZq+
+cpmf6Mq9OJJimyJBRaC0Lex4mCDUarXYLXXyNaIAQ2O+EmaNxZ2B86gHy6R3gaHr
+w66pDXq0IXznJyr50aKu7WYJHOE5HOaBgptU7lXz35+5yId7WKWorXfGIYBWCc3g
+LRDr0pgyLk1YVxz2tphR8hEJ7rJb+gMR3t5UzDElsUIqKJo/6Aziq4dRlrj5x62s
+Vmm7JJqMSyh6UDJowwFicS3KTZbKZnxRHdyy8Km6gRvi26pdNWW5+/57ETLG3cn5
++MDARAttyn8SEhbPolxttebRKDAIFnxYkQ==
+-----END CERTIFICATE-----

tests/data/client-rsa.key

@@ -0,0 +1,52 @@
+-----BEGIN PRIVATE KEY-----
+MIIJQgIBADANBgkqhkiG9w0BAQEFAASCCSwwggkoAgEAAoICAQDhdjC7h8bxFG3N
+j10CoUy+35Q75fylhsYpBOBwUPer0N2xwM1bi3s0yPQ9Ki0zr2NUlQctAhJ+HuWZ
+Jbn0x3IjoP2tQe+iAthBTxJ4WNKDu6S3y8wONzFO24jb29QRj1vzBIoTmdmZhpyt
+QosMpxSk5ESXWn8yvDq0FfZ5s/ljV9FKmwx+ho6d+lelTHM/oB4tNYaivr9ZWwLl
+AExjRKGUlaRkqCcePjoMqG/X+Iq7CV+xqTtHbCh1MHedaQJtaZKTh9ayFAjPm4km
+Gi7eX8RZElCXPcKmvqHRPfe4buQoehBJHcIP+L2J4oATGccBQdxFYp1yNOcsK2DM
+DvKMwqypiA1RvxYTZc7KaKClZGRSpfP0/9Z/KwxJbC3PDaO3JQ9MRUMVPN6ssNOc
+mmJt5IQzFk2g/wdsNwDqIn2faFq+kWtjTee8kMeLEHQ8vLNFmOK1ZL/51/VbAUni
+cJpgn556bHtU6uFn6OyY89QJJeJ9BgEMZMUeoAJAn55Ca3ZDCmjj0HvnqRbPuYNl
+ayHlkIM7OS+K7XtarTKJCvetyL6g0p7GzzOpvbXDEG/K/gNGGxDgCug9A55ynILk
+ijZDz/IcGqifVzQ10jeuSPg7IF9J4Hxwof4K5EjySNqBJseA3mgPIYEc4c2xqWsz
+nFWuGoclAsWSmYvXP/gMN4Iq9H6UGwIDAQABAoICADToqwP/F3kQrbndCFsrJhru
+1db+oDzp9Uu/+LlyzsRTxgrGL4rpnxaih+pooOXtpTY+qMnvoA5XytKXL13ZhhgF
+WjKT9BvFZiFhYHi8g15lpQB6w16cpiYWz7Wkj041ocLUUGDMLGviUpc4M/BarzYI
+2W3ZT1tFH9OOCeLCkOY2wActfo+cnRBGpNXGLI+EUECUvI0pjTb3bCT4XnS3MOHx
+AfybF176BF5fEqwQh+Hfj8Td7WrT32Ss5I0cjPTHHx4e9QuiNvUdT2CRKWmG+Mlc
+SmxLkofV2ZyEWcM+xq0XBAZchOOBoF0guaSB2pkZbwsbWs5nys4rOdJ5OYM91g2b
+/QeKUHMIXav2rJn375I2I7vsQhriH8nz0h4pTQj+bsBwKW2AAauaNg6Yma4Cm0b+
+OI+boERqqglZRyiUrbr1WJvVp39UL8fEWaCZBR5OmB7TyCratf39nc/FZ/JhvuPU
+Q3wjEdGK5NYDNAq1PXAHBAqmGX85qtJ5ZhgTnmQ+HTrcJ5M8mzP7404SUSgkFglt
+udl83w8JYGW6w7D6QjSz80zIh5/uy6A6oO6UEJIpfoH7f8lB1oJPYxLRezVyjqLt
+Gd9HfFH2nApx/OvYDn16PEmagiL7AAaqhcooWwPLe61p2ehNxG7Safc2t0R/StCK
+gyYBuZbBJ4pwypWJ5fhhAoIBAQDxLC/ntfU2/eyXFnSnDPMwCRY2ClTGyn99Ofa0
+D6lf+2sNTgfXhH81sDuTQ0MuNL6MwfZWBzq58M+eM0qLYpOYGb1v8itKuT3iFYnR
+9eA3CUahDQlTI9JfPWk4nmxuyiBmg1FBroZMF8ddRT57NFAiXBA6dgsuBegby/kr
+RaICjz1i9uFXefDXPyiZkEw5XfVFQVBtAVEa2QawjKZZP50vPMrYZC9BPok5xfRp
+Jh/9rti9BBmjFL8CEaxUfI1wIZ9syOV1FAoSf+UBqfocP3vaPpJhD8VxwkBwmzTi
+BQySv+/iwSUfAdB4a0l1W/dN3LxhXUubWEPLWp7u7mb6mIm7AoIBAQDvUrqqpAjD
+XZUUgPMiRVXC1l7Iw9lKb9iBNzW7k13NVowaY4BGsljlAEseoyW/PGT5Rx+InnM0
+zA5xif6kWk7Csgwqhe8576QkwCSJc48vFwoCujYbhXoXYMj2vQdpwtKZ2/F5ixeq
+W5uDso5a8iJ2KmeRd5gXPebfR3iajyDG122o/sWdVZWqvZWNzy4KDIEj1xIuxXW4
+AuJSAU0IBkKDnV0AiHG7Yobci4Gwd86dzwg+n2vQ/7/OYvey/a+FLIU9R1Ax4Y8/
+ri5MwEDzLGvpbqJKn24NR5JzTzmF9zFydTrkk18xgH2UKSB68bQ6C+3qB1VaEy0B
+rDCdAuK4g8khAoIBAB4aGdCeEYFPqFwjXWQMZb41JCSSnYpCdC85MOXAnq9wPihm
++OuZihc1a/oxhw0ZYD9JZmnOdTIIMKHaXQ+QukNd0xtJ6sVk4ah6b71ZJyc3bS1k
+5ykNa5CfpaZ/f6FEcU7aTSYZloGg5i1qGyZdnTLsssnZOgQAkLwHdY5FHrebEVps
+3iuA+OKk63hfXmQ6qgZ+5H72jxz+war/ozO4kPH4cIkZ2BwYpiAj6SHGtG+Bh2Pw
+QxLr3/tuIUhaU30PdUqquJkoaylr9TWD9cfY1Kik7rhWs5pDWK+1b6BWaP9YHaT7
+3ppEK7UcDwsq828wggLVFj7JgYy8PuIrt4bHy0kCggEAKaQAXK5749pFlTK2mzDr
+MiJwjYgeJ6h8SEdd7ww+FvtHF1RWvnZLp1S8vVDvwW11uDXa07+WFgqnPLQg/WHF
+MHUgTsnNDQyYR9iywsO7lxrwH/dccL9xtd2eOeg8APfoAuNVCavc60RTM7/+qu5U
+drD8IkBn0ytvH0xlPKdIsbBMIUprAewhRXsFKY5x2UfBtIW4YTD0QZcm39PgHlRQ
+gGwCAZS8DTmgc4FGiHjgF28tZRACB3RoYDWyGY+wWYCckkP1PSic7xyUa8BLzMPe
+5tfcHxXMZT0dyzhurtOK4/pny9ukhY1wzDW3tAyYKj1nIQAzpp+Nhiv6rWcSIb60
+YQKCAQEAzZI9SNeue35lxcGK5pYrZKO8L/3n4LHQ+1enua1kWwLC3xRqtdVrCBg8
+bkkRNpKHmQfelzJgCAQGlCoQj/KwnYhYVxY1Mkd6BWXw0WS9NBAWnt3c4mVms8iY
+z3tfEgJshspQ8kOUuJBdbnhcvu+as2TfqpJHNnUMLrG2iTBG9yvbCJWWb7qBj1q7
+5e89Veg2avP6TzvuKwJP0cVuHsyQ2WndFaPkxK+uylRFmcKMCASPtuM3L4Ynsoc8
+7Dswp9suPXpTmCv1LHueRAYvi2ZgWmlZbzpTSQe/9vVHiKHU1Q1k7mTVGrKz/hg4
+qnLykaAJSPCsd4jY4P3EtZLNbee3fQ==
+-----END PRIVATE KEY-----

tests/data/genkeys.fish

@@ -0,0 +1,4 @@
+openssl req -x509 -newkey rsa:4096 -sha256 -days 36500 -nodes -keyout client-rsa.key -out client-rsa.cert -subj "/C=XX/ST=StateName/L=CityName/O=CompanyName/OU=CompanySectionName/CN=tests.nats.zig"
+openssl req -x509 -newkey rsa:4096 -sha256 -days 36500 -nodes -keyout server-rsa.key -out server-rsa.cert -subj "/C=XX/ST=StateName/L=CityName/O=CompanyName/OU=CompanySectionName/CN=tests.nats.zig"
+openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:secp384r1 -days 36500 -nodes -keyout client-ecc.key -out client-ecc.cert -subj "/C=XX/ST=StateName/L=CityName/O=CompanyName/OU=CompanySectionName/CN=tests.nats.zig"
+openssl req -x509 -newkey ec -pkeyopt ec_paramgen_curve:secp384r1 -days 36500 -nodes -keyout server-ecc.key -out server-ecc.cert -subj "/C=XX/ST=StateName/L=CityName/O=CompanyName/OU=CompanySectionName/CN=tests.nats.zig"

tests/data/server-ecc.cert

@@ -0,0 +1,16 @@
+-----BEGIN CERTIFICATE-----
+MIICljCCAhygAwIBAgIUcjxQ8y9DFk5UChiI2CTG4fKNuzswCgYIKoZIzj0EAwIw
+gYAxCzAJBgNVBAYTAlhYMRIwEAYDVQQIDAlTdGF0ZU5hbWUxETAPBgNVBAcMCENp
+dHlOYW1lMRQwEgYDVQQKDAtDb21wYW55TmFtZTEbMBkGA1UECwwSQ29tcGFueVNl
+Y3Rpb25OYW1lMRcwFQYDVQQDDA50ZXN0cy5uYXRzLnppZzAgFw0yMzA5MDIyMjEy
+NDJaGA8yMTIzMDgwOTIyMTI0MlowgYAxCzAJBgNVBAYTAlhYMRIwEAYDVQQIDAlT
+dGF0ZU5hbWUxETAPBgNVBAcMCENpdHlOYW1lMRQwEgYDVQQKDAtDb21wYW55TmFt
+ZTEbMBkGA1UECwwSQ29tcGFueVNlY3Rpb25OYW1lMRcwFQYDVQQDDA50ZXN0cy5u
+YXRzLnppZzB2MBAGByqGSM49AgEGBSuBBAAiA2IABL1QQVL1CUI+g8EAFtW4OYRk
+bsa4GpLMojjWYGZ5azB6zFJsjqr626hwl8B+yUOLV8O8COZtSHhh9TOwMK9my+mF
+jSlRLkBhViJtiQS/i+lOAHLSZhyYGi0LMwR/84s9zqNTMFEwHQYDVR0OBBYEFDep
+AJQvuHNpqaxaBt5Md4mAcMcBMB8GA1UdIwQYMBaAFDepAJQvuHNpqaxaBt5Md4mA
+cMcBMA8GA1UdEwEB/wQFMAMBAf8wCgYIKoZIzj0EAwIDaAAwZQIwWiMIHiOwfO0+
+CcjrMZmSVqhBiDY8bcJgpPFCp+GFfPiTwse1eUQhYE0K2onU1mYTAjEA7atUdQxL
+8SjkuTRdvNoRJ2EOVMHMeaYMU5HPAxHhIPDWBaHmHOHgOx5hUw/chRjf
+-----END CERTIFICATE-----

tests/data/server-ecc.key

@@ -0,0 +1,6 @@
+-----BEGIN PRIVATE KEY-----
+MIG2AgEAMBAGByqGSM49AgEGBSuBBAAiBIGeMIGbAgEBBDAdXFWFvY4ujDsDxw2e
+Vga7jpgVBKAg7mEA/AYL+yjLanR52DJ6XM8iU4qbPEaMH66hZANiAAS9UEFS9QlC
+PoPBABbVuDmEZG7GuBqSzKI41mBmeWswesxSbI6q+tuocJfAfslDi1fDvAjmbUh4
+YfUzsDCvZsvphY0pUS5AYVYibYkEv4vpTgBy0mYcmBotCzMEf/OLPc4=
+-----END PRIVATE KEY-----

tests/data/server-rsa.cert

@@ -0,0 +1,34 @@
+-----BEGIN CERTIFICATE-----
+MIIF5TCCA82gAwIBAgIUUIWzfooDSLVR8sW3AiBJuqDHZMYwDQYJKoZIhvcNAQEL
+BQAwgYAxCzAJBgNVBAYTAlhYMRIwEAYDVQQIDAlTdGF0ZU5hbWUxETAPBgNVBAcM
+CENpdHlOYW1lMRQwEgYDVQQKDAtDb21wYW55TmFtZTEbMBkGA1UECwwSQ29tcGFu
+eVNlY3Rpb25OYW1lMRcwFQYDVQQDDA50ZXN0cy5uYXRzLnppZzAgFw0yMzA5MDIy
+MjA5MTdaGA8yMTIzMDgwOTIyMDkxN1owgYAxCzAJBgNVBAYTAlhYMRIwEAYDVQQI
+DAlTdGF0ZU5hbWUxETAPBgNVBAcMCENpdHlOYW1lMRQwEgYDVQQKDAtDb21wYW55
+TmFtZTEbMBkGA1UECwwSQ29tcGFueVNlY3Rpb25OYW1lMRcwFQYDVQQDDA50ZXN0
+cy5uYXRzLnppZzCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAOolz+6A
+j1S3fJl8VSglQBaOplTsukup0Li7pDluS+Sgj5G5Vvl1Ki8jPVTWz9HcYl/7jgFz
+iE7AG7qymH1+q/smzxl59MyiKkBPSvynRQ/iY35uyyCFSxgqNSFqnahzHmd1BpNX
+9WJX2iRs+UyaWjhtj7T2cbIr2UZQtCUqF/ae0jeR+hb6beQCMoGzFG3Zn9uupazg
+u49WdNmOSE/qf6DAmftnmdX7pc7u6jclMAk8WOvjL9LoxK8Nu+ZGusklJohjbUsG
+DNdktsaqZQphlXbVCa6gV6cITup2oFmHQvMcncZ2xtYs/4Ul45UxGGEWGWQLQtaJ
+KSS5B5YAkXTCga8k/cS+NRyy/NxQmuR3FeQUYYfMpiNhXweZ7zEWjosU9jjEiS6w
+UhlTCQYH6p9/jZmByE8DySNQtcexV1IBVzOYsg7MS/AwHq9ym9iqT9tx+S9Vub/f
+ysbgQU9GkKeic64iYVC+gG/Y2W8XS9LSzSKCJuvf8EUD18g3dmmRxzltwq7oH14W
+f0KjyhwdzeJUhoEvPgaEgWvlvff9uSP0utX/KmOJPJckjoQrk9RN/GDxR01U2SPf
+1pVclir3IEWSPOCNSil8Et/ut1MI7JGlDhsysZA3zoqIPAu+QYVUEytdSf+n3UHf
+ZvT7VdDQlNyq2XYaP9kBEx4P7oGfto/HHOHtAgMBAAGjUzBRMB0GA1UdDgQWBBQV
+GBaOUXiHmG3KT2uiwgOR7xBDSDAfBgNVHSMEGDAWgBQVGBaOUXiHmG3KT2uiwgOR
+7xBDSDAPBgNVHRMBAf8EBTADAQH/MA0GCSqGSIb3DQEBCwUAA4ICAQCZhqnD3uVi
+3q2plOHUBnlkYcCCoC0W7AHPSJ1pcb/wa1CgSkrXHNhBmaWfWEbs+JJ+eMHxStFr
+kdJWNbVEnmS27ruMNE6S5XCj/1zdqncZnCr3gLRdmvb3S5yZHGphgEdNrAvPKIyi
+XtfmROVh3y2/ZENRZfdJ984xU/yaHmfTy8ULiSCe0Jz9dxHvTgUCrZ5RVEIwmAkL
+gcs1Hrw/5TDs/DagsNliH2Atn6cYzhAostBN9wDkra6xUkRqscE8R7D2/XJni+W3
+toL+DTTFIww0pNlh2BWG5DMqeLdYPODhuxWWw5JhdKWx3cisPhiJ9RDKaSy532wC
+i/tjKQAnNWQRvGFGfJDMIEcPL3s1JhQM/lhSanlloMsxHmGpOJOEI6IN0clYpPaX
+7COo+qvGueQPRLDq/7vDRM3i7sQHBHeC71VdEN4TgyJQvQcGMFjGPPQ868VvJhYG
+2CbUWhBaADWlWjZ5d5tMI+WT8J/y6D6cSOymXcgbHV8DpUCB9vImmTKvgiL9TLi1
+PxZCP9nLUOCmCSnbz4vhupvxM38KesuvAfOUPKTEixjXBfD9iiEKVnMUKVv31TLO
+5sfeYf6Zn8oDFCInUgdHgQRs/UXOAOTWYsOQQhS6xfZXJHom/Mcy0Q2TxsiI2dmw
+ah/FB4DsF0XDCuDqGJWyywTbzQnh55uBBQ==
+-----END CERTIFICATE-----

tests/data/server-rsa.key

@@ -0,0 +1,52 @@
+-----BEGIN PRIVATE KEY-----
+MIIJQgIBADANBgkqhkiG9w0BAQEFAASCCSwwggkoAgEAAoICAQDqJc/ugI9Ut3yZ
+fFUoJUAWjqZU7LpLqdC4u6Q5bkvkoI+RuVb5dSovIz1U1s/R3GJf+44Bc4hOwBu6
+sph9fqv7Js8ZefTMoipAT0r8p0UP4mN+bssghUsYKjUhap2ocx5ndQaTV/ViV9ok
+bPlMmlo4bY+09nGyK9lGULQlKhf2ntI3kfoW+m3kAjKBsxRt2Z/brqWs4LuPVnTZ
+jkhP6n+gwJn7Z5nV+6XO7uo3JTAJPFjr4y/S6MSvDbvmRrrJJSaIY21LBgzXZLbG
+qmUKYZV21QmuoFenCE7qdqBZh0LzHJ3GdsbWLP+FJeOVMRhhFhlkC0LWiSkkuQeW
+AJF0woGvJP3EvjUcsvzcUJrkdxXkFGGHzKYjYV8Hme8xFo6LFPY4xIkusFIZUwkG
+B+qff42ZgchPA8kjULXHsVdSAVczmLIOzEvwMB6vcpvYqk/bcfkvVbm/38rG4EFP
+RpCnonOuImFQvoBv2NlvF0vS0s0igibr3/BFA9fIN3Zpkcc5bcKu6B9eFn9Co8oc
+Hc3iVIaBLz4GhIFr5b33/bkj9LrV/ypjiTyXJI6EK5PUTfxg8UdNVNkj39aVXJYq
+9yBFkjzgjUopfBLf7rdTCOyRpQ4bMrGQN86KiDwLvkGFVBMrXUn/p91B32b0+1XQ
+0JTcqtl2Gj/ZARMeD+6Bn7aPxxzh7QIDAQABAoICAAeIfOaa+0GJ/7e+cMzwWd3/
+6+kKjrnVdlIjM1bnrghmhAf3sw0mkFtg4l3C5X/Ge+HDqZ9xVJ7X/mxkx5QuCaF0
+b7BNpKsawoo8Itj7FrU6nuHX9bAPqclWvkvbbsQXJBDHCpWd/FaUJgALA4BL7QAo
+wjlbvm+xinWBLjKN5qR4GqJQD4BCwVtXGMHkfZFFMafzOABWYKJtcSf4tGnhzQZi
+e/HDNQdV59E/DYkFqMR7TQ8VyZmbBIzvP67acrL6/4De1grWYH5jjp/YppSNbC4d
+D4kvPnKwyT0w4NrQh75jms3iT2Zfnz7s56QKptKkz99Qn29gjVLRoyVX0lYz1dE5
+ePhyidW/EL6r6+p+6v+N1//0Yoe3Fc5bGX0k4lW0U72sgoaNTtJs+LyJGBCQi6yZ
+YWu16uauuVbP/LkkC0ZNoCZUw7S/IDknxPmAuA0cNAivAOIt3JY7Xy6xZ5lc638b
+3xitpgn7fv1kb0LkreVTlfwm/NoHVyT0gfsZ6F77XEOnw1raLnbbJ5dT6kFiM+n9
+G3uAn/GWkWLW4nCk8zgURkEHhaw7P9sheFq4/hX0Q2dCS6KWc4QgMMYeV+aFnBdG
+9LAy24njMsHBb48CCdv0YrQUujBi/lkkhdMxHzB9mJBdAQgIDOjE77e+ASWVgmPx
+DMB3vJixpWSjSy/J1RFBAoIBAQD6RU4w4Ddsj4E7sposRvJuGaFV2bfWjzaUcqeZ
+LalGMrinLogQ3qN3pz9AUmkoixbkNsMeM6OiVAYC5b4MQKVPlYdIvbh77xyHhlhf
+9woxEQ2hKtjM51K804VPr/hVPsyyO4eFWqBF2kMBbdkiMiiLeoHEfl5+2aXdCT1j
+Jy2N1dXqk5Hw93YF6qKmtj5tpwvhPduzWYfkAneLIJMkb61+Zdnd8M22q/NSVyQf
+H/MxrDK/BLDwJdSzKIR3aJmQHdx3u8ih9uzyqRGtPBfRMtmZ6Yz8rVbCfO9oQlAd
+ssaqiWDlBNJbXItUoLie63hWNmBfSPDLNpjvxaRqvomatYWNAoIBAQDvggTY8BaF
+Mu1dUd5/UgNVmFxuXLUqWKgOTsudX5iyeyNTDicouV24hEHKvTlcoGQDP/IV9CDX
+mVwE0y3Bb9+0kWUah0uyXkisyH31xqgQQ2M0CLf/ENnTuz4ldwXOaEuw1gLs7zg6
+bdqgUgG12gzI7G5cKGH5OZdBe9mx/24R6ucxRFTqD2nVH1cLNN1V3oY9P8ZGFVlp
+e7XWLoH0MBRZY6x7MRsW0jCHOWL7llGo27FqzT0zThhMEts9HrUc8JbOeQ0xUNsJ
+k2P/jyeec2NR6tzAWRlnUbNjx//bkuM8q90Js/JUJnsnSSLpfrRSspIvrg6UEgLR
+1mPU5DlJQcXhAoIBAFppx0Go/tzdSxbCAyiTyOk1oS9epCeDPXiLoziXYlvV3xem
+m8lcZTnI1fTq0Mqw1OhFUGAMz8TJDhLl5K6QfCgwINuKjqdXTrs3MZ4ZpTjsrDvZ
+OtFrkFxfHf4X2GMTeOe60c6/Wr9hhmtxv8u2yyb4bwEJliHFh0I/IEo90Rs4cTt5
+bHPdMmoYxgHsPMloW8ZXjpNQeONKcN12OzIilk7fhMHFSMwBern4eTg1VqpPR1Xy
+3+kiAaFntdNdmnySDR/EW0sH5boUkio/V2tgL3SHB0QRaxKACA1mR4MzHsplLvgN
+seEHod5E4e2nq6WZp19E/pirdLzKbgeSJiwZ/9ECggEBALQHuXkPzv3EcCOLXIG7
+tgHrCt9yFEOGbJyEogzjRLY0VTMjGlBENaxyzbmFTs7PSR4gPOo/nUgyYLbHvkb+
+vtrNx3+PX1juAhbOhc/uyXmgDbuZKiUyF2pN/sLOmrCyOOLtmzlZ/5v74zBLNDnr
+c1y8S6A+QpbBsW5pmBNx+tzBA5NG18UwXM70RcuIqy7Wm3UCsRkRByqA8QfT4Z8Z
+XNJsV8Qp/0DCMfQTMNIIBc21hcDQEUa1VxInwmBI6r6cId+FomMFcf/aqHn6sz8p
+YOi8b76tuqitAvjn5uy3ltOOJBIdDvQuELhRA0scEJNw4u2wGgk3GKN+UYA/JMhq
+BkECggEAQokEqPi8FgdR1HgqnmvX+975eXrdcp098yNygHVJzDQnqau/NZXDy3ph
+GVJ2onvzWRMbICiCx4yF/crlxFmc25mUTGxssRb1jzsDFdk+VEiTAzPjOatSBjtZ
+VtDp7oTlaJHF/GNJPathEiev2sSZp+AZnkRI7dE2B2VNrQMCD+XdCrZHExS78gzy
+FCWL3gzQIz+ajy+NVb63wLmV2oc2+d4GaWnbu8F86AhveoBRP3hVUzIHP5m+wzmp
+J/Fv/QMd3YRUTc40xV+ieBqRsstKXo3rYKyGy+dD0XKiTNHmtmHL9+/hRDnems/M
+u2Iow+n+vuh3lPHBmdmCxfYo50P+HA==
+-----END PRIVATE KEY-----

tests/main.zig

@@ -0,0 +1,11 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+comptime {
+    if (@import("builtin").is_test) {
+        _ = @import("./nats.zig");
+        _ = @import("./connection.zig");
+        _ = @import("./message.zig");
+        _ = @import("./subscription.zig");
+    }
+}

tests/message.zig

@@ -0,0 +1,117 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+const std = @import("std");
+
+const nats = @import("nats");
+
+const util = @import("./util.zig");
+
+test "nats.Message" {
+    const message_subject: [:0]const u8 = "hello";
+    const message_reply: [:0]const u8 = "reply";
+    const message_data: [:0]const u8 = "world";
+
+    // have to initialize the library so the reference counter can correctly destroy
+    // objects, otherwise we segfault on trying to free the memory.
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    {
+        const message = try nats.Message.create(message_subject, null, message_data);
+        defer message.destroy();
+    }
+
+    {
+        const message = try nats.Message.create(message_subject, message_reply, null);
+        defer message.destroy();
+    }
+
+    {
+        const message = try nats.Message.create(message_subject, null, null);
+        defer message.destroy();
+    }
+
+    const message = try nats.Message.create(message_subject, message_reply, message_data);
+    defer message.destroy();
+
+    const subject = message.getSubject();
+    try std.testing.expectEqualStrings(message_subject, subject);
+
+    const reply = message.getReply() orelse return error.TestUnexpectedResult;
+    try std.testing.expectEqualStrings(message_reply, reply);
+
+    const data = message.getData() orelse return error.TestUnexpectedResult;
+    try std.testing.expectEqualStrings(message_data, data);
+
+    try std.testing.expectEqual(message_data.len, message.getDataLength());
+
+    const message_header: [:0]const u8 = "foo";
+    const message_hvalues: []const [:0]const u8 = &.{ "bar", "baz" };
+    try message.setHeaderValue(message_header, message_hvalues[0]);
+
+    try std.testing.expectEqualStrings(message_hvalues[0], try message.getHeaderValue(message_header));
+    try message.addHeaderValue(message_header, message_hvalues[1]);
+    try std.testing.expectEqualStrings(message_hvalues[0], try message.getHeaderValue(message_header));
+
+    {
+        var idx: usize = 0;
+        var val_iter = try message.getHeaderValueIterator(message_header);
+        defer val_iter.destroy();
+
+        while (val_iter.next()) |value| : (idx += 1) {
+            try std.testing.expect(idx < message_hvalues.len);
+            try std.testing.expectEqualStrings(message_hvalues[idx], value);
+        }
+    }
+
+    {
+        var header_iter = try message.getHeaderIterator();
+        defer header_iter.destroy();
+
+        while (header_iter.next()) |header| {
+            try std.testing.expectEqualStrings(message_header, header.key);
+            try std.testing.expectEqualStrings(message_hvalues[0], try header.value());
+
+            var idx: usize = 0;
+            var val_iter = try header.valueIterator();
+            defer val_iter.destroy();
+
+            while (val_iter.next()) |value| : (idx += 1) {
+                try std.testing.expect(idx < message_hvalues.len);
+                try std.testing.expectEqualStrings(message_hvalues[idx], value);
+            }
+
+            try std.testing.expect(val_iter.peek() == null);
+        }
+        try std.testing.expect(header_iter.peek() == null);
+    }
+
+    try message.deleteHeader(message_header);
+    _ = message.isNoResponders();
+}
+
+test "send nats.Message" {
+    var server = try util.TestServer.launch(.{});
+    defer server.stop();
+
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const connection = try nats.Connection.connectTo(server.url);
+    defer connection.destroy();
+
+    const message_subject: [:0]const u8 = "hello";
+    const message_reply: [:0]const u8 = "reply";
+    const message_data: [:0]const u8 = "world";
+    const message_header: [:0]const u8 = "foo";
+    const message_hvalues: []const [:0]const u8 = &.{ "bar", "baz" };
+
+    const message = try nats.Message.create(message_subject, message_reply, message_data);
+    defer message.destroy();
+
+    try message.setHeaderValue(message_header, message_hvalues[0]);
+    try message.addHeaderValue(message_header, message_hvalues[1]);
+
+    try connection.publishMessage(message);
+}

tests/nats.zig

@@ -0,0 +1,99 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+const std = @import("std");
+
+const nats = @import("nats");
+
+test "version" {
+    const version = nats.getVersion();
+    const vernum = nats.getVersionNumber();
+
+    try std.testing.expectEqualStrings("3.9.3", version);
+    try std.testing.expectEqual(@as(u32, 0x03_09_03), vernum);
+    try std.testing.expect(nats.checkCompatibility());
+}
+
+test "time" {
+    const now = nats.now();
+    const nownano = nats.nowInNanoseconds();
+
+    nats.sleep(1);
+
+    const later = nats.now();
+    const laternano = nats.nowInNanoseconds();
+
+    try std.testing.expect(later >= now);
+    try std.testing.expect(laternano >= nownano);
+}
+
+test "init" {
+    {
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+    }
+
+    {
+        // a completely random number
+        try nats.init(900_142_069);
+        nats.deinit();
+    }
+
+    {
+        try nats.init(0);
+        try nats.deinitWait(1000);
+    }
+}
+
+test "misc" {
+    {
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+
+        try nats.setMessageDeliveryPoolSize(500);
+    }
+
+    {
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+
+        // just test that the function is wrapped properly
+        nats.releaseThreadMemory();
+    }
+
+    blk: {
+        try nats.init(nats.default_spin_count);
+        defer nats.deinit();
+
+        // this is a mess of a test that is designed to fail because actually we're
+        // testing out the error reporting functions instead of signing. Nice bait
+        // and switch.
+        const signed = nats.sign("12345678", "12345678") catch {
+            const err = nats.getLastError();
+            std.debug.print("as expected, signing failed: {s}\n", .{err.desc});
+
+            var stackmem = [_]u8{0} ** 512;
+            var stackbuf: []u8 = &stackmem;
+
+            nats.getLastErrorStack(&stackbuf) catch {
+                std.debug.print("Actually, the error stack was too big\n", .{});
+                break :blk;
+            };
+
+            std.debug.print("stack: {s}\n", .{stackbuf});
+            break :blk;
+        };
+
+        std.heap.raw_c_allocator.free(signed);
+    }
+}
+
+test "inbox" {
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const inbox = try nats.createInbox();
+    defer nats.destroyInbox(inbox);
+
+    std.debug.print("inbox: {s}\n", .{inbox});
+}

tests/subscription.zig

@@ -0,0 +1,298 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+const std = @import("std");
+
+const nats = @import("nats");
+
+const util = @import("./util.zig");
+
+test "nats.Subscription" {
+    var server = try util.TestServer.launch(.{});
+    defer server.stop();
+
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const connection = try nats.Connection.connectTo(server.url);
+    defer connection.destroy();
+
+    const message_subject: [:0]const u8 = "hello";
+    const message_reply: [:0]const u8 = "reply";
+    const message_data: [:0]const u8 = "world";
+
+    const message = try nats.Message.create(message_subject, message_reply, message_data);
+    defer message.destroy();
+
+    {
+        const subscription = try connection.subscribeSync(message_subject);
+        defer subscription.destroy();
+
+        try subscription.autoUnsubscribe(1);
+        try subscription.setPendingLimits(.{ .messages = 10, .bytes = 1000 });
+        _ = try subscription.getPendingLimits();
+        _ = try subscription.getPending();
+        _ = try subscription.getMaxPending();
+        try subscription.clearMaxPending();
+        _ = try subscription.getDelivered();
+        _ = try subscription.getDropped();
+        _ = try subscription.getStats();
+        _ = try subscription.queuedMessageCount();
+        _ = subscription.getId();
+        const subj = subscription.getSubject() orelse return error.TestUnexpectedResult;
+        try std.testing.expectEqualStrings(message_subject, subj);
+
+        try connection.publishMessage(message);
+        const roundtrip = try subscription.nextMessage(1000);
+        try std.testing.expectEqualStrings(
+            message_data,
+            roundtrip.getData() orelse return error.TestUnexpectedResult,
+        );
+
+        try std.testing.expect(subscription.isValid() == false);
+    }
+
+    {
+        const subscription = try connection.queueSubscribeSync(message_subject, "queuegroup");
+        defer subscription.destroy();
+
+        try subscription.drain();
+        try subscription.waitForDrainCompletion(1000);
+        _ = subscription.drainCompletionStatus();
+    }
+
+    {
+        const subscription = try connection.subscribeSync(message_subject);
+        defer subscription.destroy();
+
+        try subscription.drain();
+        try subscription.waitForDrainCompletion(1000);
+        _ = subscription.drainCompletionStatus();
+    }
+
+    {
+        const subscription = try connection.subscribeSync(message_subject);
+        defer subscription.destroy();
+
+        try subscription.drainTimeout(1000);
+        try subscription.waitForDrainCompletion(1000);
+    }
+}
+
+fn onMessage(
+    userdata: *const u32,
+    connection: *nats.Connection,
+    subscription: *nats.Subscription,
+    message: *nats.Message,
+) void {
+    _ = subscription;
+    _ = userdata;
+
+    if (message.getReply()) |reply| {
+        connection.publish(reply, "greetings") catch @panic("OH NO");
+    } else @panic("HOW");
+}
+
+fn onClose(userdata: *[]const u8) void {
+    userdata.* = "closed";
+}
+
+test "nats.Subscription (async)" {
+    var server = try util.TestServer.launch(.{});
+    defer server.stop();
+
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const connection = try nats.Connection.connectTo(server.url);
+    defer connection.destroy();
+
+    const message_subject: [:0]const u8 = "hello";
+    const message_reply: [:0]const u8 = "reply";
+    const message_data: [:0]const u8 = "world";
+
+    const message = try nats.Message.create(message_subject, message_reply, message_data);
+    defer message.destroy();
+
+    {
+        var closed: []const u8 = "test";
+        {
+            const count: u32 = 0;
+            const subscription = try connection.subscribe(*const u32, message_subject, onMessage, &count);
+            defer subscription.destroy();
+
+            try subscription.setCompletionCallback(*[]const u8, onClose, &closed);
+
+            const response = try connection.requestMessage(message, 1000);
+            try std.testing.expectEqualStrings(
+                "greetings",
+                response.getData() orelse return error.TestUnexpectedResult,
+            );
+        }
+        // we have to sleep to allow the close callback to run. I am worried this may
+        // still end up being flaky, however.
+        nats.sleep(1);
+        try std.testing.expectEqualStrings("closed", closed);
+    }
+
+    {
+        const count: u32 = 0;
+        const subscription = try connection.subscribeTimeout(
+            *const u32,
+            message_subject,
+            1000,
+            onMessage,
+            &count,
+        );
+        defer subscription.destroy();
+
+        const response = try connection.requestMessage(message, 1000);
+        try std.testing.expectEqualStrings(
+            "greetings",
+            response.getData() orelse return error.TestUnexpectedResult,
+        );
+    }
+
+    {
+        const count: u32 = 0;
+        const subscription = try connection.queueSubscribe(
+            *const u32,
+            message_subject,
+            "queuegroup",
+            onMessage,
+            &count,
+        );
+        defer subscription.destroy();
+
+        const response = try connection.requestMessage(message, 1000);
+        try std.testing.expectEqualStrings(
+            "greetings",
+            response.getData() orelse return error.TestUnexpectedResult,
+        );
+    }
+
+    {
+        var count: u32 = 0;
+        const subscription = try connection.queueSubscribeTimeout(
+            *const u32,
+            message_subject,
+            "queuegroup",
+            1000,
+            onMessage,
+            &count,
+        );
+        defer subscription.destroy();
+
+        const response = try connection.requestMessage(message, 1000);
+        try std.testing.expectEqualStrings(
+            "greetings",
+            response.getData() orelse return error.TestUnexpectedResult,
+        );
+    }
+}
+
+fn onVoidMessage(
+    userdata: void,
+    connection: *nats.Connection,
+    subscription: *nats.Subscription,
+    message: *nats.Message,
+) void {
+    _ = subscription;
+    _ = userdata;
+
+    if (message.getReply()) |reply| {
+        connection.publish(reply, "greetings") catch @panic("OH NO");
+    } else @panic("HOW");
+}
+
+fn onVoidClose(userdata: void) void {
+    _ = userdata;
+}
+
+test "nats.Subscription (async, void)" {
+    var server = try util.TestServer.launch(.{});
+    defer server.stop();
+
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const connection = try nats.Connection.connectTo(server.url);
+    defer connection.destroy();
+
+    const message_subject: [:0]const u8 = "hello";
+    const message_reply: [:0]const u8 = "reply";
+    const message_data: [:0]const u8 = "world";
+
+    const message = try nats.Message.create(message_subject, message_reply, message_data);
+    defer message.destroy();
+
+    {
+        {
+            const subscription = try connection.subscribe(void, message_subject, onVoidMessage, {});
+            defer subscription.destroy();
+
+            try subscription.setCompletionCallback(void, onVoidClose, {});
+
+            const response = try connection.requestMessage(message, 1000);
+            try std.testing.expectEqualStrings(
+                "greetings",
+                response.getData() orelse return error.TestUnexpectedResult,
+            );
+        }
+        // we have to sleep to allow the close callback to run.
+        nats.sleep(1);
+    }
+}
+
+fn onNullMessage(
+    userdata: ?*void,
+    connection: *nats.Connection,
+    subscription: *nats.Subscription,
+    message: *nats.Message,
+) void {
+    _ = subscription;
+    _ = userdata;
+
+    if (message.getReply()) |reply| {
+        connection.publish(reply, "greetings") catch @panic("OH NO");
+    } else @panic("HOW");
+}
+
+fn onNullClose(userdata: ?*void) void {
+    _ = userdata;
+}
+
+test "nats.Subscription (async, null)" {
+    var server = try util.TestServer.launch(.{});
+    defer server.stop();
+
+    try nats.init(nats.default_spin_count);
+    defer nats.deinit();
+
+    const connection = try nats.Connection.connectTo(server.url);
+    defer connection.destroy();
+
+    const message_subject: [:0]const u8 = "hello";
+    const message_reply: [:0]const u8 = "reply";
+    const message_data: [:0]const u8 = "world";
+
+    const message = try nats.Message.create(message_subject, message_reply, message_data);
+    defer message.destroy();
+
+    {
+        {
+            const subscription = try connection.subscribe(?*void, message_subject, onNullMessage, null);
+            defer subscription.destroy();
+
+            try subscription.setCompletionCallback(?*void, onNullClose, null);
+
+            const response = try connection.requestMessage(message, 1000);
+            try std.testing.expectEqualStrings(
+                "greetings",
+                response.getData() orelse return error.TestUnexpectedResult,
+            );
+        }
+        // we have to sleep to allow the close callback to run.
+        nats.sleep(1);
+    }
+}

tests/util.zig

@@ -0,0 +1,156 @@
+// This file is licensed under the CC0 1.0 license.
+// See: https://creativecommons.org/publicdomain/zero/1.0/legalcode
+
+const std = @import("std");
+
+const KeyCert = struct { key: [:0]const u8, cert: [:0]const u8 };
+const server_rsa: KeyCert = .{
+    .key = @embedFile("./data/server-rsa.key"),
+    .cert = @embedFile("./data/server-rsa.cert"),
+};
+
+const server_ecc: KeyCert = .{
+    .key = @embedFile("./data/server-ecc.key"),
+    .cert = @embedFile("./data/server-ecc.cert"),
+};
+
+const TestLaunchError = error{
+    NoLaunchStringFound,
+};
+
+pub const TestServer = struct {
+    allocator: std.mem.Allocator,
+    process: std.process.Child,
+    key_dir: ?std.testing.TmpDir,
+    url: [:0]u8,
+
+    pub const LaunchOptions = struct {
+        executable: []const u8 = "nats-server",
+        port: u16 = 4222,
+        auth: union(enum) {
+            none: void,
+            token: []const u8,
+            password: struct { user: []const u8, pass: []const u8 },
+        } = .none,
+        tls: enum {
+            none,
+            rsa,
+            ecc,
+        } = .none,
+        allocator: std.mem.Allocator = std.testing.allocator,
+
+        pub fn connectionString(self: LaunchOptions) ![:0]u8 {
+            const authstr: []u8 = switch (self.auth) {
+                // dupe this so we don't have to handle an edge case where it
+                // does not need to be freed
+                .none => try self.allocator.dupe(u8, ""),
+                .token => |tok| try std.fmt.allocPrint(
+                    self.allocator,
+                    "{s}@",
+                    .{tok},
+                ),
+                .password => |auth| try std.fmt.allocPrint(
+                    self.allocator,
+                    "{s}:{s}@",
+                    .{ auth.user, auth.pass },
+                ),
+            };
+            defer self.allocator.free(authstr);
+
+            return try std.fmt.allocPrintZ(self.allocator, "nats://{s}127.0.0.1:{d}", .{ authstr, self.port });
+        }
+    };
+
+    pub fn launch(options: LaunchOptions) !TestServer {
+        var portbuf = [_]u8{0} ** 5;
+        const strport = try std.fmt.bufPrint(&portbuf, "{d}", .{options.port});
+
+        var key_dir: ?std.testing.TmpDir = null;
+        var key_path: ?[]const u8 = null;
+        var cert_path: ?[]const u8 = null;
+
+        errdefer if (key_dir) |*kd| kd.cleanup();
+
+        // ChildProcess copies these, so we can free them before the process has
+        // closed.
+        defer {
+            if (key_path) |kp| options.allocator.free(kp);
+            if (cert_path) |cp| options.allocator.free(cp);
+        }
+
+        const args: [][]const u8 = blk: {
+            const executable: []const []const u8 = &.{options.executable};
+            const listen: []const []const u8 = &.{ "-a", "127.0.0.1" };
+            const port: []const []const u8 = &.{ "-p", strport };
+            const auth: []const []const u8 = switch (options.auth) {
+                .none => &[_][]const u8{},
+                .token => |tok| &[_][]const u8{ "--auth", tok },
+                .password => |auth| &[_][]const u8{ "--user", auth.user, "--pass", auth.pass },
+            };
+            const tls: []const []const u8 = switch (options.tls) {
+                .none => &[_][]const u8{},
+                .rsa, .ecc => |keytype| keyb: {
+                    const pair = switch (keytype) {
+                        .rsa => server_rsa,
+                        .ecc => server_ecc,
+                        else => unreachable,
+                    };
+
+                    const out_dir = std.testing.tmpDir(.{});
+                    key_dir = out_dir;
+
+                    try out_dir.dir.writeFile(.{ .sub_path = "server.key", .data = pair.key });
+                    try out_dir.dir.writeFile(.{ .sub_path = "server.cert", .data = pair.cert });
+                    // since testing.tmpDir will actually bury itself in zig-cache/tmp,
+                    // there's not an easy way to extract files from within the temp
+                    // directory except through using realPath, as far as I can tell
+                    // (or reproducing the path naming logic, but that seems fragile).
+                    const out_key = try out_dir.dir.realpathAlloc(options.allocator, "server.key");
+                    const out_cert = try out_dir.dir.realpathAlloc(options.allocator, "server.cert");
+
+                    key_path = out_key;
+                    cert_path = out_cert;
+                    break :keyb &[_][]const u8{ "--tls", "--tlscert", out_cert, "--tlskey", out_key };
+                },
+            };
+
+            break :blk try std.mem.concat(
+                options.allocator,
+                []const u8,
+                &.{ executable, listen, port, auth, tls },
+            );
+        };
+
+        defer options.allocator.free(args);
+
+        var child = std.process.Child.init(args, options.allocator);
+        child.stdin_behavior = .Ignore;
+        child.stdout_behavior = .Pipe;
+        child.stderr_behavior = .Pipe;
+
+        try child.spawn();
+        var poller = std.io.poll(options.allocator, enum { stderr }, .{ .stderr = child.stderr.? });
+        defer poller.deinit();
+
+        while (try poller.poll()) {
+            if (std.mem.indexOf(u8, poller.fifo(.stderr).buf, "[INF] Server is ready")) |_| {
+                return .{
+                    .allocator = options.allocator,
+                    .process = child,
+                    .key_dir = key_dir,
+                    .url = try options.connectionString(),
+                };
+            }
+        }
+
+        _ = try child.kill();
+        std.debug.print("output: {s}\n", .{poller.fifo(.stderr).buf});
+        return error.NoLaunchStringFound;
+    }
+
+    pub fn stop(self: *TestServer) void {
+        if (self.key_dir) |*key_dir| key_dir.cleanup();
+        self.allocator.free(self.url);
+        _ = self.process.kill() catch return;
+    }
+};