feat(dispatcher): add event relay dispatcher and receiver modules

CMakeLists.txt

@@ -8,6 +8,16 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON)
 if (CMAKE_INSTALL_PREFIX_INITIALIZED_TO_DEFAULT)
     set(CMAKE_INSTALL_PREFIX /usr)
 endif ()
+# Top-level testing switch: child components (e.g. src/dispatcher) declare
+# their own `option(ENABLE_TESTING ... OFF)`, but the cache variable set here
+# takes precedence, so a single -DENABLE_TESTING=ON at configure time
+# propagates to every component's test subdirectory.
+option(ENABLE_TESTING "Enable building tests" OFF)
 
+# Let `ctest` from the top-level build dir traverse into component test
+# dirs (each component registers its tests via its own ENABLE_TESTING).
+# Must precede add_subdirectory: testing support is only wired into
+# subdirectories added after this call.
+enable_testing()
 add_subdirectory(${PROJECT_SOURCE_DIR}/src)
 add_subdirectory(${PROJECT_SOURCE_DIR}/config)

src/dispatcher/CMakeLists.txt

@@ -11,18 +11,26 @@ set(CMAKE_C_STANDARD_REQUIRED ON)
 
 find_package(PkgConfig REQUIRED)
 pkg_check_modules(GLIB REQUIRED glib-2.0)
+pkg_check_modules(GIO REQUIRED gio-2.0)
+pkg_check_modules(GIO_UNIX REQUIRED gio-unix-2.0)
 
 add_library(deepin-anything-dispatcher STATIC
     event_dispatcher.c
     event_receiver.c
+    event_relay_dispatcher.c
+    event_relay_receiver.c
 )
 
 target_include_directories(deepin-anything-dispatcher PUBLIC
     ${GLIB_INCLUDE_DIRS}
+    ${GIO_INCLUDE_DIRS}
+    ${GIO_UNIX_INCLUDE_DIRS}
 )
 
 target_link_libraries(deepin-anything-dispatcher PUBLIC
     ${GLIB_LIBRARIES}
+    ${GIO_LIBRARIES}
+    ${GIO_UNIX_LIBRARIES}
 )
 
 # Build tests if enabled

src/dispatcher/event_relay_dispatcher.c

@@ -0,0 +1,131 @@
+// SPDX-FileCopyrightText: 2026 UnionTech Software Technology Co., Ltd.
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#define _GNU_SOURCE
+#define G_LOG_USE_STRUCTURED
+#include "event_relay_dispatcher.h"
+
+#include <errno.h>
+#include <fcntl.h>
+#include <stddef.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+/* ── Data structure ─────────────────────────────────────────────── */
+
+struct EventRelayDispatcher {
+    GPtrArray *socket_list;       /* sender-side fds from socketpair()       */
+    guint      max_event_channel; /* 0 = unlimited                           */
+};
+
+/* ── Helpers ────────────────────────────────────────────────────── */
+
+/**
+ * close_fd:
+ * @ptr: a #GINT_TO_POINTER-encoded fd
+ *
+ * GDestroyNotify-compatible wrapper that closes the fd stored as @ptr.
+ */
+static void close_fd(gpointer ptr)
+{
+    int fd = GPOINTER_TO_INT(ptr);
+    if (fd >= 0)
+        close(fd);
+}
+
+/* ── Public API ─────────────────────────────────────────────────── */
+
+EventRelayDispatcher *event_relay_dispatcher_new(guint max_event_channel)
+{
+    EventRelayDispatcher *d = g_new0(EventRelayDispatcher, 1);
+    d->socket_list = g_ptr_array_new_full(8, close_fd);
+    d->max_event_channel = max_event_channel;
+    return d;
+}
+
+void event_relay_dispatcher_free(EventRelayDispatcher *dispatcher)
+{
+    if (dispatcher == NULL)
+        return;
+
+    /* Closes every fd via the close_fd destroy notify. */
+    if (dispatcher->socket_list != NULL)
+        g_ptr_array_free(dispatcher->socket_list, TRUE);
+
+    g_free(dispatcher);
+}
+
+int event_relay_dispatcher_get_event_channel(EventRelayDispatcher *dispatcher)
+{
+    g_return_val_if_fail(dispatcher != NULL, -1);
+
+    /* Check channel limit. */
+    if (dispatcher->max_event_channel > 0 &&
+        dispatcher->socket_list->len >= dispatcher->max_event_channel) {
+        g_debug("relay: max_event_channel limit (%u) reached",
+                dispatcher->max_event_channel);
+        return -1;
+    }
+
+    int fds[2];
+    if (socketpair(AF_UNIX, SOCK_SEQPACKET | SOCK_NONBLOCK, 0, fds) < 0) {
+        g_warning("relay: socketpair() failed: %s", strerror(errno));
+        return -1;
+    }
+
+    /* Enlarge send buffer so bursts of events don't cause EAGAIN. */
+    int buf_size = 1 << 20;  /* 1 MiB */
+    if (setsockopt(fds[0], SOL_SOCKET, SO_SNDBUF, &buf_size, sizeof(buf_size)) < 0)
+        g_debug("relay: setsockopt(SO_SNDBUF) on sender failed: %s",
+                 strerror(errno));
+
+    /* Store sender end (fds[0]) in the list, return client end (fds[1]). */
+    g_ptr_array_add(dispatcher->socket_list, GINT_TO_POINTER(fds[0]));
+
+    g_debug("relay: created event channel (sender fd=%d, client fd=%d, total=%u)",
+            fds[0], fds[1], dispatcher->socket_list->len);
+
+    return fds[1];
+}
+
+gboolean event_relay_dispatcher_send(EventRelayDispatcher *dispatcher,
+                                     const void *buf, size_t len)
+{
+    g_return_val_if_fail(dispatcher != NULL, FALSE);
+    g_return_val_if_fail(buf != NULL, FALSE);
+    g_return_val_if_fail(len > 0, FALSE);
+
+    /* Iterate in reverse so we can safely remove during iteration. */
+    gboolean ok = TRUE;
+
+    for (gint i = dispatcher->socket_list->len - 1; i >= 0; i--) {
+        int fd = GPOINTER_TO_INT(g_ptr_array_index(dispatcher->socket_list, i));
+
+        ssize_t ret;
+        do {
+            ret = send(fd, buf, len, MSG_NOSIGNAL);
+        } while (ret < 0 && errno == EINTR);
+
+        if (ret < 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                g_debug("relay: slow client fd %d, kicking", fd);
+                /* Remove from array first so the destroy notify owns the
+                 * close — no double-close, no fd-reuse hazard. */
+                g_ptr_array_remove_index_fast(dispatcher->socket_list, i);
+            } else if (errno == EPIPE || errno == ECONNRESET) {
+                g_debug("relay: broken connection fd %d", fd);
+                g_ptr_array_remove_index_fast(dispatcher->socket_list, i);
+            } else {
+                g_warning("relay: send() to fd %d failed: %s",
+                          fd, strerror(errno));
+                ok = FALSE;
+                /* Continue to remaining channels — broadcast semantics
+                 * mean one client's error must not starve the others. */
+            }
+        }
+    }
+
+    return ok;
+}

src/dispatcher/event_relay_dispatcher.h

@@ -0,0 +1,98 @@
+// SPDX-FileCopyrightText: 2026 UnionTech Software Technology Co., Ltd.
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#ifndef EVENT_RELAY_DISPATCHER_H
+#define EVENT_RELAY_DISPATCHER_H
+
+#include <glib.h>
+#include <sys/types.h>
+
+G_BEGIN_DECLS
+
+/**
+ * EventRelayDispatcher:
+ *
+ * An opaque relay event dispatcher. Unlike #EventDispatcher (which filters
+ * events by per-UID subscription prefixes), the relay dispatcher broadcasts
+ * every event to every connected channel without filtering. The downstream
+ * relay agent is responsible for its own filtering.
+ *
+ * Communication is via pre-established #socketpair endpoints using
+ * %SOCK_SEQPACKET (preserves message boundaries). Channel establishment is
+ * driven by a D-Bus <literal>GetEventChannel</literal> method call: the
+ * server-side handler calls event_relay_dispatcher_get_event_channel() to
+ * create a socketpair, stores one end, and returns the other end to the
+ * client via D-Bus fd passing.
+ */
+typedef struct EventRelayDispatcher EventRelayDispatcher;
+
+/**
+ * event_relay_dispatcher_new: (constructor)
+ * @max_event_channel: Maximum number of concurrent event channels (0 = unlimited)
+ *
+ * Creates a new relay event dispatcher. The dispatcher maintains a list of
+ * socketpair endpoints (one per channel). When a client requests a channel
+ * via event_relay_dispatcher_get_event_channel(), a new socketpair is created
+ * and one end is stored in the list while the other is returned to the caller.
+ *
+ * Returns: (transfer full) (nullable): a new #EventRelayDispatcher, or %NULL
+ *     on failure
+ */
+EventRelayDispatcher *event_relay_dispatcher_new(guint max_event_channel);
+
+/**
+ * event_relay_dispatcher_free: (skip)
+ * @dispatcher: (nullable)
+ *
+ * Frees the relay dispatcher and all associated resources. Closes every
+ * socketpair endpoint in the list. Safe to call on %NULL.
+ *
+ * Not thread-safe relative to get_event_channel/send — the caller must
+ * ensure no concurrent calls.
+ */
+void event_relay_dispatcher_free(EventRelayDispatcher *dispatcher);
+
+/**
+ * event_relay_dispatcher_get_event_channel:
+ * @dispatcher: an #EventRelayDispatcher
+ *
+ * Creates a new event channel by calling socketpair(%AF_UNIX,
+ * %SOCK_SEQPACKET | %SOCK_NONBLOCK, 0). One end of the pair is stored
+ * in the dispatcher's internal socket list; the other end is returned
+ * as a file descriptor.
+ *
+ * If @max_event_channel is non-zero and the number of active channels
+ * has already reached the limit, this function returns -1.
+ *
+ * The D-Bus service framework calls this function to respond to a
+ * <literal>GetEventChannel</literal> method call. The returned fd is
+ * passed back to the client via D-Bus fd passing. The service framework
+ * also returns an @event_protocol_id that identifies the data structure
+ * of the buffer used in send() — the current protocol ID is 0, meaning
+ * #dispatch_event_t.
+ *
+ * Returns: a file descriptor (the client end of the socketpair), or -1
+ *     on failure or if the channel limit has been reached
+ */
+int event_relay_dispatcher_get_event_channel(EventRelayDispatcher *dispatcher);
+
+/**
+ * event_relay_dispatcher_send:
+ * @dispatcher: an #EventRelayDispatcher
+ * @buf: the data to send
+ * @len: the length of @buf in bytes
+ *
+ * Sends @buf to every active channel in the dispatcher's socket list.
+ * Disconnected or slow clients (EAGAIN/EPIPE/ECONNRESET on send) are
+ * automatically cleaned up — their fd is closed and removed from the list.
+ *
+ * Returns: %TRUE if the data was processed (even if no clients received it),
+ *          %FALSE on critical error
+ */
+gboolean event_relay_dispatcher_send(EventRelayDispatcher *dispatcher,
+                                     const void *buf, size_t len);
+
+G_END_DECLS
+
+#endif /* EVENT_RELAY_DISPATCHER_H */