src: implement the icipc server process as a pipewire module

This can be loaded with `pipewire -c pipewire-ic-ipc.conf`

Signed-off-by: George Kiagiadakis <george.kiagiadakis@collabora.com>

5 files changed, 437 insertions, 4 deletions

diff --git a/meson.build b/meson.build
index c4556be..0359271 100644
--- a/meson.build
+++ b/meson.build
@@ -9,6 +9,7 @@ project('icipc', ['c'],
 )
 
 threads_dep = dependency('threads', required: true)
+pipewire_dep = dependency('libpipewire-0.3', required: get_option('server'))
 
 pkgconfig = import('pkgconfig')
 cc = meson.get_compiler('c')
diff --git a/meson_options.txt b/meson_options.txt
new file mode 100644
index 0000000..503d1d7
--- /dev/null
+++ b/meson_options.txt
@@ -0,0 +1,4 @@
+option('server', type : 'boolean', value : 'true',
+       description: 'Build the server-side (requires pipewire dep)')
+option('client', type : 'boolean', value : 'true',
+       description: 'Build the client-side (no dependencies)')
diff --git a/src/meson.build b/src/meson.build
index b9a198a..7ef8754 100644
--- a/src/meson.build
+++ b/src/meson.build
@@ -1,4 +1,19 @@
-executable('icipc-client',
-  'icipc-client.c',
-  dependencies : [icipc_dep, threads_dep],
-)
+if get_option('client')
+  executable('icipc-client',
+    'icipc-client.c',
+    dependencies : [icipc_dep, threads_dep],
+  )
+endif
+
+if get_option('server')
+  pipewire_module_access = shared_library(
+    'pipewire-module-protocol-ic-ipc',
+    [ 'module-protocol-ic-ipc.c' ],
+    c_args : [
+      '-DPACKAGE_VERSION="' + meson.project_version() + '"',
+    ],
+    install : true,
+    install_dir : pipewire_dep.get_variable(pkgconfig: 'moduledir'),
+    dependencies : [icipc_dep, pipewire_dep],
+  )
+endif
diff --git a/src/module-protocol-ic-ipc.c b/src/module-protocol-ic-ipc.c
new file mode 100644
index 0000000..7328d18
--- /dev/null
+++ b/src/module-protocol-ic-ipc.c
@@ -0,0 +1,386 @@
+/* PipeWire AGL Cluster IPC
+ *
+ * Copyright © 2021 Collabora Ltd.
+ *    @author George Kiagiadakis <george.kiagiadakis@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include <pipewire/impl.h>
+#include <pipewire/array.h>
+#include <pipewire/extensions/metadata.h>
+#include <spa/utils/result.h>
+#include <icipc.h>
+
+#define NAME "protocol-ic-ipc"
+
+#define SERVER_SUSPEND_REQUEST_NAME "SUSPEND"
+#define SERVER_RESUME_REQUEST_NAME "RESUME"
+
+static const struct spa_dict_item module_props[] = {
+        { PW_KEY_MODULE_AUTHOR,
+                "George Kiagiadakis <george.kiagiadakis@collabora.com>" },
+        { PW_KEY_MODULE_DESCRIPTION,
+                "Sends commands to PipeWire from the "
+                "AGL Instrument Cluster container" },
+        { PW_KEY_MODULE_USAGE, "[ icipc.name=(<name>|<path>), "
+                                  PW_KEY_REMOTE_NAME"=<name> ]" },
+        { PW_KEY_MODULE_VERSION, PACKAGE_VERSION },
+};
+
+struct impl {
+        struct pw_context *context;
+        struct pw_loop *loop;
+        struct icipc_server *server;
+
+        struct pw_array clients;
+
+        char *pipewire_remote;
+        struct spa_source *timeout_source;
+        struct pw_core *core;
+        struct pw_registry *registry;
+        struct pw_metadata *metadata;
+
+        struct spa_hook module_listener;
+        struct spa_hook core_proxy_listener;
+        struct spa_hook core_listener;
+        struct spa_hook registry_proxy_listener;
+        struct spa_hook registry_listener;
+        struct spa_hook metadata_proxy_listener;
+};
+
+#ifndef pw_array_add_int
+#define pw_array_add_int(a,i) *((int*) pw_array_add(a, sizeof(int))) = (i)
+#endif
+
+static void update_timer(struct impl *impl, time_t sec, long nsec)
+{
+        struct timespec value, interval;
+        value.tv_sec = sec;
+        value.tv_nsec = nsec;
+        interval.tv_sec = 0;
+        interval.tv_nsec = 0;
+        pw_loop_update_timer(impl->loop, impl->timeout_source,
+                             &value, &interval, false);
+        pw_loop_update_source(impl->loop, impl->timeout_source);
+}
+
+static void disconnect_from_pw(struct impl *impl) {
+        if (impl->core)
+                pw_core_disconnect(impl->core);
+}
+
+static void impl_free(struct impl *impl) {
+        pw_log_debug(NAME " %p: destroy", impl);
+
+        disconnect_from_pw(impl);
+        if (impl->timeout_source) {
+                update_timer(impl, 0, 0);
+                pw_loop_destroy_source(impl->loop, impl->timeout_source);
+        }
+        if (impl->server)
+                icipc_server_free(impl->server);
+        pw_array_clear(&impl->clients);
+        free(impl->pipewire_remote);
+        free(impl);
+}
+
+static void module_destroy(void *data) {
+        struct impl *impl = data;
+        impl_free(impl);
+}
+
+static const struct pw_impl_module_events module_events = {
+        PW_VERSION_IMPL_MODULE_EVENTS,
+        .destroy = module_destroy,
+};
+
+static void core_destroy(void *d) {
+        struct impl *impl = d;
+        spa_hook_remove(&impl->core_listener);
+        impl->core = NULL;
+}
+
+static const struct pw_proxy_events core_proxy_events = {
+        PW_VERSION_PROXY_EVENTS,
+        .destroy = core_destroy,
+};
+
+static void registry_destroy(void *d) {
+        struct impl *impl = d;
+        spa_hook_remove(&impl->registry_listener);
+        impl->registry = NULL;
+}
+
+static void registry_removed(void *d) {
+        struct impl *impl = d;
+        pw_proxy_destroy((struct pw_proxy *)impl->registry);
+}
+
+static const struct pw_proxy_events registry_proxy_events = {
+        PW_VERSION_PROXY_EVENTS,
+        .destroy = registry_destroy,
+        .removed = registry_removed,
+};
+
+static void metadata_destroy(void *d) {
+        struct impl *impl = d;
+        impl->metadata = NULL;
+}
+
+static void metadata_removed(void *d) {
+        struct impl *impl = d;
+        pw_proxy_destroy((struct pw_proxy *)impl->metadata);
+}
+
+static const struct pw_proxy_events metadata_proxy_events = {
+        PW_VERSION_PROXY_EVENTS,
+        .destroy = metadata_destroy,
+        .removed = metadata_removed,
+};
+
+static void on_core_error(
+                void *d, uint32_t id, int seq, int res, const char *message) {
+        struct impl *impl = d;
+
+        pw_log_error("error id:%u seq:%d res:%d (%s): %s",
+                      id, seq, res, spa_strerror(res), message);
+
+        /* pipewire disconnected */
+        if (id == PW_ID_CORE && res == -EPIPE)
+                update_timer(impl, 0, 1);
+}
+
+static const struct pw_core_events core_events = {
+        PW_VERSION_CORE_EVENTS,
+        .error = on_core_error,
+};
+
+static void on_global_added (
+                void *object, uint32_t id,
+                uint32_t permissions, const char *type, uint32_t version,
+                const struct spa_dict *props) {
+        struct impl *impl = object;
+        const char *str;
+
+        if (type && props && !strcmp(type, PW_TYPE_INTERFACE_Metadata) &&
+            (str = spa_dict_lookup(props, "metadata.name")) &&
+            !strcmp(str, "default")) {
+                impl->metadata = pw_registry_bind(impl->registry, id, type,
+                                                  version, 0);
+                pw_proxy_add_listener((struct pw_proxy*)impl->metadata,
+                                      &impl->metadata_proxy_listener,
+                                      &metadata_proxy_events, impl);
+        }
+}
+
+static const struct pw_registry_events registry_events = {
+        PW_VERSION_REGISTRY_EVENTS,
+        .global = on_global_added,
+};
+
+static void connect_to_pipewire(struct impl *impl) {
+        if (!impl->core) {
+                struct pw_properties *p = pw_properties_new(
+                                PW_KEY_REMOTE_NAME, impl->pipewire_remote,
+                                NULL);
+                impl->core = pw_context_connect(impl->context, p, 0);
+                if (!impl->core) {
+                        pw_log_warn(NAME " %p: failed to connect to pipewire:"
+                                    " %m", impl);
+                        update_timer(impl, 5, 0);
+                        return;
+                }
+                pw_log_info(NAME " %p: connected to pw", impl);
+        }
+
+        update_timer(impl, 0, 0);
+        impl->registry =
+                pw_core_get_registry(impl->core, PW_VERSION_REGISTRY, 0);
+
+        pw_proxy_add_listener((struct pw_proxy*)impl->core,
+                              &impl->core_proxy_listener,
+                              &core_proxy_events, impl);
+        pw_core_add_listener(impl->core,
+                             &impl->core_listener,
+                             &core_events, impl);
+
+        pw_proxy_add_listener((struct pw_proxy*)impl->registry,
+                              &impl->registry_proxy_listener,
+                              &registry_proxy_events, impl);
+        pw_registry_add_listener(impl->registry,
+                                 &impl->registry_listener,
+                                 &registry_events, impl);
+}
+
+static void on_timeout(void *data, uint64_t expirations) {
+        struct impl *impl = data;
+        pw_log_debug(NAME " %p: timeout", impl);
+
+        if (impl->core) {
+                pw_log_info(NAME " %p: disconnect", impl);
+                disconnect_from_pw(impl);
+                update_timer(impl, 5, 0);
+        } else
+                connect_to_pipewire(impl);
+}
+
+static int do_suspend_resume(
+                struct spa_loop *loop,
+                bool async,
+                uint32_t seq,
+                const void *data,
+                size_t size,
+                void *user_data) {
+        struct impl *impl = user_data;
+        const int suspend = *(const int *)data;
+
+        if (impl->metadata) {
+                pw_metadata_set_property(impl->metadata, 0, "suspend.playback",
+                        "Spa:Bool", suspend ? "1" : NULL);
+        }
+        return 0;
+}
+
+static bool suspend_handler(
+                struct icipc_server *s,
+                int client_fd,
+                const char *name,
+                const struct icipc_data *args,
+                void *data) {
+        struct impl *impl = data;
+        int *c;
+
+        pw_array_for_each(c, &impl->clients) {
+                if (*c == client_fd)
+                        return icipc_server_reply_ok (s, client_fd, NULL);
+        }
+        pw_array_add_int(&impl->clients, client_fd);
+
+        /* do suspend if this is the first client asking for suspend */
+        if (impl->clients.size == sizeof(int)) {
+                const int suspend = 1;
+                pw_loop_invoke(impl->loop, do_suspend_resume, 0, &suspend,
+                               sizeof(suspend), true, impl);
+        }
+        return icipc_server_reply_ok (s, client_fd, NULL);
+}
+
+static bool resume_handler(
+                struct icipc_server *s,
+                int client_fd,
+                const char *name,
+                const struct icipc_data *args,
+                void *data) {
+        struct impl *impl = data;
+        size_t old_size = impl->clients.size;
+        int *c;
+
+        pw_array_for_each(c, &impl->clients) {
+                if (*c == client_fd) {
+                        pw_array_remove(&impl->clients, c);
+                        break;
+                }
+        }
+        /* resume if there are no more clients suspending */
+        if (old_size > 0 && impl->clients.size == 0) {
+                const int suspend = 0;
+                pw_loop_invoke(impl->loop, do_suspend_resume, 0, &suspend,
+                               sizeof(suspend), true, impl);
+        }
+        /* don't reply if called with name == NULL from client_handler() */
+        return name ? icipc_server_reply_ok (s, client_fd, NULL) : true;
+}
+
+static void client_handler (
+                struct icipc_server *s,
+                int client_fd,
+                enum icipc_receiver_sender_state client_state,
+                void *data) {
+        struct impl *impl = data;
+
+        switch (client_state) {
+        case ICIPC_RECEIVER_SENDER_STATE_CONNECTED:
+                pw_log_info (NAME " %p: client connected %d", impl, client_fd);
+                break;
+        case ICIPC_RECEIVER_SENDER_STATE_DISCONNECTED: {
+                pw_log_info (NAME " %p: client disconnected %d", impl, client_fd);
+                resume_handler(s, client_fd, NULL, NULL, impl);
+                break;
+        }
+        default:
+                break;
+        }
+}
+
+SPA_EXPORT
+int pipewire__module_init(struct pw_impl_module *module, const char *args) {
+        struct impl *impl;
+        struct pw_properties *props = NULL;
+        const char *str = NULL;
+        int res = 0;
+
+        impl = calloc(1, sizeof(struct impl));
+        if (impl == NULL)
+                return -errno;
+
+        pw_log_debug(NAME " %p: new %s", impl, args);
+
+        impl->context = pw_impl_module_get_context(module);
+        impl->loop = pw_context_get_main_loop(impl->context);
+
+        /* setup icipc server */
+
+        if (args) {
+                props = pw_properties_new_string(args);
+                str = pw_properties_get(props, "icipc.name");
+        }
+        if (!str)
+                str = "icipc-0";
+
+        impl->server = icipc_server_new(str, false);
+        if (!impl->server) {
+                res = -ENOMEM;
+                goto out;
+        }
+
+        pw_array_init(&impl->clients, 4*sizeof(int));
+        icipc_server_set_client_handler(impl->server, client_handler, impl);
+        icipc_server_set_request_handler(impl->server,
+                                         SERVER_SUSPEND_REQUEST_NAME,
+                                         suspend_handler, impl);
+        icipc_server_set_request_handler(impl->server,
+                                         SERVER_RESUME_REQUEST_NAME,
+                                         resume_handler, impl);
+
+        /* connect to pipewire */
+
+        str = props ? pw_properties_get(props, PW_KEY_REMOTE_NAME) : NULL;
+        if (str)
+                impl->pipewire_remote = strdup(str);
+
+        impl->timeout_source = pw_loop_add_timer(impl->loop, on_timeout, impl);
+        update_timer(impl, 0, 1);
+
+        /* start icipc */
+
+        if (!icipc_receiver_start((struct icipc_receiver *) impl->server)) {
+                res = -errno;
+                pw_log_error("failed to start icipc server: %m");
+                goto out;
+        }
+
+        /* finish module setup */
+
+        pw_impl_module_add_listener(module, &impl->module_listener,
+                                    &module_events, impl);
+        pw_impl_module_update_properties(module,
+                                         &SPA_DICT_INIT_ARRAY(module_props));
+
+out:
+        if (props)
+                pw_properties_free(props);
+        if (res < 0)
+                impl_free(impl);
+        return res;
+}
diff --git a/src/pipewire-ic-ipc.conf b/src/pipewire-ic-ipc.conf
new file mode 100644
index 0000000..d2c1f18
--- /dev/null
+++ b/src/pipewire-ic-ipc.conf
@@ -0,0 +1,27 @@
+context.properties = {
+    support.dbus = false
+    core.daemon = false
+    #log.level = 2
+    application.name = pipewire-ic-ipc
+}
+
+context.spa-libs = {
+    support.*       = support/libspa-support
+}
+
+context.modules = [
+    # The Instrument Cluster IPC server
+    {   name = libpipewire-module-protocol-ic-ipc
+        args = {
+            #icipc.name = icipc-0
+            #remote.name = pipewire-0
+        }
+    }
+
+    # The native communication protocol.
+    {   name = libpipewire-module-protocol-native }
+
+    # Allows applications to create metadata objects. It creates
+    # a factory for Metadata objects.
+    {   name = libpipewire-module-metadata }
+]