Merge pull request #150 from Raiden1411/async_adventures

Author: Raiden1411

build.zig

@@ -373,8 +373,16 @@ fn addDependencies(
         .optimize = optimize,
     });
 
+    const aio = b.dependency("aio", .{
+        .target = target,
+        .optimize = optimize,
+    });
+
     mod.addImport("c_kzg_4844", c_kzg_4844_dep.module("c_kzg_4844"));
     mod.linkLibrary(c_kzg_4844_dep.artifact("c_kzg_4844"));
+
+    mod.addImport("aio", aio.module("aio"));
+    mod.addImport("coro", aio.module("coro"));
 }
 /// Builds and runs the benchmarks
 fn buildBenchmark(

build.zig.zon

@@ -13,6 +13,10 @@
         .c_kzg_4844 = .{
             .path = "./pkg/c-kzg-4844",
         },
+        .aio = .{
+            .url = "git+https://github.com/Cloudef/zig-aio.git#8d0507183396220a77f36e5182e5c91995d89449",
+            .hash = "aio-0.0.0-776t3n5WBQASyyJ0vCpZxIX7U6NTVEvaWVf1-9sjcVYa",
+        },
     },
     .paths = .{
         // This makes *all* files, recursively, included in this package. It is generally

examples/autobahn/autobahn.zig

@@ -2,7 +2,8 @@ const std = @import("std");
 const clients = @import("zabi").clients;
 
 const Allocator = std.mem.Allocator;
-const WebSocketClient = clients.WebSocketClient;
+const WebSocketClient = clients.blocking.WebSocketClient;
+const AsyncWebSocketClient = clients.non_blocking.AsyncWebSocketClient;
 
 pub fn main() !void {
     var gpa = std.heap.GeneralPurposeAllocator(.{}){};

src/clients/IPC.zig

@@ -47,7 +47,7 @@ const FeeHistory = transaction.FeeHistory;
 const Gwei = types.Gwei;
 const Hash = types.Hash;
 const Hex = types.Hex;
-const IpcReader = @import("ipc_reader.zig").IpcReader;
+const IpcReader = @import("blocking/IpcReader.zig");
 const JsonParsed = std.json.Parsed;
 const Log = log.Log;
 const LogRequest = log.LogRequest;

src/clients/WebSocket.zig

@@ -79,7 +79,7 @@ const Uri = std.Uri;
 const Value = std.json.Value;
 const WatchLogsRequest = log.WatchLogsRequest;
 const Wei = types.Wei;
-const WsClient = @import("WebSocketClient.zig");
+const WsClient = @import("blocking/WebSocketClient.zig");
 
 const WebSocketHandler = @This();
 

src/clients/async/AsyncIpcReader.zig

@@ -0,0 +1,146 @@
+//! Socket reader that is expected to be reading socket messages that are json messages.
+//! Will only allocate more memory if required.
+//!
+//! Calling `deinit` will close the socket and clear the buffer.
+const aio = @import("aio");
+const coro = @import("coro");
+const std = @import("std");
+const testing = std.testing;
+
+const Allocator = std.mem.Allocator;
+const Stream = std.net.Stream;
+const Buffer = std.fifo.LinearFifo(u8, .Dynamic);
+
+const Self = @This();
+
+/// Set of possible errors when reading from the socket.
+pub const ReadError = aio.Recv.Error || coro.io.Error ||
+    Allocator.Error || error{Closed};
+
+/// Set of possible error when writting to the stream.
+pub const WriteError = coro.io.Error || aio.Send.Error;
+
+pub const InitError = coro.io.Error || aio.Connect.Error ||
+    aio.Socket.Error || Allocator.Error;
+
+/// LinearFifo that grows as needed.
+buffer: Buffer,
+/// Socket stream to read from the unix socket.
+fd: std.posix.socket_t,
+/// Amount of bytes to discard on a successfull read.
+overflow: usize,
+/// The tell reader if the stream is closed.
+closed: bool,
+
+/// Sets the initial reader state in order to perform any necessary actions.
+///
+/// **Example**
+/// ```zig
+/// const stream = std.net.connectUnixSocket("/tmp/tmp.socket");
+///
+/// const ipc_reader = try AsyncIpcReader.init(std.heap.page_allocator, stream);
+/// defer ipc_reader.deinit();
+/// ```
+pub fn init(allocator: Allocator, path: []const u8) !@This() {
+    var socket: std.posix.socket_t = undefined;
+
+    try coro.io.single(.socket, .{
+        .domain = std.posix.AF.UNIX,
+        .flags = std.posix.SOCK.STREAM | std.posix.SOCK.CLOEXEC,
+        .protocol = std.posix.IPPROTO.IP,
+        .out_socket = &socket,
+    });
+
+    var addr = try std.net.Address.initUnix(path);
+    try coro.io.single(.connect, .{
+        .socket = socket,
+        .addr = &addr.any,
+        .addrlen = addr.getOsSockLen(),
+    });
+
+    return .{
+        .buffer = Buffer.init(allocator),
+        .fd = socket,
+        .overflow = 0,
+        .closed = false,
+    };
+}
+/// Frees the buffer and closes the stream.
+pub fn deinit(self: *@This()) void {
+    if (@atomicRmw(bool, &self.closed, .Xchg, true, .acq_rel) == false) {
+        self.buffer.deinit();
+        coro.io.single(.close_socket, .{ .socket = self.fd }) catch {};
+    }
+}
+/// "Reads" a json message and moves the necessary position members in order
+/// to have the necessary message.
+pub fn jsonMessage(self: *@This()) usize {
+    self.buffer.discard(self.overflow);
+    self.buffer.realign();
+
+    if (self.buffer.count <= 1) {
+        self.overflow = 0;
+        return 0;
+    }
+
+    var depth: usize = 0;
+    var index: usize = 0;
+    while (index < self.buffer.buf.len) : (index += 1) {
+        switch (self.buffer.buf[index]) {
+            '{' => depth += 1,
+            '}' => depth -= 1,
+            else => {},
+        }
+
+        // Check if we read a message or not.
+        if (depth == 0) {
+            self.overflow = index + 1;
+            return index + 1;
+        }
+    }
+
+    self.overflow = 0;
+    return 0;
+}
+/// Reads the bytes directly from the socket. Will allocate more memory as needed.
+pub fn read(self: *@This()) ReadError!void {
+    const buffer = try self.buffer.writableWithSize(std.math.maxInt(u16));
+    var read_bytes: usize = 0;
+
+    try coro.io.single(.recv, .{
+        .socket = self.fd,
+        .buffer = buffer,
+        .out_read = &read_bytes,
+    });
+
+    if (read_bytes == 0)
+        return error.Closed;
+
+    self.buffer.update(read_bytes);
+}
+/// Reads one message from the socket stream.
+///
+/// Will only make the socket read request if the buffer is at max capacity.
+/// Will grow the buffer as needed.
+pub fn readMessage(self: *@This()) ReadError![]u8 {
+    while (true) {
+        const size = self.jsonMessage();
+
+        if (size == 0) {
+            try self.read();
+            continue;
+        }
+
+        return @constCast(self.buffer.readableSliceOfLen(size));
+    }
+}
+/// Writes a message to the socket stream.
+pub fn writeMessage(
+    self: *@This(),
+    message: []u8,
+) WriteError!void {
+    return coro.io.single(.send, .{
+        .socket = self.fd,
+        .buffer = message,
+    });
+}