forked from
rockorager.dev/zig-atproto
this repo has no description
jetstream: implement event kinds
+106
-4
+106
-4
src/jetstream.zig
+106
-4
src/jetstream.zig
···
64
64
cursor_us: ?i64 = null,
65
65
};
66
66
67
+
/// Calls ctx.cb with a .userptr pointing to an Event
67
68
pub fn init(
68
69
self: *Stream,
69
70
gpa: Allocator,
···
160
161
.callback = self.ctx.cb,
161
162
.userdata = self.ctx.ptr,
162
163
.msg = self.ctx.msg,
163
-
.result = .{ .userbytes = err },
164
+
.result = .{ .userptr = err },
164
165
};
165
166
try self.ctx.cb(io, task);
166
167
}
···
301
302
302
303
fn decodeFrames(self: *Stream, io: *ourio.Ring, bytes: []const u8) !void {
303
304
try self.buffer.appendSlice(self.gpa, bytes);
305
+
306
+
var arena: std.heap.ArenaAllocator = .init(self.gpa);
307
+
defer arena.deinit();
308
+
304
309
var iter: FrameIterator = .{ .bytes = self.buffer.items };
305
310
while (try iter.next()) |data| {
311
+
defer _ = arena.reset(.retain_capacity);
306
312
switch (data) {
307
313
.text => |s| {
314
+
const event: Event = .{ .raw = s, .value = try json.parseFromSliceLeaky(
315
+
Event,
316
+
arena.allocator(),
317
+
s,
318
+
.{ .allocate = .alloc_if_needed },
319
+
) };
320
+
308
321
const task: ourio.Task = .{
309
322
.callback = self.ctx.cb,
310
323
.userdata = self.ctx.ptr,
311
324
.msg = self.ctx.msg,
312
-
.result = .{ .userbytes = s },
325
+
.result = .{ .userptr = &event },
313
326
};
314
327
try self.ctx.cb(io, task);
315
328
},
···
325
338
raw: []const u8,
326
339
value: json.Value,
327
340
328
-
pub const Kind = enum {
341
+
pub const KindEnum = enum {
329
342
commit,
330
343
identity,
331
344
account,
332
345
};
333
346
347
+
pub const Kind = union(KindEnum) {
348
+
commit: Commit,
349
+
identity: Identity,
350
+
account: Account,
351
+
};
352
+
334
353
pub fn did(self: Event) atproto.Did {
335
354
const value = self.value.object.get("did").?;
336
355
return atproto.Did.init(value.string) catch unreachable;
···
343
362
344
363
pub fn kind(self: Event) Kind {
345
364
const value = self.value.object.get("kind").?;
346
-
return std.meta.stringToEnum(Kind, value.string);
365
+
const k = std.meta.stringToEnum(KindEnum, value.string).?;
366
+
switch (k) {
367
+
.commit => return .{ .commit = self.value.object.get("commit").? },
368
+
.identity => return .{ .identity = self.value.object.get("identity").? },
369
+
.account => return .{ .account = self.value.object.get("account").? },
370
+
}
347
371
}
372
+
373
+
pub const Commit = struct {
374
+
value: json.Value,
375
+
376
+
pub const Operation = enum {
377
+
create,
378
+
update,
379
+
delete,
380
+
};
381
+
382
+
pub fn rev(self: Commit) []const u8 {
383
+
return self.value.object.get("rev").?;
384
+
}
385
+
386
+
pub fn operation(self: Commit) Operation {
387
+
const op = self.value.object.get("operation").?;
388
+
return std.meta.stringToEnum(Operation, op.string).?;
389
+
}
390
+
391
+
pub fn collection(self: Commit) []const u8 {
392
+
return self.value.object.get("collection").?.string;
393
+
}
394
+
395
+
pub fn rkey(self: Commit) []const u8 {
396
+
return self.value.object.get("rkey").?.string;
397
+
}
398
+
399
+
pub fn cid(self: Commit) []const u8 {
400
+
return self.value.object.get("cid").?.string;
401
+
}
402
+
403
+
pub fn record(self: Commit) json.Value {
404
+
return self.object.get("record").?;
405
+
}
406
+
};
407
+
408
+
pub const Identity = struct {
409
+
value: json.Value,
410
+
411
+
pub fn did(self: Identity) atproto.Did {
412
+
const value = self.value.object.get("did").?;
413
+
return atproto.Did.init(value.string) catch unreachable;
414
+
}
415
+
416
+
pub fn seq(self: Identity) i64 {
417
+
return self.value.object.get("seq").?.integer;
418
+
}
419
+
420
+
pub fn time(self: Identity) []const u8 {
421
+
return self.value.object.get("time").?.string;
422
+
}
423
+
424
+
pub fn handle(self: Identity) ?[]const u8 {
425
+
const value = self.value.object.get("handle") orelse return null;
426
+
return value.string;
427
+
}
428
+
};
429
+
430
+
pub const Account = struct {
431
+
value: json.Value,
432
+
433
+
pub fn active(self: Account) bool {
434
+
return self.value.object.get("active").?.bool;
435
+
}
436
+
437
+
pub fn did(self: Account) atproto.Did {
438
+
const value = self.value.object.get("did").?;
439
+
return atproto.Did.init(value.string) catch unreachable;
440
+
}
441
+
442
+
pub fn seq(self: Account) i64 {
443
+
return self.value.object.get("seq").?.integer;
444
+
}
445
+
446
+
pub fn time(self: Account) []const u8 {
447
+
return self.value.object.get("time").?.string;
448
+
}
449
+
};
348
450
};
349
451
350
452
const FrameIterator = struct {