1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
const std = @import("std");
/// 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 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 };
}
/// Feed a chunk of raw bytes. Returns a list of complete SSE events
/// 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) = .empty;
errdefer {
for (events.items) |ev| self.allocator.free(ev);
events.deinit(self.allocator);
}
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];
// 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) 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.
line_start = newline + 1;
}
if (has_data) {
try events.append(self.allocator, try payload.toOwnedSlice(self.allocator));
} else {
payload.deinit(self.allocator);
}
scan_pos = delim + 2;
}
// 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(self.allocator);
}
/// Free a list of events returned by `feed`.
pub fn freeEvents(self: *SSEParser, events: []Event) void {
for (events) |ev| self.allocator.free(ev);
self.allocator.free(events);
}
pub fn deinit(self: *SSEParser) void {
self.buf.deinit(self.allocator);
}
};
// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------
const testing = std.testing;
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 testing.expectEqual(@as(usize, 1), events.len);
try testing.expectEqualStrings("hello", events[0]);
}
test "SSEParser - partial then complete" {
var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
const events1 = try parser.feed("data: hel");
defer parser.freeEvents(events1);
try testing.expectEqual(@as(usize, 0), events1.len);
const events2 = try parser.feed("lo\n\n");
defer parser.freeEvents(events2);
try testing.expectEqual(@as(usize, 1), events2.len);
try testing.expectEqualStrings("hello", events2[0]);
}
test "SSEParser - multiple events in single chunk" {
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 testing.expectEqual(@as(usize, 2), events.len);
try testing.expectEqualStrings("one", events[0]);
try testing.expectEqualStrings("two", events[1]);
}
test "SSEParser - data DONE signal" {
var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
const events = try parser.feed("data: [DONE]\n\n");
defer parser.freeEvents(events);
try testing.expectEqual(@as(usize, 1), events.len);
try testing.expectEqualStrings("[DONE]", events[0]);
}
test "SSEParser - empty event (keep-alive)" {
var parser = SSEParser.init(testing.allocator);
defer parser.deinit();
const events = try parser.feed("\n\n");
defer parser.freeEvents(events);
try testing.expectEqual(@as(usize, 0), events.len);
}
test "SSEParser - ignores non-data fields" {
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 testing.expectEqual(@as(usize, 1), events.len);
try testing.expectEqualStrings("payload", events[0]);
}
test "SSEParser - multi-line data joined with newline" {
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 testing.expectEqual(@as(usize, 1), events.len);
try testing.expectEqualStrings("line1\nline2", events[0]);
}
|
