diff options
Diffstat (limited to 'libpanto/src/sse.zig')
| -rw-r--r-- | libpanto/src/sse.zig | 137 |
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]); } |
