jetstream: properly handle ping, add example

build.zig

···
 
     const test_step = b.step("test", "Run unit tests");
     test_step.dependOn(&run_lib_unit_tests.step);
+
+    // Example
+    const example_mod = b.createModule(.{
+        .root_source_file = b.path("example/stream.zig"),
+        .target = target,
+        .optimize = optimize,
+    });
+    example_mod.addImport("ourio", ourio_dep.module("ourio"));
+    example_mod.addImport("stda", ourio_dep.module("stda"));
+    example_mod.addImport("atproto", lib_mod);
+    const exe = b.addExecutable(.{
+        .name = "stream",
+        .root_module = example_mod,
+    });
+    const run_cmd = b.addRunArtifact(exe);
+    run_cmd.step.dependOn(b.getInstallStep());
+
+    if (b.args) |args| {
+        run_cmd.addArgs(args);
+    }
+
+    const run_step = b.step("run", "Run the app");
+    run_step.dependOn(&run_cmd.step);
 }

example/stream.zig

···
+const std = @import("std");
+const builtin = @import("builtin");
+const ourio = @import("ourio");
+const atproto = @import("atproto");
+const stda = @import("stda");
+
+const dns = stda.net.dns;
+
+pub fn main() !void {
+    var debug_allocator: std.heap.DebugAllocator(.{}) = .init;
+    const gpa, const is_debug = gpa: {
+        break :gpa switch (builtin.mode) {
+            .Debug, .ReleaseSafe => .{ debug_allocator.allocator(), true },
+            .ReleaseFast, .ReleaseSmall => .{ std.heap.smp_allocator, false },
+        };
+    };
+    defer if (is_debug) {
+        _ = debug_allocator.deinit();
+    };
+
+    var io: ourio.Ring = try .init(gpa, 64);
+    defer io.deinit();
+
+    var bundle: std.crypto.Certificate.Bundle = .{};
+    try bundle.rescan(gpa);
+    defer bundle.deinit(gpa);
+
+    var resolver: dns.Resolver = undefined;
+    try resolver.init(gpa, &io, .{});
+    try io.run(.until_done);
+
+    var counter: Counter = .{};
+
+    var jetstream: atproto.Jetstream = undefined;
+    try jetstream.init(gpa, &io, bundle, .{}, &resolver, .{ .cb = print, .msg = 1, .ptr = &counter });
+
+    _ = try io.timer(.{ .sec = 1 }, .{ .cb = print, .msg = 2, .ptr = &counter });
+
+    try io.run(.until_done);
+}
+
+const Counter = struct {
+    tick: u8 = 0,
+    count: usize = 0,
+};
+
+fn print(io: *ourio.Ring, task: ourio.Task) anyerror!void {
+    const counter = task.userdataCast(Counter);
+
+    switch (task.msg) {
+        1 => {
+            const result = task.result.?;
+            _ = try result.userptr;
+            counter.count += 1;
+        },
+        2 => {
+            const writer = std.io.getStdOut().writer();
+            try writer.writeAll("\r\x1b[2K");
+            try writer.print("{d} events per second", .{counter.count});
+            counter.count = 0;
+            _ = try io.timer(.{ .sec = 1 }, .{ .cb = print, .msg = 2, .ptr = counter });
+        },
+        else => {},
+    }
+}

src/jetstream.zig

···
             defer _ = arena.reset(.retain_capacity);
             switch (data) {
                 .text => |s| {
-                    const event: Event = .{ .raw = s, .value = try json.parseFromSliceLeaky(
-                        Event,
+                    var event: Event = .{ .raw = s, .value = try json.parseFromSliceLeaky(
+                        json.Value,
                         arena.allocator(),
                         s,
                         .{ .allocate = .alloc_if_needed },
···
                     };
                     try self.ctx.cb(io, task);
                 },
-                else => return error.UnsupportedOp,
+
+                .close => {
+                    const task: ourio.Task = .{
+                        .callback = self.ctx.cb,
+                        .userdata = self.ctx.ptr,
+                        .msg = self.ctx.msg,
+                        .result = .{ .userptr = error.ConnectionResetByPeer },
+                    };
+                    try self.ctx.cb(io, task);
+                },
+
+                .ping => |v| {
+                    const byte1: FrameIterator.Byte1 = .{
+                        .final = true,
+                        .reserved = 0,
+                        .opcode = .pong,
+                    };
+                    const byte2: FrameIterator.Byte2 = .{
+                        .len = @intCast(v.len),
+                        .mask = true,
+                    };
+                    const mask: []const u8 = &.{ 0x12, 0x34, 0x56, 0x78 };
+                    try self.state.conn.write(self.gpa, &.{ @bitCast(byte1), @bitCast(byte2) });
+                    try self.state.conn.write(self.gpa, mask);
+                    // TODO: actually apply the mask. As of 2025-05-21 the jetstream doesn't send a
+                    // ping payload so it doesn't matter
+                    try self.state.conn.write(self.gpa, v);
+                    try self.state.conn.flush(self.gpa, io);
+                },
+
+                else => {
+                    return error.UnsupportedOp;
+                },
             }
         }
 
···
         const value = self.value.object.get("kind").?;
         const k = std.meta.stringToEnum(KindEnum, value.string).?;
         switch (k) {
-            .commit => return .{ .commit = self.value.object.get("commit").? },
-            .identity => return .{ .identity = self.value.object.get("identity").? },
-            .account => return .{ .account = self.value.object.get("account").? },
+            .commit => return .{ .commit = .{ .value = self.value.object.get("commit").? } },
+            .identity => return .{ .identity = .{ .value = self.value.object.get("identity").? } },
+            .account => return .{ .account = .{ .value = self.value.object.get("account").? } },
         }
     }
 
···
         text: []const u8,
         binary: []const u8,
         close,
-        ping,
+        ping: []const u8,
         pong,
     };
 
···
             .text => return .{ .text = self.bytes[data_start..end] },
             .binary => return .{ .binary = self.bytes[data_start..end] },
             .close => return .close,
-            .ping => return .ping,
+            .ping => return .{ .ping = self.bytes[data_start..end] },
             .pong => return .pong,
             else => return error.InvalidOpcode,
         }

src/root.zig

···
 
 pub const Did = did.Did;
 pub const Jetstream = jetstream.Stream;
+pub const JetstreamEvent = jetstream.Event;
 
 test {
     _ = @import("did.zig");