Implement server

src/client.zig

@@ -99,6 +99,8 @@ pub const Client = struct {
         var arena = ArenaAllocator.init(parent_alloc);
         const self = try arena.allocator.create(Self);
         self.arena = arena;
+        self.ifce = null;
+        self.mqtt = null;
         errdefer self.deinit();
         const alloc = &self.arena.allocator;
 
@@ -125,8 +127,8 @@ pub const Client = struct {
     }
 
     pub fn deinit(self: *Self) void {
-        self.mqtt.?.deinit();
-        self.ifce.?.deinit();
+        if(self.mqtt) |m| m.deinit();
+        if(self.ifce) |i| i.deinit();
         self.arena.deinit();
     }
 
@@ -184,6 +186,6 @@ pub fn main() !void {
     const alloc = &gpa.allocator;
     const conf_path = try std.fs.cwd().realpathAlloc(alloc, "zika_config.json");
     const conf = try config.get(alloc, conf_path);
-    const client = try Client(void).init(alloc, &conf);
+    const client = try Client.init(alloc, &conf);
     try client.run();
 }

src/config.zig

@@ -43,7 +43,6 @@ pub const DriverConfig = struct {
 pub const ServerConfig = struct {
     id_length: u8 = 4,
     topic: []const u8,
-    max_tunnels: u16 = 16,
     pool_start: []const u8,
     pool_end: []const u8,
 };

src/mqtt.zig

@@ -38,6 +38,7 @@ pub fn Client(comptime T: type) type {
 
         alloc: *Allocator,
         mosq: *Mosq.mosquitto,
+        mosq_thread: ?*std.Thread,
 
         conf: Conf,
         host_cstr: []const u8,
@@ -116,20 +117,28 @@ pub fn Client(comptime T: type) type {
         }
 
         pub fn deinit(self: *Self) void {
+            _ = Mosq.mosquitto_disconnect(self.mosq);
+            if(self.mosq_thread) |t| std.Thread.wait(t);
             Mosq.mosquitto_destroy(self.mosq);
         }
 
         pub fn connect(self: *Self) !void {
-            var rc = Mosq.mosquitto_loop_start(self.mosq);
+            _ = Mosq.mosquitto_threaded_set(self.mosq, true);
+            self.mosq_thread = try std.Thread.spawn(thread_main, self);
+
+            const keepalive = self.conf.opts.keepalive_interval;
+            const rc = Mosq.mosquitto_connect_async(self.mosq, @ptrCast([*c]const u8, self.host_cstr), self.conf.port, keepalive);
             if (rc != Mosq.MOSQ_ERR_SUCCESS) {
-                std.log.err("mosquitto_loop_start: {s}", .{Mosq.mosquitto_strerror(rc)});
+                std.log.err("mosquitto_connect_async: {s}", .{Mosq.mosquitto_strerror(rc)});
                 return Error.ConnectFailed;
             }
+        }
 
+        fn thread_main(self: *Self) !void {
             const keepalive = self.conf.opts.keepalive_interval;
-            rc = Mosq.mosquitto_connect_async(self.mosq, @ptrCast([*c]const u8, self.host_cstr), self.conf.port, keepalive);
+            const rc = Mosq.mosquitto_loop_forever(self.mosq, keepalive*1000, 1);
             if (rc != Mosq.MOSQ_ERR_SUCCESS) {
-                std.log.err("mosquitto_connect_async: {s}", .{Mosq.mosquitto_strerror(rc)});
+                std.log.err("mosquitto_loop_forever: {s}", .{Mosq.mosquitto_strerror(rc)});
                 return Error.ConnectFailed;
             }
         }