io_interface: Mock netListenIp, netAccept, netConnectIp, netSend, netRead, netWrite

src/io_interface.zig

@@ -1079,32 +1079,97 @@ fn randomSecure(userdata: ?*anyopaque, buffer: []u8) RandomSecureError!void {
     @panic("Not implemented yet");
 }
 
+fn nodeFromUserdata(userdata: ?*anyopaque) *Node {
+    return @ptrCast(@alignCast(userdata.?));
+}
+
+fn ipAddressToNodeAddress(address: *const net.IpAddress) ?Node.Address {
+    return switch (address.*) {
+        .ip4 => |ip4| .{
+            .ipv4 = std.mem.readInt(u32, &ip4.bytes, .big),
+            .port = ip4.port,
+        },
+        .ip6 => null,
+    };
+}
+
+fn nodeAddressToIpAddress(address: Node.Address) net.IpAddress {
+    var bytes: [4]u8 = undefined;
+    std.mem.writeInt(u32, &bytes, address.ipv4, .big);
+    return .{ .ip4 = .{
+        .bytes = bytes,
+        .port = address.port,
+    } };
+}
+
 fn netListenIp(userdata: ?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Socket {
-    _ = userdata;
+    const node = nodeFromUserdata(userdata);
-    _ = address;
+
-    _ = options;
+    if (options.mode != .stream)
-    @panic("Not implemented yet");
+        return error.SocketModeUnsupported;
+    if (options.protocol != .tcp)
+        return error.ProtocolUnsupportedBySystem;
+
+    const node_address = ipAddressToNodeAddress(address) orelse return error.AddressFamilyUnsupported;
+    const handle = node.listen(node_address) catch |err| switch (err) {
+        error.DescriptorLimit => return error.ProcessFdQuotaExceeded,
+        error.AddressNotAvailable => return error.AddressUnavailable,
+        error.AddressAlreadyUsed => return error.AddressInUse,
+        error.OutOfMemory => return error.SystemResources,
+    };
+
+    return .{
+        .handle = handle,
+        .address = nodeAddressToIpAddress(node_address),
+    };
 }
 
 fn netAccept(userdata: ?*anyopaque, listen_handle: net.Socket.Handle, options: net.Server.AcceptOptions) net.Server.AcceptError!net.Socket {
-    _ = userdata;
+    const node = nodeFromUserdata(userdata);
-    _ = listen_handle;
     _ = options;
-    @panic("Not implemented yet");
+
+    const handle = node.accept(listen_handle) catch |err| switch (err) {
+        error.DescriptorLimit => return error.ProcessFdQuotaExceeded,
+        error.InvalidHandle => return error.SocketNotListening,
+        error.AcceptQueueEmpty => return error.WouldBlock,
+        error.OutOfMemory => return error.SystemResources,
+    };
+
+    return .{
+        .handle = handle,
+        .address = .{ .ip4 = .unspecified(0) },
+    };
 }
 
 fn netBindIp(userdata: ?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket {
     _ = userdata;
     _ = address;
     _ = options;
-    @panic("Not implemented yet");
+    return error.OptionUnsupported;
 }
 
 fn netConnectIp(userdata: ?*anyopaque, address: *const net.IpAddress, options: net.IpAddress.ConnectOptions) net.IpAddress.ConnectError!net.Socket {
-    _ = userdata;
+    const node = nodeFromUserdata(userdata);
-    _ = address;
+
-    _ = options;
+    if (options.mode != .stream)
-    @panic("Not implemented yet");
+        return error.SocketModeUnsupported;
+    if (options.protocol) |protocol| {
+        if (protocol != .tcp)
+            return error.ProtocolUnsupportedBySystem;
+    }
+
+    const node_address = ipAddressToNodeAddress(address) orelse return error.AddressFamilyUnsupported;
+    const handle = node.connect(node_address) catch |err| switch (err) {
+        error.DescriptorLimit => return error.ProcessFdQuotaExceeded,
+        error.UnavailableHost => return error.HostUnreachable,
+        error.PeerNotListeningOnAddress => return error.ConnectionRefused,
+        error.OutOfMemory => return error.SystemResources,
+    };
+
+    return .{
+        .handle = handle,
+        .address = nodeAddressToIpAddress(node_address),
+    };
 }
 
 fn netListenUnix(userdata: ?*anyopaque, address: *const net.UnixAddress, options: net.UnixAddress.ListenOptions) net.UnixAddress.ListenError!net.Socket.Handle {
@@ -1127,28 +1192,78 @@ fn netSocketCreatePair(userdata: ?*anyopaque, options: net.Socket.CreatePairOpti
 }
 
 fn netSend(userdata: ?*anyopaque, handle: net.Socket.Handle, messages: []net.OutgoingMessage, flags: net.SendFlags) struct { ?net.Socket.SendError, usize } {
-    _ = userdata;
+    const node = nodeFromUserdata(userdata);
-    _ = handle;
-    _ = messages;
     _ = flags;
-    @panic("Not implemented yet");
+
+    for (messages, 0..) |*message, index| {
+        const source = message.data_ptr[0..message.data_len];
+        const written = node.writeSocket(handle, source) catch |err| switch (err) {
+            error.InvalidHandle => return .{ error.SocketUnconnected, index },
+            error.NotConnected => return .{ error.SocketUnconnected, index },
+            error.OutOfMemory => return .{ error.SystemResources, index },
+        };
+        message.data_len = written;
+    }
+
+    return .{ null, messages.len };
 }
 
 /// Returns 0 on end of stream.
 fn netRead(userdata: ?*anyopaque, fd: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize {
-    _ = userdata;
+    const node = nodeFromUserdata(userdata);
-    _ = fd;
+    var copied: usize = 0;
-    _ = data;
+
-    @panic("Not implemented yet");
+    for (data) |buffer| {
+        if (buffer.len == 0)
+            continue;
+
+        const n = node.readSocket(fd, buffer) catch |err| switch (err) {
+            error.InvalidHandle => return error.SocketUnconnected,
+        };
+        copied += n;
+
+        if (n < buffer.len)
+            break;
+    }
+
+    return copied;
 }
 
 fn netWrite(userdata: ?*anyopaque, handle: net.Socket.Handle, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize {
-    _ = userdata;
+    const node = nodeFromUserdata(userdata);
-    _ = handle;
+    var copied: usize = 0;
-    _ = header;
+
-    _ = data;
+    if (header.len != 0)
-    _ = splat;
+        copied += node.writeSocket(handle, header) catch |err| switch (err) {
-    @panic("Not implemented yet");
+            error.InvalidHandle => return error.SocketUnconnected,
+            error.NotConnected => return error.SocketUnconnected,
+            error.OutOfMemory => return error.SystemResources,
+        };
+
+    if (data.len != 0) {
+        for (data[0 .. data.len - 1]) |bytes| {
+            if (bytes.len == 0)
+                continue;
+            copied += node.writeSocket(handle, bytes) catch |err| switch (err) {
+                error.InvalidHandle => return error.SocketUnconnected,
+                error.NotConnected => return error.SocketUnconnected,
+                error.OutOfMemory => return error.SystemResources,
+            };
+        }
+
+        const pattern = data[data.len - 1];
+        for (0..splat) |_| {
+            if (pattern.len == 0)
+                break;
+            copied += node.writeSocket(handle, pattern) catch |err| switch (err) {
+                error.InvalidHandle => return error.SocketUnconnected,
+                error.NotConnected => return error.SocketUnconnected,
+                error.OutOfMemory => return error.SystemResources,
+            };
+        }
+    }
+
+    return copied;
 }
 
 fn netWriteFile(userdata: ?*anyopaque, socket_handle: net.Socket.Handle, header: []const u8, file_reader: *Io.File.Reader, limit: Io.Limit) net.Stream.Writer.WriteFileError!usize {
@@ -1161,16 +1276,20 @@ fn netWriteFile(userdata: ?*anyopaque, socket_handle: net.Socket.Handle, header:
 }
 
 fn netClose(userdata: ?*anyopaque, handles: []const net.Socket.Handle) void {
-    _ = userdata;
+    const node = nodeFromUserdata(userdata);
-    _ = handles;
+
-    @panic("Not implemented yet");
+    for (handles) |handle| {
+        node.closeSocket(handle) catch {};
+    }
 }
 
 fn netShutdown(userdata: ?*anyopaque, handle: net.Socket.Handle, how: net.ShutdownHow) net.ShutdownError!void {
-    _ = userdata;
+    const node = nodeFromUserdata(userdata);
-    _ = handle;
     _ = how;
-    @panic("Not implemented yet");
+
+    node.closeSocket(handle) catch |err| switch (err) {
+        error.InvalidHandle => return error.SocketUnconnected,
+    };
 }
 
 fn netInterfaceNameResolve(userdata: ?*anyopaque, name: *const net.Interface.Name) net.Interface.Name.ResolveError!net.Interface {

src/network.zig

@@ -237,7 +237,7 @@ pub const Host = struct {
         socket.accept_queue.deinit(self.gpa);
     }
 
-    pub fn send(_: *Host, socket: *ConnSocket, source: []const u8) usize {
+    pub fn send(_: *Host, socket: *ConnSocket, source: []const u8) SendError!usize {
         const peer_conn = socket.peer_conn orelse return SendError.NotConnected;
         try peer_conn.input_buffer.appendSlice(peer_conn.host.gpa, source);
         return source.len;

src/test_simulator.zig

@@ -4,6 +4,7 @@ const Simulator = @import("simulator.zig");
 
 const Io = std.Io;
 const Clock = Io.Clock;
+const net = Io.net;
 
 fn testClockGrowsMonotonically(init: std.process.Init) anyerror!void {
     const io = init.io;
@@ -26,3 +27,72 @@ test "clock grows monotonically" {
 
     while (sim.scheduleOne()) {}
 }
+
+var networking_server_received = false;
+var networking_client_received = false;
+
+fn networkingServer(init: std.process.Init) anyerror!void {
+    const address: net.IpAddress = .{ .ip4 = .loopback(8080) };
+
+    var server = try address.listen(init.io, .{});
+    defer server.deinit(init.io);
+
+    var stream = try server.accept(init.io);
+    defer stream.close(init.io);
+
+    var reader_buffer: [16]u8 = undefined;
+    var reader = stream.reader(init.io, &reader_buffer);
+    var request: [4]u8 = undefined;
+    try reader.interface.readSliceAll(&request);
+    try std.testing.expectEqualStrings("ping", &request);
+    networking_server_received = true;
+
+    var writer_buffer: [16]u8 = undefined;
+    var writer = stream.writer(init.io, &writer_buffer);
+    try writer.interface.writeAll("pong");
+    try writer.interface.flush();
+}
+
+fn networkingClient(init: std.process.Init) anyerror!void {
+    const address: net.IpAddress = .{ .ip4 = .loopback(8080) };
+
+    var stream = try address.connect(init.io, .{ .mode = .stream });
+    defer stream.close(init.io);
+
+    var writer_buffer: [16]u8 = undefined;
+    var writer = stream.writer(init.io, &writer_buffer);
+    try writer.interface.writeAll("ping");
+    try writer.interface.flush();
+
+    try Io.sleep(init.io, .fromMilliseconds(1), .awake);
+
+    var reader_buffer: [16]u8 = undefined;
+    var reader = stream.reader(init.io, &reader_buffer);
+    var response: [4]u8 = undefined;
+    try reader.interface.readSliceAll(&response);
+    try std.testing.expectEqualStrings("pong", &response);
+    networking_client_received = true;
+}
+
+test "networking mocks support tcp stream listen connect accept read write" {
+    networking_server_received = false;
+    networking_client_received = false;
+
+    const loopback = net.Ip4Address.loopback(8080);
+    const server_addresses = [_]u32{std.mem.readInt(u32, &loopback.bytes, .big)};
+
+    var sim: Simulator = undefined;
+    sim.init(std.heap.page_allocator, std.Options.debug_io);
+    defer sim.deinit();
+
+    try sim.addExecutable("server", networkingServer);
+    try sim.addExecutable("client", networkingClient);
+
+    try sim.spawn("server", .{ .addresses = &server_addresses });
+    try sim.spawn("client", .{});
+
+    while (sim.scheduleOne()) {}
+
+    try std.testing.expect(networking_server_received);
+    try std.testing.expect(networking_client_received);
+}