Core: Refactor ZMQ usage into ZMQService

settings/default/network.lua

@@ -37,9 +37,10 @@ xi.settings.network =
     HTTP_HOST   = 'localhost',
     HTTP_PORT   = 8088,
 
-    -- Central message server settings (ensure these are the same on both all map servers and the central (lobby) server
-    ZMQ_IP   = '127.0.0.1',
-    ZMQ_PORT = 54003,
+    -- Central message server settings
+    ZMQ_TRANSPORT = 'tcp',
+    ZMQ_IP        = '127.0.0.1',
+    ZMQ_PORT      = 54003,
 
     -- ===========================
     -- NOTE: The settings that follow will not necessarily need to be modified

src/common/CMakeLists.txt

@@ -1,4 +1,13 @@
 set(COMMON_SOURCES
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/channel.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/endpoint.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/endpoint.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/router_endpoint.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/router_endpoint.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/dealer_endpoint.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/dealer_endpoint.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/zmq_service.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/zmq/zmq_service.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/application.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/application.h
     ${CMAKE_CURRENT_SOURCE_DIR}/arguments.cpp
@@ -19,6 +28,7 @@ set(COMMON_SOURCES
     ${CMAKE_CURRENT_SOURCE_DIR}/engine.h
     ${CMAKE_CURRENT_SOURCE_DIR}/filewatcher.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/filewatcher.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/ipp_message.h
     ${CMAKE_CURRENT_SOURCE_DIR}/ipp.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/ipp.h
     ${CMAKE_CURRENT_SOURCE_DIR}/logging.cpp

src/common/application.cpp

@@ -295,6 +295,11 @@ auto Application::scheduler() -> Scheduler&
     return scheduler_;
 }
 
+auto Application::zmqService() -> ZMQService&
+{
+    return zmqService_;
+}
+
 auto Application::args() const -> Arguments&
 {
     return *args_;

src/common/application.h

@@ -24,6 +24,7 @@
 #include "arguments.h"
 #include "common/engine.h"
 #include "common/scheduler.h"
+#include "common/zmq/zmq_service.h"
 
 #include <asio.hpp> // for signal_set
 
@@ -98,13 +99,17 @@ class Application
     //
 
     auto scheduler() -> Scheduler&;
+    auto zmqService() -> ZMQService&;
     auto args() const -> Arguments&;
     auto console() const -> ConsoleService&;
 
 protected:
     std::chrono::steady_clock::time_point startTime_{ std::chrono::steady_clock::now() };
 
-    Scheduler        scheduler_;
+    Scheduler scheduler_;
+
+    ZMQService zmqService_;
+
     asio::signal_set signals_;
 
     std::string serverName_;

src/common/ipp_message.h

@@ -0,0 +1,33 @@
+/*
+===========================================================================
+
+  Copyright (c) 2026 LandSandBoat Dev Teams
+
+  This program is free software: you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation, either version 3 of the License, or
+  (at your option) any later version.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see http://www.gnu.org/licenses/
+
+===========================================================================
+*/
+
+#pragma once
+
+#include <common/ipp.h>
+
+#include <vector>
+
+// An IP+Port-addressed message: a routing id plus an opaque payload.
+struct IPPMessage
+{
+    IPP                ipp;
+    std::vector<uint8> payload;
+};

src/common/zmq/channel.h

@@ -0,0 +1,67 @@
+/*
+===========================================================================
+
+  Copyright (c) 2026 LandSandBoat Dev Teams
+
+  This program is free software: you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation, either version 3 of the License, or
+  (at your option) any later version.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see http://www.gnu.org/licenses/
+
+===========================================================================
+*/
+
+#pragma once
+
+#include <common/logging.h>
+
+#include <atomic>
+#include <cstddef>
+#include <utility>
+
+#include <concurrentqueue.h>
+
+namespace ipc
+{
+
+//
+// ipc::Channel
+//
+//   A non-owning handle to one IPC endpoint's message queues.
+//
+template <typename MessageT>
+class Channel
+{
+public:
+    Channel(moodycamel::ConcurrentQueue<MessageT>& incoming, moodycamel::ConcurrentQueue<MessageT>& outgoing)
+    : incoming_(incoming)
+    , outgoing_(outgoing)
+    {
+    }
+
+    // Dequeue one inbound message. Returns false when none are available.
+    [[nodiscard]] auto tryReceive(MessageT& out) -> bool
+    {
+        return incoming_.try_dequeue(out);
+    }
+
+    // Queue a message for the I/O thread to send.
+    auto send(MessageT message) -> void
+    {
+        outgoing_.enqueue(std::move(message));
+    }
+
+private:
+    moodycamel::ConcurrentQueue<MessageT>& incoming_;
+    moodycamel::ConcurrentQueue<MessageT>& outgoing_;
+};
+
+} // namespace ipc

src/common/zmq/dealer_endpoint.cpp

@@ -0,0 +1,84 @@
+/*
+===========================================================================
+
+  Copyright (c) 2026 LandSandBoat Dev Teams
+
+  This program is free software: you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation, either version 3 of the License, or
+  (at your option) any later version.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see http://www.gnu.org/licenses/
+
+===========================================================================
+*/
+
+#include <common/zmq/dealer_endpoint.h>
+
+#include <common/logging.h>
+
+#include <utility>
+
+DealerEndpoint::DealerEndpoint(std::string endpoint, uint64 routingId)
+: endpoint_(std::move(endpoint))
+, routingId_(routingId)
+{
+}
+
+auto DealerEndpoint::open(zmq::context_t& ctx) -> bool
+{
+    socket_ = zmq::socket_t(ctx, zmq::socket_type::dealer);
+    socket_.set(zmq::sockopt::routing_id, zmq::const_buffer(&routingId_, sizeof(uint64)));
+    try
+    {
+        socket_.connect(endpoint_);
+        opened_.store(true, std::memory_order_release);
+    }
+    catch (const zmq::error_t& err)
+    {
+        ShowError(fmt::format("ZMQService: unable to connect dealer socket '{}': {}", endpoint_, err.what()));
+        opened_.store(false, std::memory_order_release);
+    }
+    return opened_.load(std::memory_order_acquire);
+}
+
+auto DealerEndpoint::close() -> void
+{
+    if (socket_)
+    {
+        socket_.close();
+    }
+}
+
+auto DealerEndpoint::socketHandle() -> void*
+{
+    return socket_.handle();
+}
+
+auto DealerEndpoint::onReadable() -> void
+{
+    while (true)
+    {
+        zmq::message_t msg;
+        if (!socket_.recv(msg, zmq::recv_flags::dontwait))
+        {
+            break;
+        }
+        incomingQueue_.enqueue(std::move(msg));
+    }
+}
+
+auto DealerEndpoint::flushOutbound() -> void
+{
+    zmq::message_t out;
+    while (outgoingQueue_.try_dequeue(out))
+    {
+        socket_.send(out, zmq::send_flags::dontwait);
+    }
+}

src/common/zmq/dealer_endpoint.h

@@ -0,0 +1,50 @@
+/*
+===========================================================================
+
+  Copyright (c) 2026 LandSandBoat Dev Teams
+
+  This program is free software: you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation, either version 3 of the License, or
+  (at your option) any later version.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see http://www.gnu.org/licenses/
+
+===========================================================================
+*/
+
+#pragma once
+
+#include <common/cbasetypes.h>
+#include <common/zmq/endpoint.h>
+
+#include <string>
+
+#include <concurrentqueue.h>
+#include <zmq.hpp>
+
+class DealerEndpoint final : public ZmqEndpoint
+{
+public:
+    DealerEndpoint(std::string endpoint, uint64 routingId);
+
+    auto open(zmq::context_t& ctx) -> bool override;
+    auto close() -> void override;
+    auto socketHandle() -> void* override;
+    auto onReadable() -> void override;
+    auto flushOutbound() -> void override;
+
+    moodycamel::ConcurrentQueue<zmq::message_t> incomingQueue_;
+    moodycamel::ConcurrentQueue<zmq::message_t> outgoingQueue_;
+
+private:
+    std::string   endpoint_;
+    uint64        routingId_;
+    zmq::socket_t socket_;
+};

src/common/zmq/endpoint.cpp

@@ -0,0 +1,29 @@
+/*
+===========================================================================
+
+  Copyright (c) 2026 LandSandBoat Dev Teams
+
+  This program is free software: you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation, either version 3 of the License, or
+  (at your option) any later version.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU General Public License for more details.
+
+  You should have received a copy of the GNU General Public License
+  along with this program.  If not, see http://www.gnu.org/licenses/
+
+===========================================================================
+*/
+
+#include <common/zmq/endpoint.h>
+
+ZmqEndpoint::~ZmqEndpoint() = default;
+
+auto ZmqEndpoint::opened() const -> bool
+{
+    return opened_.load(std::memory_order_acquire);
+}