diff options
| -rw-r--r-- | src/Server.zig | 37 |
1 files changed, 19 insertions, 18 deletions
diff --git a/src/Server.zig b/src/Server.zig index cfd9d97..f0ee1fe 100644 --- a/src/Server.zig +++ b/src/Server.zig @@ -345,43 +345,44 @@ fn publishMessage( // The chunk_count starts off at the minimum number of chunks per message, and // then increases as branches add additional chunks. // The msg_chunks_buf.len is the maximum number of chunks in a message. - // Each of the appendBounded calls has their error marked as unreachable, - // because it is an error for there to be more appendBounded calls than chunks - // in the chunks buf. + // We can appendAssumeCapacity because it is a programmer error to append + // more than max_msg_chunks. + // If we need to append more chunks, this value should be increased. // The reason for this strategy is to avoid any intermediary allocations between // the publishers read buffer, and the subscribers write buffer. - var chunk_count: usize = 7; - var msg_chunks_buf: [10][]const u8 = undefined; + const min_msg_chunks, const max_msg_chunks = .{ 7, 10 }; + var chunk_count: usize = min_msg_chunks; + var msg_chunks_buf: [max_msg_chunks][]const u8 = undefined; var msg_chunks: ArrayList([]const u8) = .initBuffer(&msg_chunks_buf); switch (pub_or_hpub) { - .@"pub" => _ = msg_chunks.appendBounded("MSG ") catch unreachable, - .hpub => _ = msg_chunks.appendBounded("HMSG ") catch unreachable, + .@"pub" => _ = msg_chunks.appendAssumeCapacity("MSG "), + .hpub => _ = msg_chunks.appendAssumeCapacity("HMSG "), } - msg_chunks.appendBounded(msg.subject) catch unreachable; - msg_chunks.appendBounded(" ") catch unreachable; - msg_chunks.appendBounded(subscription.sid) catch unreachable; - msg_chunks.appendBounded(" ") catch unreachable; + msg_chunks.appendAssumeCapacity(msg.subject); + msg_chunks.appendAssumeCapacity(" "); + msg_chunks.appendAssumeCapacity(subscription.sid); + msg_chunks.appendAssumeCapacity(" "); if (msg.reply_to) |reply_to| { chunk_count += 2; - msg_chunks.appendBounded(reply_to) catch unreachable; - msg_chunks.appendBounded(" ") catch unreachable; + msg_chunks.appendAssumeCapacity(reply_to); + msg_chunks.appendAssumeCapacity(" "); } switch (pub_or_hpub) { .hpub => { chunk_count += 1; var hlen_buf: [std.fmt.count("{d} ", .{std.math.maxInt(usize)})]u8 = undefined; - msg_chunks.appendBounded( + msg_chunks.appendAssumeCapacity( std.fmt.bufPrint(&hlen_buf, "{d} ", .{hpubmsg.header_bytes}) catch unreachable, - ) catch unreachable; + ); }, else => {}, } var len_buf: [std.fmt.count("{d}\r\n", .{std.math.maxInt(usize)})]u8 = undefined; - msg_chunks.appendBounded( + msg_chunks.appendAssumeCapacity( std.fmt.bufPrint(&len_buf, "{d}\r\n", .{msg.payload.len - 2}) catch unreachable, - ) catch unreachable; - msg_chunks.appendBounded(msg.payload) catch unreachable; + ); + msg_chunks.appendAssumeCapacity(msg.payload); subscription.send(io, msg_chunks.items[0..chunk_count]) catch |err| switch (err) { error.Closed => {}, |