Skip to content

Commit 0f8b47b

Browse files
francisbouvierfrancisbouvier
committed
Move MsgBuffer in it's own file for unit test purpose
Signed-off-by: Francis Bouvier <[email protected]>
1 parent 5eae158 commit 0f8b47b

File tree

3 files changed

+157
-152
lines changed

3 files changed

+157
-152
lines changed

‎src/msg.zig‎

Lines changed: 153 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
const std = @import("std");
2+
3+
// pub const MaxStdOutSize = 512; // ensure debug msg are not too long
4+
5+
/// MsgBuffer return messages from a raw text read stream,
6+
/// according to the following format `<msg_size>:<msg>`.
7+
/// It handles both:
8+
/// - combined messages in one read
9+
/// - single message in several read (multipart)
10+
/// It is safe (and good practice) to reuse the same MsgBuffer
11+
/// on several reads of the same stream.
12+
pub const MsgBuffer = struct {
13+
size: usize = 0,
14+
buf: []u8,
15+
pos: usize = 0,
16+
17+
pub fn init(alloc: std.mem.Allocator, size: usize) std.mem.Allocator.Error!MsgBuffer {
18+
const buf = try alloc.alloc(u8, size);
19+
return .{ .buf = buf };
20+
}
21+
22+
pub fn deinit(self: MsgBuffer, alloc: std.mem.Allocator) void {
23+
alloc.free(self.buf);
24+
}
25+
26+
fn isFinished(self: *MsgBuffer) bool {
27+
return self.pos >= self.size;
28+
}
29+
30+
fn isEmpty(self: MsgBuffer) bool {
31+
return self.size == 0 and self.pos == 0;
32+
}
33+
34+
fn reset(self: *MsgBuffer) void {
35+
self.size = 0;
36+
self.pos = 0;
37+
}
38+
39+
// read input
40+
// - `do_func` is a callback to execute on each message of the input
41+
// - `data` is a arbitrary payload that will be passed to the callback along with
42+
// the message itself
43+
pub fn read(
44+
self: *MsgBuffer,
45+
alloc: std.mem.Allocator,
46+
input: []const u8,
47+
data: anytype,
48+
comptime do_func: fn (data: @TypeOf(data), msg: []const u8) anyerror!void,
49+
) !void {
50+
var _input = input; // make input writable
51+
52+
while (true) {
53+
var msg: []const u8 = undefined;
54+
55+
// msg size
56+
var msg_size: usize = undefined;
57+
if (self.isEmpty()) {
58+
// parse msg size metadata
59+
const size_pos = std.mem.indexOfScalar(u8, _input, ':').?;
60+
const size_str = _input[0..size_pos];
61+
msg_size = try std.fmt.parseInt(u32, size_str, 10);
62+
_input = _input[size_pos + 1 ..];
63+
} else {
64+
msg_size = self.size;
65+
}
66+
67+
// multipart
68+
const is_multipart = !self.isEmpty() or _input.len < msg_size;
69+
if (is_multipart) {
70+
71+
// set msg size on empty MsgBuffer
72+
if (self.isEmpty()) {
73+
self.size = msg_size;
74+
}
75+
76+
// get the new position of the cursor
77+
const new_pos = self.pos + _input.len;
78+
79+
// check if the current input can fit in MsgBuffer
80+
if (new_pos > self.buf.len) {
81+
// max_size is the max between msg size and current new cursor position
82+
const max_size = @max(self.size, new_pos);
83+
// resize the MsgBuffer to fit
84+
self.buf = try alloc.realloc(self.buf, max_size);
85+
}
86+
87+
// copy the current input into MsgBuffer
88+
@memcpy(self.buf[self.pos..new_pos], _input[0..]);
89+
90+
// set the new cursor position
91+
self.pos = new_pos;
92+
93+
// if multipart is not finished, go fetch the next input
94+
if (!self.isFinished()) return;
95+
96+
// otherwhise multipart is finished, use its buffer as input
97+
_input = self.buf[0..self.pos];
98+
self.reset();
99+
}
100+
101+
// handle several JSON msg in 1 read
102+
const is_combined = _input.len > msg_size;
103+
msg = _input[0..msg_size];
104+
if (is_combined) {
105+
_input = _input[msg_size..];
106+
}
107+
108+
try @call(.auto, do_func, .{ data, msg });
109+
110+
if (!is_combined) break;
111+
}
112+
}
113+
};
114+
115+
fn doTest(nb: *u8, _: []const u8) anyerror!void {
116+
nb.* += 1;
117+
}
118+
119+
test "MsgBuffer" {
120+
const Case = struct {
121+
input: []const u8,
122+
nb: u8,
123+
};
124+
const alloc = std.testing.allocator;
125+
const cases = [_]Case{
126+
// simple
127+
.{ .input = "2:ok", .nb = 1 },
128+
// combined
129+
.{ .input = "2:ok3:foo7:bar2:ok", .nb = 3 }, // "bar2:ok" is a message, no need to escape "2:" here
130+
// multipart
131+
.{ .input = "9:multi", .nb = 0 },
132+
.{ .input = "part", .nb = 1 },
133+
// multipart & combined
134+
.{ .input = "9:multi", .nb = 0 },
135+
.{ .input = "part2:ok", .nb = 2 },
136+
// several multipart
137+
.{ .input = "23:multi", .nb = 0 },
138+
.{ .input = "several", .nb = 0 },
139+
.{ .input = "complex", .nb = 0 },
140+
.{ .input = "part", .nb = 1 },
141+
// combined & multipart
142+
.{ .input = "2:ok9:multi", .nb = 1 },
143+
.{ .input = "part", .nb = 1 },
144+
};
145+
var nb: u8 = undefined;
146+
var msg_buf = try MsgBuffer.init(alloc, 10);
147+
defer msg_buf.deinit(alloc);
148+
for (cases) |case| {
149+
nb = 0;
150+
try msg_buf.read(alloc, case.input, &nb, doTest);
151+
try std.testing.expect(nb == case.nb);
152+
}
153+
}

‎src/run_tests.zig‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -295,6 +295,9 @@ const kb = 1024;
295295
const ms = std.time.ns_per_ms;
296296

297297
test {
298+
const msgTest = @import("msg.zig");
299+
std.testing.refAllDecls(msgTest);
300+
298301
const asyncTest = @import("async/test.zig");
299302
std.testing.refAllDecls(asyncTest);
300303

‎src/server.zig‎

Lines changed: 1 addition & 152 deletions
Original file line numberDiff line numberDiff line change
@@ -7,8 +7,8 @@ const RecvError = public.IO.RecvError;
77
const SendError = public.IO.SendError;
88
const TimeoutError = public.IO.TimeoutError;
99

10+
const MsgBuffer = @import("msg.zig").MsgBuffer;
1011
const Browser = @import("browser/browser.zig").Browser;
11-
1212
const cdp = @import("cdp/cdp.zig");
1313

1414
const NoError = error{NoError};
@@ -91,157 +91,6 @@ pub const Cmd = struct {
9191
}
9292
};
9393

94-
/// MsgBuffer return messages from a raw text read stream,
95-
/// according to the following format `<msg_size>:<msg>`.
96-
/// It handles both:
97-
/// - combined messages in one read
98-
/// - single message in several read (multipart)
99-
/// It is safe (and good practice) to reuse the same MsgBuffer
100-
/// on several reads of the same stream.
101-
const MsgBuffer = struct {
102-
size: usize = 0,
103-
buf: []u8,
104-
pos: usize = 0,
105-
106-
fn init(alloc: std.mem.Allocator, size: usize) std.mem.Allocator.Error!MsgBuffer {
107-
const buf = try alloc.alloc(u8, size);
108-
return .{ .buf = buf };
109-
}
110-
111-
fn deinit(self: MsgBuffer, alloc: std.mem.Allocator) void {
112-
alloc.free(self.buf);
113-
}
114-
115-
fn isFinished(self: *MsgBuffer) bool {
116-
return self.pos >= self.size;
117-
}
118-
119-
fn isEmpty(self: MsgBuffer) bool {
120-
return self.size == 0 and self.pos == 0;
121-
}
122-
123-
fn reset(self: *MsgBuffer) void {
124-
self.size = 0;
125-
self.pos = 0;
126-
}
127-
128-
// read input
129-
// - `do_func` is a callback to execute on each message of the input
130-
// - `data` is a arbitrary payload that will be passed to the callback along with
131-
// the message itself
132-
fn read(
133-
self: *MsgBuffer,
134-
alloc: std.mem.Allocator,
135-
input: []const u8,
136-
data: anytype,
137-
comptime do_func: fn (data: @TypeOf(data), msg: []const u8) anyerror!void,
138-
) !void {
139-
var _input = input; // make input writable
140-
141-
while (true) {
142-
var msg: []const u8 = undefined;
143-
144-
// msg size
145-
var msg_size: usize = undefined;
146-
if (self.isEmpty()) {
147-
// parse msg size metadata
148-
const size_pos = std.mem.indexOfScalar(u8, _input, ':').?;
149-
const size_str = _input[0..size_pos];
150-
msg_size = try std.fmt.parseInt(u32, size_str, 10);
151-
_input = _input[size_pos + 1 ..];
152-
} else {
153-
msg_size = self.size;
154-
}
155-
156-
// multipart
157-
const is_multipart = !self.isEmpty() or _input.len < msg_size;
158-
if (is_multipart) {
159-
160-
// set msg size on empty MsgBuffer
161-
if (self.isEmpty()) {
162-
self.size = msg_size;
163-
}
164-
165-
// get the new position of the cursor
166-
const new_pos = self.pos + _input.len;
167-
168-
// check if the current input can fit in MsgBuffer
169-
if (new_pos > self.buf.len) {
170-
// max_size is the max between msg size and current new cursor position
171-
const max_size = @max(self.size, new_pos);
172-
// resize the MsgBuffer to fit
173-
self.buf = try alloc.realloc(self.buf, max_size);
174-
}
175-
176-
// copy the current input into MsgBuffer
177-
@memcpy(self.buf[self.pos..new_pos], _input[0..]);
178-
179-
// set the new cursor position
180-
self.pos = new_pos;
181-
182-
// if multipart is not finished, go fetch the next input
183-
if (!self.isFinished()) return;
184-
185-
// otherwhise multipart is finished, use its buffer as input
186-
_input = self.buf[0..self.pos];
187-
self.reset();
188-
}
189-
190-
// handle several JSON msg in 1 read
191-
const is_combined = _input.len > msg_size;
192-
msg = _input[0..msg_size];
193-
std.log.debug("msg: {s}", .{msg[0..@min(MaxStdOutSize, msg_size)]});
194-
if (is_combined) {
195-
_input = _input[msg_size..];
196-
}
197-
198-
try @call(.auto, do_func, .{ data, msg });
199-
200-
if (!is_combined) break;
201-
}
202-
}
203-
};
204-
205-
fn doTest(nb: *u8, _: []const u8) anyerror!void {
206-
nb.* += 1;
207-
}
208-
209-
test "MsgBuffer" {
210-
const Case = struct {
211-
input: []const u8,
212-
nb: u8,
213-
};
214-
const alloc = std.testing.allocator;
215-
const cases = [_]Case{
216-
// simple
217-
.{ .input = "2:ok", .nb = 1 },
218-
// combined
219-
.{ .input = "2:ok3:foo7:bar2:ok", .nb = 3 }, // "bar2:ok" is a message, no need to escape "2:" here
220-
// multipart
221-
.{ .input = "9:multi", .nb = 0 },
222-
.{ .input = "part", .nb = 1 },
223-
// multipart & combined
224-
.{ .input = "9:multi", .nb = 0 },
225-
.{ .input = "part2:ok", .nb = 2 },
226-
// several multipart
227-
.{ .input = "23:multi", .nb = 0 },
228-
.{ .input = "several", .nb = 0 },
229-
.{ .input = "complex", .nb = 0 },
230-
.{ .input = "part", .nb = 1 },
231-
// combined & multipart
232-
.{ .input = "2:ok9:multi", .nb = 1 },
233-
.{ .input = "part", .nb = 1 },
234-
};
235-
var nb: u8 = undefined;
236-
var msg_buf = try MsgBuffer.init(alloc, 10);
237-
defer msg_buf.deinit(alloc);
238-
for (cases) |case| {
239-
nb = 0;
240-
try msg_buf.read(alloc, case.input, &nb, doTest);
241-
try std.testing.expect(nb == case.nb);
242-
}
243-
}
244-
24594
// I/O Send
24695
// --------
24796

0 commit comments

Comments
 (0)