diff options
Diffstat (limited to 'packet_reader')
| -rw-r--r-- | packet_reader/packet_reader.ha | 94 |
1 files changed, 66 insertions, 28 deletions
diff --git a/packet_reader/packet_reader.ha b/packet_reader/packet_reader.ha index d541a11..ae38be7 100644 --- a/packet_reader/packet_reader.ha +++ b/packet_reader/packet_reader.ha @@ -4,7 +4,7 @@ use endian; use drawing; use drawing::{pos,CHUNKSIZE}; -export def VERSION: u8 = 2; +export def VERSION: u8 = 3; export type error = !str; @@ -13,12 +13,14 @@ export type packet_reader = struct { good: []u8, wbuf: []u8, wgood: []u8, + sock: io::file, }; -export fn new() packet_reader = { +export fn new(sock: io::file) packet_reader = { let pr = packet_reader { buf = alloc([0...],512*512*4*2), // ehhh wbuf = alloc([0...], 512*512*4*2), // ehhhh + sock = sock, ... }; pr.good = pr.buf[0..0]; @@ -26,6 +28,11 @@ export fn new() packet_reader = { return pr; }; +export fn finish(pr: *packet_reader) void = { + free(pr.buf); + free(pr.wbuf); +}; + fn cast_u32s_to_u8s(in: []u32) []u8 = (in: *[*]u32: *[*]u8)[..len(in)*4]; fn cast_u8s_to_u32s(in: []u8) []u32 = @@ -52,24 +59,51 @@ export type packet = ( | packet_position | packet_reqchunk); -// call when input is ready. could block otherwise -export fn read(pr: *packet_reader, sock: io::handle) (void | io::error | io::EOF) = { - const remaining_amt = len(pr.good); - const read_pos = if (remaining_amt > 0) { - // still some unconsumed content in the buffer - // move unconsumed stuff to start of buffer - // fmt::printfln("moving {} remaining bytes",remaining_amt)!; - pr.buf[0..remaining_amt] = pr.good; - yield remaining_amt; - } else 0z; - const nread = match(io::read(sock, pr.buf[read_pos..])?) { - case io::EOF => return io::EOF; - case let n: size => yield n; +// call when input ready, so it doesn't block +export fn service( + pr: *packet_reader, + should_read: bool, + should_write: bool, +) (void | io::error | io::EOF) = { + const sock = pr.sock; + + if (should_read) { + // reading + const remaining_amt = len(pr.good); + const read_pos = if (remaining_amt > 0) { + // still some unconsumed content in the buffer + // move unconsumed stuff to start of buffer + // fmt::printfln("moving {} remaining bytes",remaining_amt)!; + pr.buf[0..remaining_amt] = pr.good; + yield remaining_amt; + } else 0z; + const nread = match(io::read(sock, pr.buf[read_pos..])?) { + case io::EOF => return io::EOF; + case let n: size => yield n; + }; + // fmt::printfln("read {} bytes",nread)!; + const total_amt = read_pos + nread; + pr.good = pr.buf[0..total_amt]; + // fmt::printfln("now {} bytes in buffer",total_amt)!; + }; + + + if (should_write) { + // writing + const amt_to_write = len(pr.wgood); + if (amt_to_write == 0) return; + const nwritten = io::write(sock, pr.wgood)?; + // fmt::printfln("written {} of {} bytes", nwritten, amt_to_write)!; + if (nwritten < amt_to_write) { + // fmt::println("moving buffer back")!; + const remaining_len = amt_to_write - nwritten; + pr.wbuf[0..remaining_len] = pr.wgood[nwritten..]; + pr.wgood = pr.wbuf[0..remaining_len]; + } else { + pr.wgood = pr.wbuf[0..0]; + }; + // fmt::printfln(" {} to write next time", len(pr.wgood))!; }; - // fmt::printfln("read {} bytes",nread)!; - const total_amt = read_pos + nread; - pr.good = pr.buf[0..total_amt]; - // fmt::printfln("now {} bytes in buffer",total_amt)!; }; // packet format: @@ -78,7 +112,7 @@ export fn read(pr: *packet_reader, sock: io::handle) (void | io::error | io::EOF // [size-8]u8 data... // size includes size of header (size and type fields) -export fn next(pr: *packet_reader) (packet | done | error) = { +export fn read_next(pr: *packet_reader) (packet | done | error) = { // either parse a full packet out of the front of good, // move good along that many bytes, // and return the packet, @@ -136,24 +170,28 @@ export fn next(pr: *packet_reader) (packet | done | error) = { }; }; -export fn send_raw(sock: io::file, ty: packet_type, datas: []u8...) (void | io::error) = { +fn append_wbuf(pr: *packet_reader, ty: packet_type, datas: []u8...) (void | error) = { // ehh const header: [8]u8 = [0...]; let total_len = 8u32; for (const data .. datas) total_len += len(data): u32; + + const remaining_space = len(pr.wbuf) - len(pr.wgood); + if (remaining_space < total_len) return "wbuff full": error; + endian::leputu32(header[0..4], total_len); endian::leputu32(header[4..8], ty); - io::writeall(sock, header)?; + static append(pr.wgood, header...); for (const data .. datas) - io::writeall(sock, data)?; + static append(pr.wgood, data...); }; -export fn send(sock: io::file, packet: packet) (void | io::error) = { +export fn write(pr: *packet_reader, packet: packet) (void | error) = { match (packet) { case let op: packet_drawop => const ser_op = drawing::ser_op(op); defer free(ser_op); - send_raw(sock, packet_type::DRAW_OP, ser_op)?; + append_wbuf(pr, packet_type::DRAW_OP, ser_op)?; case let packet: packet_sendchunk => const pos_buf: [8]u8 = [0...]; endian::leputu32(pos_buf[0..4],packet.world_pos.0: u32); @@ -161,16 +199,16 @@ export fn send(sock: io::file, packet: packet) (void | io::error) = { const chunk_data_compressed = rle_encode(packet.chunk_data); defer free(chunk_data_compressed); const compressed_data_bytes = cast_u32s_to_u8s(chunk_data_compressed); - send_raw(sock, packet_type::SEND_CHUNK, pos_buf, compressed_data_bytes)?; + append_wbuf(pr, packet_type::SEND_CHUNK, pos_buf, compressed_data_bytes)?; case let pos: packet_position => const pos_buf: [8]u8 = [0...]; endian::leputu32(pos_buf[0..4], pos.0: u32); endian::leputu32(pos_buf[4..8], pos.1: u32); - send_raw(sock, packet_type::POSITION, pos_buf)?; + append_wbuf(pr, packet_type::POSITION, pos_buf)?; case let world_pos: packet_reqchunk => const pos_buf: [8]u8 = [0...]; endian::leputu32(pos_buf[0..4], world_pos.0: u32); endian::leputu32(pos_buf[4..8], world_pos.1: u32); - send_raw(sock, packet_type::REQ_CHUNK, pos_buf)?; + append_wbuf(pr, packet_type::REQ_CHUNK, pos_buf)?; }; }; |