diff options
| -rw-r--r-- | src/Server.zig | 6 | ||||
| -rw-r--r-- | src/Server/Client.zig | 10 |
2 files changed, 15 insertions, 1 deletions
diff --git a/src/Server.zig b/src/Server.zig index e20f5c0..49f54e2 100644 --- a/src/Server.zig +++ b/src/Server.zig @@ -47,6 +47,7 @@ const Subscription = struct { // would put an invalid set series of bytes in the receivers queue. _ = try self.queue.putUncancelable(io, chunk, chunk.len); } + try io.checkCancel(); } }; @@ -382,7 +383,10 @@ fn publishMessage( ) catch unreachable; msg_chunks.appendBounded(msg.payload) catch unreachable; - try subscription.send(io, msg_chunks.items[0..chunk_count]); + subscription.send(io, msg_chunks.items[0..chunk_count]) catch |err| switch (err) { + error.Closed => {}, + error.Canceled => |e| return e, + }; } } diff --git a/src/Server/Client.zig b/src/Server/Client.zig index 77034fd..6ad1804 100644 --- a/src/Server/Client.zig +++ b/src/Server/Client.zig @@ -40,6 +40,16 @@ pub fn start(self: *Client, io: std.Io) !void { std.debug.assert(self.to_client.end == 0); while (true) { self.to_client.end = try self.recv_queue.get(io, self.to_client.buffer, 1); + // Wait 1 nanosecond to see if more data is in the queue. + // If there is, add it to the write buffer before sending it. + // The reason for this is because if we send the first chunk as soon as we get it, + // we will likely be sending a partial message, which will end up being way slower. + try io.sleep(.fromNanoseconds(1), .awake); + self.to_client.end += try self.recv_queue.get( + io, + self.to_client.buffer[self.to_client.end..], + 0, + ); try self.to_client.flush(); } } |