zig-atproto / 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 => {},
    }
}