summaryrefslogtreecommitdiff
path: root/libpanto/src/sse.zig
diff options
context:
space:
mode:
Diffstat (limited to 'libpanto/src/sse.zig')
-rw-r--r--libpanto/src/sse.zig137
1 files changed, 54 insertions, 83 deletions
diff --git a/libpanto/src/sse.zig b/libpanto/src/sse.zig
index 150c3ea..c54d532 100644
--- a/libpanto/src/sse.zig
+++ b/libpanto/src/sse.zig
@@ -1,99 +1,81 @@
const std = @import("std");
-/// Parsed SSE event data — the content after "data: " (prefix removed).
-/// For multi-line data events, lines are joined with newlines per the SSE spec.
-///
-/// Borrowed from the parser's internal buffer. Call `SSEParser.freeEvents()`
-/// to free the events list when done (event data itself is borrowed and
-/// does not need individual freeing).
-const Event = []const u8;
+/// A parsed SSE event payload: the concatenated `data:` content.
+/// Owned by the caller; free with `freeEvents`.
+const Event = []u8;
/// Incremental SSE line parser.
///
/// The HTTP client delivers arbitrary-sized read buffers; this module
-/// reassembles them into complete `data: ...\n\n` events.
-///
-/// Event data is borrowed from the internal buffer. Call `freeEvents()` to
-/// free the events list returned by `feed`; the event data itself is borrowed
-/// from the parser's buffer and does not need individual freeing.
+/// reassembles them into complete `data: ...\n\n` events and returns
+/// owned event payload slices.
pub const SSEParser = struct {
buf: std.ArrayList(u8) = .empty,
allocator: std.mem.Allocator,
pub fn init(allocator: std.mem.Allocator) SSEParser {
- return .{
- .allocator = allocator,
- };
+ return .{ .allocator = allocator };
}
/// Feed a chunk of raw bytes. Returns a list of complete SSE events
- /// found in the buffer. Returns an empty list if no complete events
- /// are available yet.
- ///
- /// Event data is borrowed from the internal buffer. Call `freeEvents()`
- /// on the returned list when done.
+ /// found in the buffer (may be empty). The caller owns the returned
+ /// slice and every event within; free with `freeEvents`.
pub fn feed(self: *SSEParser, chunk: []const u8) ![]Event {
try self.buf.appendSlice(self.allocator, chunk);
- var events = std.ArrayList(Event).init(self.allocator);
- errdefer events.deinit();
+ var events: std.ArrayList(Event) = .empty;
+ errdefer {
+ for (events.items) |ev| self.allocator.free(ev);
+ events.deinit(self.allocator);
+ }
- var write_pos: usize = 0;
var scan_pos: usize = 0;
-
while (std.mem.indexOf(u8, self.buf.items[scan_pos..], "\n\n")) |rel_delim| {
const delim = scan_pos + rel_delim;
const event_bytes = self.buf.items[scan_pos..delim];
- // Compact data lines in-place. Compacted data is always shorter
- // than the original event bytes (we strip at least "data: " per
- // line), so write_pos <= scan_pos and we never overwrite data
- // we haven't yet read.
- const data_start = write_pos;
+ // Collect data: lines into an owned payload string.
+ var payload: std.ArrayList(u8) = .empty;
+ errdefer payload.deinit(self.allocator);
+
var line_start: usize = 0;
var has_data = false;
-
while (line_start < event_bytes.len) {
const newline = std.mem.indexOfScalarPos(u8, event_bytes, line_start, '\n') orelse event_bytes.len;
const line = event_bytes[line_start..newline];
if (std.mem.startsWith(u8, line, "data: ")) {
- if (has_data) {
- self.buf.items[write_pos] = '\n';
- write_pos += 1;
- }
- const content = line["data: ".len..];
- std.mem.copyForwards(u8, self.buf.items[write_pos .. write_pos + content.len], content);
- write_pos += content.len;
+ if (has_data) try payload.append(self.allocator, '\n');
+ try payload.appendSlice(self.allocator, line["data: ".len..]);
has_data = true;
}
- // Ignore other SSE fields (event:, id:, retry:) and comments (starting with :)
+ // Ignore other SSE fields (event:, id:, retry:) and comments.
line_start = newline + 1;
}
if (has_data) {
- try events.append(self.buf.items[data_start..write_pos]);
+ try events.append(self.allocator, try payload.toOwnedSlice(self.allocator));
+ } else {
+ payload.deinit(self.allocator);
}
scan_pos = delim + 2;
}
- // Shift remaining (incomplete) bytes to the front, retaining
- // the backing allocation for reuse.
+ // Drop processed bytes; retain the buffer allocation.
const remaining_len = self.buf.items.len - scan_pos;
if (remaining_len > 0 and scan_pos > 0) {
std.mem.copyForwards(u8, self.buf.items[0..remaining_len], self.buf.items[scan_pos..]);
}
self.buf.shrinkRetainingCapacity(remaining_len);
- return events.toOwnedSlice();
+ return events.toOwnedSlice(self.allocator);
}
/// Free a list of events returned by `feed`.
- /// Event data is borrowed from the parser's buffer and does not need
- /// to be freed individually; only the events array itself is freed.
pub fn freeEvents(self: *SSEParser, events: []Event) void {
+ for (events) |ev| self.allocator.free(ev);
self.allocator.free(events);
}
@@ -102,98 +84,87 @@ pub const SSEParser = struct {
}
};
-test "SSEParser - single complete event" {
- const allocator = std.testing.allocator;
+// -----------------------------------------------------------------------------
+// Tests
+// -----------------------------------------------------------------------------
+
+const testing = std.testing;
- var parser = SSEParser.init(allocator);
+test "SSEParser - single complete event" {
+ var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
const events = try parser.feed("data: hello\n\n");
defer parser.freeEvents(events);
- try std.testing.expectEqual(@as(usize, 1), events.len);
- try std.testing.expectEqualStrings("hello", events[0]);
+ try testing.expectEqual(@as(usize, 1), events.len);
+ try testing.expectEqualStrings("hello", events[0]);
}
test "SSEParser - partial then complete" {
- const allocator = std.testing.allocator;
-
- var parser = SSEParser.init(allocator);
+ var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
- // Feed partial — no complete event yet
const events1 = try parser.feed("data: hel");
defer parser.freeEvents(events1);
- try std.testing.expectEqual(@as(usize, 0), events1.len);
+ try testing.expectEqual(@as(usize, 0), events1.len);
- // Feed remainder — completes the event
const events2 = try parser.feed("lo\n\n");
defer parser.freeEvents(events2);
- try std.testing.expectEqual(@as(usize, 1), events2.len);
- try std.testing.expectEqualStrings("hello", events2[0]);
+ try testing.expectEqual(@as(usize, 1), events2.len);
+ try testing.expectEqualStrings("hello", events2[0]);
}
test "SSEParser - multiple events in single chunk" {
- const allocator = std.testing.allocator;
-
- var parser = SSEParser.init(allocator);
+ var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
const events = try parser.feed("data: one\n\ndata: two\n\n");
defer parser.freeEvents(events);
- try std.testing.expectEqual(@as(usize, 2), events.len);
- try std.testing.expectEqualStrings("one", events[0]);
- try std.testing.expectEqualStrings("two", events[1]);
+ try testing.expectEqual(@as(usize, 2), events.len);
+ try testing.expectEqualStrings("one", events[0]);
+ try testing.expectEqualStrings("two", events[1]);
}
test "SSEParser - data DONE signal" {
- const allocator = std.testing.allocator;
-
- var parser = SSEParser.init(allocator);
+ var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
const events = try parser.feed("data: [DONE]\n\n");
defer parser.freeEvents(events);
- try std.testing.expectEqual(@as(usize, 1), events.len);
- try std.testing.expectEqualStrings("[DONE]", events[0]);
+ try testing.expectEqual(@as(usize, 1), events.len);
+ try testing.expectEqualStrings("[DONE]", events[0]);
}
test "SSEParser - empty event (keep-alive)" {
- const allocator = std.testing.allocator;
-
- var parser = SSEParser.init(allocator);
+ var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
- // A blank line (just \n\n) is a keep-alive comment with no data
const events = try parser.feed("\n\n");
defer parser.freeEvents(events);
- try std.testing.expectEqual(@as(usize, 0), events.len);
+ try testing.expectEqual(@as(usize, 0), events.len);
}
test "SSEParser - ignores non-data fields" {
- const allocator = std.testing.allocator;
-
- var parser = SSEParser.init(allocator);
+ var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
const events = try parser.feed("event: message\ndata: payload\nid: 42\n\n");
defer parser.freeEvents(events);
- try std.testing.expectEqual(@as(usize, 1), events.len);
- try std.testing.expectEqualStrings("payload", events[0]);
+ try testing.expectEqual(@as(usize, 1), events.len);
+ try testing.expectEqualStrings("payload", events[0]);
}
test "SSEParser - multi-line data joined with newline" {
- const allocator = std.testing.allocator;
-
- var parser = SSEParser.init(allocator);
+ var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
const events = try parser.feed("data: line1\ndata: line2\n\n");
defer parser.freeEvents(events);
- try std.testing.expectEqual(@as(usize, 1), events.len);
- try std.testing.expectEqualStrings("line1\nline2", events[0]);
+ try testing.expectEqual(@as(usize, 1), events.len);
+ try testing.expectEqualStrings("line1\nline2", events[0]);
}