Introduce more general notification capabilities

Replaces the existing, very specialized Notification with something more
general.

Currently, the existing page_navigate and page_navigated have been migrated.

Telemetry's page navigation event now also hooks into these events to generate
the telemetry record.

src/app.zig

@@ -5,6 +5,7 @@ const js = @import("runtime/js.zig");
 const Loop = @import("runtime/loop.zig").Loop;
 const HttpClient = @import("http/client.zig").Client;
 const Telemetry = @import("telemetry/telemetry.zig").Telemetry;
+const Notification = @import("notification.zig").Notification;
 
 const log = std.log.scoped(.app);
 
@@ -17,6 +18,7 @@ pub const App = struct {
     telemetry: Telemetry,
     http_client: HttpClient,
     app_dir_path: ?[]const u8,
+    notification: *Notification,
 
     pub const RunMode = enum {
         help,
@@ -41,6 +43,9 @@ pub const App = struct {
         loop.* = try Loop.init(allocator);
         errdefer loop.deinit();
 
+        const notification = try Notification.init(allocator, null);
+        errdefer notification.deinit();
+
         const app_dir_path = getAndMakeAppDir(allocator);
 
         app.* = .{
@@ -48,12 +53,14 @@ pub const App = struct {
             .allocator = allocator,
             .telemetry = undefined,
             .app_dir_path = app_dir_path,
+            .notification = notification,
             .http_client = try HttpClient.init(allocator, 5, .{
                 .tls_verify_host = config.tls_verify_host,
             }),
             .config = config,
         };
         app.telemetry = Telemetry.init(app, config.run_mode);
+        try app.telemetry.register(app.notification);
 
         return app;
     }
@@ -67,6 +74,7 @@ pub const App = struct {
         self.loop.deinit();
         allocator.destroy(self.loop);
         self.http_client.deinit();
+        self.notification.deinit();
         allocator.destroy(self);
     }
 };

src/browser/browser.zig

@@ -58,6 +58,7 @@ pub const Browser = struct {
     allocator: Allocator,
     http_client: *http.Client,
     page_arena: ArenaAllocator,
+    notification: *Notification,
 
     pub fn init(app: *App) !Browser {
         const allocator = app.allocator;
@@ -67,11 +68,15 @@ pub const Browser = struct {
         });
         errdefer env.deinit();
 
+        const notification = try Notification.init(allocator, app.notification);
+        errdefer notification.deinit();
+
         return .{
             .app = app,
             .env = env,
             .session = null,
             .allocator = allocator,
+            .notification = notification,
             .http_client = &app.http_client,
             .page_arena = ArenaAllocator.init(allocator),
         };
@@ -81,13 +86,14 @@ pub const Browser = struct {
         self.closeSession();
         self.env.deinit();
         self.page_arena.deinit();
+        self.notification.deinit();
     }
 
-    pub fn newSession(self: *Browser, ctx: anytype) !*Session {
+    pub fn newSession(self: *Browser) !*Session {
         self.closeSession();
         self.session = @as(Session, undefined);
         const session = &self.session.?;
-        try Session.init(session, self, ctx);
+        try Session.init(session, self);
         return session;
     }
 
@@ -119,23 +125,7 @@ pub const Session = struct {
 
     page: ?Page = null,
 
-    // recipient of notification, passed as the first parameter to notify
+    fn init(self: *Session, browser: *Browser) !void {
-    notify_ctx: *anyopaque,
-    notify_func: *const fn (ctx: *anyopaque, notification: *const Notification) anyerror!void,
-
-    fn init(self: *Session, browser: *Browser, ctx: anytype) !void {
-        const ContextT = @TypeOf(ctx);
-        const ContextStruct = switch (@typeInfo(ContextT)) {
-            .@"struct" => ContextT,
-            .pointer => |ptr| ptr.child,
-            .void => NoopContext,
-            else => @compileError("invalid context type"),
-        };
-
-        // ctx can be void, to be able to store it in our *anyopaque field, we
-        // need to play a little game.
-        const any_ctx: *anyopaque = if (@TypeOf(ctx) == void) @constCast(@ptrCast(&{})) else ctx;
-
         var executor = try browser.env.newExecutor();
         errdefer executor.deinit();
 
@@ -143,8 +133,6 @@ pub const Session = struct {
         self.* = .{
             .browser = browser,
             .executor = executor,
-            .notify_ctx = any_ctx,
-            .notify_func = ContextStruct.notify,
             .arena = ArenaAllocator.init(allocator),
             .storage_shed = storage.Shed.init(allocator),
             .cookie_jar = storage.CookieJar.init(allocator),
@@ -213,12 +201,6 @@ pub const Session = struct {
             .reason = .anchor,
         });
     }
-
-    fn notify(self: *const Session, notification: *const Notification) void {
-        self.notify_func(self.notify_ctx, notification) catch |err| {
-            log.err("notify {}: {}", .{ std.meta.activeTag(notification.*), err });
-        };
-    }
 };
 
 // Page navigates to an url.
@@ -350,20 +332,15 @@ pub const Page = struct {
         // redirect)
         self.url = request_url;
 
-        session.browser.app.telemetry.record(.{ .navigate = .{
-            .proxy = false,
-            .tls = std.ascii.eqlIgnoreCase(request_url.scheme(), "https"),
-        } });
-
         // load the data
         var request = try self.newHTTPRequest(.GET, &self.url, .{ .navigation = true });
         defer request.deinit();
 
-        session.notify(&.{ .page_navigate = .{
+        session.browser.notification.dispatch(.page_navigate, &.{
             .url = &self.url,
             .reason = opts.reason,
             .timestamp = timestamp(),
-        } });
+        });
 
         var response = try request.sendSync(.{});
 
@@ -399,10 +376,10 @@ pub const Page = struct {
             self.raw_data = arr.items;
         }
 
-        session.notify(&.{ .page_navigated = .{
+        session.browser.notification.dispatch(.page_navigated, &.{
             .url = &self.url,
             .timestamp = timestamp(),
-        } });
+        });
     }
 
     // https://html.spec.whatwg.org/#read-html

src/cdp/cdp.zig

@@ -312,7 +312,7 @@ pub fn BrowserContext(comptime CDP_T: type) type {
         fn init(self: *Self, id: []const u8, cdp: *CDP_T) !void {
             const allocator = cdp.allocator;
 
-            const session = try cdp.browser.newSession(self);
+            const session = try cdp.browser.newSession();
             const arena = session.arena.allocator();
 
             const inspector = try cdp.browser.env.newInspector(arena, self);
@@ -337,6 +337,10 @@ pub fn BrowserContext(comptime CDP_T: type) type {
                 .inspector = inspector,
             };
             self.node_search_list = Node.Search.List.init(allocator, &self.node_registry);
+            errdefer self.deinit();
+
+            try cdp.browser.notification.register(.page_navigate, self, onPageNavigate);
+            try cdp.browser.notification.register(.page_navigated, self, onPageNavigated);
         }
 
         pub fn deinit(self: *Self) void {
@@ -352,6 +356,7 @@ pub fn BrowserContext(comptime CDP_T: type) type {
             }
             self.node_registry.deinit();
             self.node_search_list.deinit();
+            self.cdp.browser.notification.unregisterAll(self);
         }
 
         pub fn reset(self: *Self) void {
@@ -394,13 +399,14 @@ pub fn BrowserContext(comptime CDP_T: type) type {
             return if (raw_url.len == 0) null else raw_url;
         }
 
-        pub fn notify(ctx: *anyopaque, notification: *const Notification) !void {
+        pub fn onPageNavigate(ctx: *anyopaque, data: *const Notification.PageNavigate) !void {
             const self: *Self = @alignCast(@ptrCast(ctx));
-
+            return @import("domains/page.zig").pageNavigate(self, data);
-            switch (notification.*) {
-                .page_navigate => |*pn| return @import("domains/page.zig").pageNavigate(self, pn),
-                .page_navigated => |*pn| return @import("domains/page.zig").pageNavigated(self, pn),
         }
+
+        pub fn onPageNavigated(ctx: *anyopaque, data: *const Notification.PageNavigated) !void {
+            const self: *Self = @alignCast(@ptrCast(ctx));
+            return @import("domains/page.zig").pageNavigated(self, data);
         }
 
         pub fn callInspector(self: *const Self, msg: []const u8) void {

src/main.zig

@@ -97,7 +97,7 @@ pub fn main() !void {
             var browser = try Browser.init(app);
             defer browser.deinit();
 
-            var session = try browser.newSession({});
+            var session = try browser.newSession();
 
             // page
             const page = try session.createPage();

src/notification.zig

@@ -1,9 +1,71 @@
+const std = @import("std");
+
 const URL = @import("url.zig").URL;
 const browser = @import("browser/browser.zig");
 
-pub const Notification = union(enum) {
+const Allocator = std.mem.Allocator;
-    page_navigate: PageNavigate,
+
-    page_navigated: PageNavigated,
+const log = std.log.scoped(.notification);
+
+const List = std.DoublyLinkedList(Listener);
+const Node = List.Node;
+
+// Allows code to register for and emit events.
+// Keeps two lists
+// 1 - for a given event type, a linked list of all the listeners
+// 2 - for a given listener, a list of all it's registration
+// The 2nd one is so that a listener can unregister all of it's listeners
+// (there's currently no need for a listener to unregister only 1 or more
+// specific listener).
+//
+// Scoping is important. Imagine we created a global singleton registry, and our
+// CDP code registers for the "network_bytes_sent" event, because it needs to
+// send messages to the client when this happens. Our HTTP client could then
+// emit a "network_bytes_sent" message. It would be easy, and it would work.
+// That is, it would work until the Telemetry code makes an HTTP request, and
+// because everything's just one big global, that gets picked up by the
+// registered CDP listener, and the telemetry network activity gets sent to the
+// CDP client.
+//
+// To avoid this, one way or another, we need scoping. We could still have
+// a global registry but every "register" and every "emit" has some type of
+// "scope". This would have a run-time cost and still require some coordination
+// between components to share a common scope.
+//
+// Instead, the approach that we take is to have a notification per
+// scope. This makes some things harder, but we only plan on having 2
+// notifications at a given time: one in a Browser and one in the App.
+// What about something like Telemetry, which lives outside of a Browser but
+// still cares about Browser-events (like .page_navigate)? When the Browser
+// notification is created, a `notification_created` event is raised in the
+// App's notification, which Telemetry is registered for. This allows Telemetry
+// to register for events in the Browser notification. See the Telemetry's
+// register function.
+pub const Notification = struct {
+    // Every event type (which are hard-coded), has a list of Listeners.
+    // When the event happens, we dispatch to those listener.
+    event_listeners: EventListeners,
+
+    // list of listeners for a specified receiver
+    // @intFromPtr(listener) -> [@intFromPtr(listener1), @intFromPtr(listener2, ...]
+    // Used when `unregisterAll` is called.
+    listeners: std.AutoHashMapUnmanaged(usize, std.ArrayListUnmanaged(*Node)),
+
+    allocator: Allocator,
+    node_pool: std.heap.MemoryPool(Node),
+
+    const EventListeners = struct {
+        page_navigate: List = .{},
+        page_navigated: List = .{},
+        notification_created: List = .{},
+    };
+
+    const Events = union(enum) {
+        page_navigate: *const PageNavigate,
+        page_navigated: *const PageNavigated,
+        notification_created: *Notification,
+    };
+    const EventType = std.meta.FieldEnum(Events);
 
     pub const PageNavigate = struct {
         timestamp: u32,
@@ -15,4 +77,187 @@ pub const Notification = union(enum) {
         timestamp: u32,
         url: *const URL,
     };
+
+    pub fn init(allocator: Allocator, parent: ?*Notification) !*Notification {
+        // This is put on the heap because we want to raise a .notification_created
+        // event, so that, something like Telemetry, can receive the
+        // .page_navigate event on all notification instances. That can only work
+        // if we dispatch .notification_created with a *Notification.
+        const notification = try allocator.create(Notification);
+        errdefer allocator.destroy(notification);
+
+        notification.* = .{
+            .listeners = .{},
+            .event_listeners = .{},
+            .allocator = allocator,
+            .node_pool = std.heap.MemoryPool(Node).init(allocator),
+        };
+
+        if (parent) |pn| {
+            pn.dispatch(.notification_created, notification);
+        }
+
+        return notification;
+    }
+
+    pub fn deinit(self: *Notification) void {
+        const allocator = self.allocator;
+
+        var it = self.listeners.valueIterator();
+        while (it.next()) |listener| {
+            listener.deinit(allocator);
+        }
+        self.listeners.deinit(allocator);
+        self.node_pool.deinit();
+        allocator.destroy(self);
+    }
+
+    pub fn register(self: *Notification, comptime event: EventType, receiver: anytype, func: EventFunc(event)) !void {
+        var list = &@field(self.event_listeners, @tagName(event));
+
+        var node = try self.node_pool.create();
+        errdefer self.node_pool.destroy(node);
+
+        node.data = .{
+            .list = list,
+            .func = @ptrCast(func),
+            .receiver = receiver,
+            .struct_name = @typeName(@typeInfo(@TypeOf(receiver)).pointer.child),
+        };
+
+        const allocator = self.allocator;
+        const gop = try self.listeners.getOrPut(allocator, @intFromPtr(receiver));
+        if (gop.found_existing == false) {
+            gop.value_ptr.* = .{};
+        }
+        try gop.value_ptr.append(allocator, node);
+
+        // we don't add this until we've successfully added the entry to
+        // self.listeners
+        list.append(node);
+    }
+
+    pub fn unregisterAll(self: *Notification, receiver: *anyopaque) void {
+        const node_pool = &self.node_pool;
+
+        var kv = self.listeners.fetchRemove(@intFromPtr(receiver)) orelse return;
+        for (kv.value.items) |node| {
+            node.data.list.remove(node);
+            node_pool.destroy(node);
+        }
+        kv.value.deinit(self.allocator);
+    }
+
+    pub fn dispatch(self: *Notification, comptime event: EventType, data: ArgType(event)) void {
+        const list = &@field(self.event_listeners, @tagName(event));
+
+        var node = list.first;
+        while (node) |n| {
+            const listener = n.data;
+            const func: EventFunc(event) = @alignCast(@ptrCast(listener.func));
+            func(listener.receiver, data) catch |err| {
+                log.err("{s} '{s}' dispatch error: {}", .{ listener.struct_name, @tagName(event), err });
+            };
+            node = n.next;
+        }
+    }
+};
+
+// Given an event type enum, returns the type of arg the event emits
+fn ArgType(comptime event: Notification.EventType) type {
+    inline for (std.meta.fields(Notification.Events)) |f| {
+        if (std.mem.eql(u8, f.name, @tagName(event))) {
+            return f.type;
+        }
+    }
+    unreachable;
+}
+
+// Given an event type enum, returns the listening function type
+fn EventFunc(comptime event: Notification.EventType) type {
+    return *const fn (*anyopaque, ArgType(event)) anyerror!void;
+}
+
+// An listener. This is 1 receiver, with its function, and the linked list
+// node that goes in the appropriate EventListeners list.
+const Listener = struct {
+    // the receiver of the event, i.e. the self parameter to `func`
+    receiver: *anyopaque,
+
+    // the function to call
+    func: *const anyopaque,
+
+    // For logging slightly better error
+    struct_name: []const u8,
+
+    // The event list this listener belongs to.
+    // We need this in order to be able to remove the node from the list
+    list: *List,
+};
+
+const testing = std.testing;
+test "Notification" {
+    var notifier = try Notification.init(testing.allocator, null);
+    defer notifier.deinit();
+
+    // noop
+    notifier.dispatch(.page_navigate, &.{
+        .timestamp = 4,
+        .url = undefined,
+        .reason = undefined,
+    });
+
+    var tc = TestClient{};
+
+    try notifier.register(.page_navigate, &tc, TestClient.pageNavigate);
+    notifier.dispatch(.page_navigate, &.{
+        .timestamp = 4,
+        .url = undefined,
+        .reason = undefined,
+    });
+    try testing.expectEqual(4, tc.page_navigate);
+
+    notifier.unregisterAll(&tc);
+    notifier.dispatch(.page_navigate, &.{
+        .timestamp = 10,
+        .url = undefined,
+        .reason = undefined,
+    });
+    try testing.expectEqual(4, tc.page_navigate);
+
+    try notifier.register(.page_navigate, &tc, TestClient.pageNavigate);
+    try notifier.register(.page_navigated, &tc, TestClient.pageNavigated);
+    notifier.dispatch(.page_navigate, &.{
+        .timestamp = 10,
+        .url = undefined,
+        .reason = undefined,
+    });
+    notifier.dispatch(.page_navigated, &.{ .timestamp = 6, .url = undefined });
+    try testing.expectEqual(14, tc.page_navigate);
+    try testing.expectEqual(6, tc.page_navigated);
+
+    notifier.unregisterAll(&tc);
+    notifier.dispatch(.page_navigate, &.{
+        .timestamp = 100,
+        .url = undefined,
+        .reason = undefined,
+    });
+    notifier.dispatch(.page_navigated, &.{ .timestamp = 100, .url = undefined });
+    try testing.expectEqual(14, tc.page_navigate);
+    try testing.expectEqual(6, tc.page_navigated);
+}
+
+const TestClient = struct {
+    page_navigate: u32 = 0,
+    page_navigated: u32 = 0,
+
+    fn pageNavigate(ptr: *anyopaque, data: *const Notification.PageNavigate) !void {
+        const self: *TestClient = @alignCast(@ptrCast(ptr));
+        self.page_navigate += data.timestamp;
+    }
+
+    fn pageNavigated(ptr: *anyopaque, data: *const Notification.PageNavigated) !void {
+        const self: *TestClient = @alignCast(@ptrCast(ptr));
+        self.page_navigated += data.timestamp;
+    }
 };

src/telemetry/telemetry.zig

@@ -5,8 +5,9 @@ const Allocator = std.mem.Allocator;
 
 const App = @import("../app.zig").App;
 const Loop = @import("jsruntime").Loop;
-const uuidv4 = @import("../id.zig").uuidv4;
+const Notification = @import("../notification.zig").Notification;
 
+const uuidv4 = @import("../id.zig").uuidv4;
 const log = std.log.scoped(.telemetry);
 const IID_FILE = "iid";
 
@@ -56,6 +57,32 @@ fn TelemetryT(comptime P: type) type {
                 log.warn("failed to record event: {}", .{err});
             };
         }
+
+        // Called outside of `init` because we need a stable pointer for self.
+        // We care page_navigate events, but those happen on a Browser's
+        // notification. This doesn't exist yet, and there isn't only going to
+        // be 1, browsers come and go.
+        // What we can do is register for the `notification_created` event.
+        // In the callback for that, `onNotificationCreated`, we can then register
+        // for the browser-events that we care about.
+        pub fn register(self: *Self, notification: *Notification) !void {
+            if (self.disabled) {
+                return;
+            }
+            try notification.register(.notification_created, self, onNotificationCreated);
+        }
+
+        fn onNotificationCreated(ctx: *anyopaque, new: *Notification) !void {
+            return new.register(.page_navigate, ctx, onPageNavigate);
+        }
+
+        fn onPageNavigate(ctx: *anyopaque, data: *const Notification.PageNavigate) !void {
+            const self: *Self = @alignCast(@ptrCast(ctx));
+            self.record(.{ .navigate = .{
+                .proxy = false,
+                .tls = std.ascii.eqlIgnoreCase(data.url.scheme(), "https"),
+            } });
+        }
     };
 }