lib: add wpipc library

Simple library that uses sockets for inter-process communication. It provides an
API to create server and client objects. Users can add custom handlers in the
server, and clients can send requests for those custom handlers.

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

4 files changed, 513 insertions, 0 deletions

diff --git a/tests/client-server.c b/tests/client-server.c
new file mode 100644
index 0000000..ca6d1c3
--- /dev/null
+++ b/tests/client-server.c
@@ -0,0 +1,124 @@
+/* PipeWire AGL Cluster IPC
+ *
+ * Copyright © 2021 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include <glib.h>
+#include <spa/pod/builder.h>
+#include <spa/pod/parser.h>
+#include <icipc/icipc.h>
+
+#define TEST_ADDRESS "/tmp/icipc-client-server"
+
+static bool
+increment_request_handler (struct icipc_server *self, int client_fd,
+    const char *name, const struct spa_pod *args, void *data)
+{
+  int32_t val = 0;
+  g_assert_true (spa_pod_is_int (args));
+  g_assert_true (spa_pod_get_int (args, &val) == 0);
+  struct spa_pod_int res = SPA_POD_INIT_Int (val + 1);
+  return icipc_server_reply_ok (self, client_fd, (struct spa_pod *)&res);
+}
+
+static bool
+error_request_handler (struct icipc_server *self, int client_fd,
+    const char *name, const struct spa_pod *args, void *data)
+{
+  return icipc_server_reply_error (self, client_fd, "error message");
+}
+
+struct reply_data {
+  int32_t incremented;
+  const char *error;
+  int n_replies;
+  GMutex mutex;
+};
+
+static void
+wait_for_reply (struct reply_data *data, int n_replies)
+{
+  while (true) {
+    g_mutex_lock (&data->mutex);
+    if (data->n_replies == n_replies) {
+      g_mutex_unlock (&data->mutex);
+      break;
+    }
+    g_mutex_unlock (&data->mutex);
+  }
+}
+
+static void
+reply_handler (struct icipc_sender *self, const uint8_t *buffer, size_t size, void *p)
+{
+  struct reply_data *data = p;
+  g_assert_nonnull (data);
+
+  g_mutex_lock (&data->mutex);
+
+  const struct spa_pod *pod = icipc_client_send_request_finish (self, buffer, size, &data->error);
+  if (pod) {
+    g_assert_true (spa_pod_is_int (pod));
+    g_assert_true (spa_pod_get_int (pod, &data->incremented) == 0);
+  }
+  data->n_replies++;
+
+  g_mutex_unlock (&data->mutex);
+}
+
+static void
+test_icipc_server_client ()
+{
+  struct icipc_server *s = icipc_server_new (TEST_ADDRESS, true);
+  g_assert_nonnull (s);
+  struct icipc_client *c = icipc_client_new (TEST_ADDRESS, true);
+  g_assert_nonnull (c);
+  struct reply_data data;
+  g_mutex_init (&data.mutex);
+
+  /* add request handlers */
+  g_assert_true (icipc_server_set_request_handler (s, "INCREMENT", increment_request_handler, NULL));
+  g_assert_true (icipc_server_set_request_handler (s, "ERROR", error_request_handler, NULL));
+
+  /* send an INCREMENT request of 3, and make sure the returned value is 4 */
+  data.incremented = -1;
+  data.error = NULL;
+  data.n_replies = 0;
+  struct spa_pod_int i = SPA_POD_INIT_Int (3);
+  g_assert_true (icipc_client_send_request (c, "INCREMENT", (struct spa_pod *)&i, reply_handler, &data));
+  wait_for_reply (&data, 1);
+  g_assert_null (data.error);
+  g_assert_cmpint (data.incremented, ==, 4);
+
+  /* send an ERROR request, and make sure the returned value is an error */
+  data.error = NULL;
+  data.n_replies = 0;
+  g_assert_true (icipc_client_send_request (c, "ERROR", NULL, reply_handler, &data));
+  wait_for_reply (&data, 1);
+  g_assert_cmpstr (data.error, ==, "error message");
+
+  /* send an unhandled request, and make sure the server replies with an error */
+  data.error = NULL;
+  data.n_replies = 0;
+  g_assert_true (icipc_client_send_request (c, "UNHANDLED-REQUEST", NULL, reply_handler, &data));
+  wait_for_reply (&data, 1);
+  g_assert_cmpstr (data.error, ==, "request handler not found");
+
+  /* clean up */
+  g_mutex_clear (&data.mutex);
+  icipc_client_free (c);
+  icipc_server_free (s);
+}
+
+gint
+main (gint argc, gchar *argv[])
+{
+  g_test_init (&argc, &argv, NULL);
+
+  g_test_add_func ("/icipc/icipc-server-client", test_icipc_server_client);
+
+  return g_test_run ();
+}
diff --git a/tests/meson.build b/tests/meson.build
new file mode 100644
index 0000000..12771c0
--- /dev/null
+++ b/tests/meson.build
@@ -0,0 +1,26 @@
+common_deps = [icipc_dep, glib_dep]
+common_env = [
+  'G_TEST_SRCDIR=@0@'.format(meson.current_source_dir()),
+  'G_TEST_BUILDDIR=@0@'.format(meson.current_build_dir()),
+]
+
+test(
+  'test-icipc-sender-receiver',
+  executable('test-sender-receiver', 'sender-receiver.c', dependencies: common_deps),
+  env: common_env,
+  workdir : meson.current_source_dir(),
+)
+
+test(
+  'test-icipc-protocol',
+  executable('test-protocol', 'protocol.c', dependencies: common_deps),
+  env: common_env,
+  workdir : meson.current_source_dir(),
+)
+
+test(
+  'test-icipc-client-server',
+  executable('test-client-server', 'client-server.c', dependencies: common_deps),
+  env: common_env,
+  workdir : meson.current_source_dir(),
+)
diff --git a/tests/protocol.c b/tests/protocol.c
new file mode 100644
index 0000000..5e272a1
--- /dev/null
+++ b/tests/protocol.c
@@ -0,0 +1,77 @@
+/* PipeWire AGL Cluster IPC
+ *
+ * Copyright © 2021 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include <glib.h>
+#include <spa/pod/builder.h>
+#include <spa/pod/parser.h>
+#include <icipc/icipc.h>
+
+static void
+test_icipc_protocol ()
+{
+  uint8_t b[1024];
+
+  /* request null value */
+  {
+    icipc_protocol_build_request (b, sizeof(b), "name", NULL);
+    const char *name = NULL;
+    const struct spa_pod *value = NULL;
+    g_assert_true (icipc_protocol_parse_request (b, sizeof(b), &name, &value));
+    g_assert_cmpstr (name, ==, "name");
+    g_assert_true (spa_pod_is_none (value));
+  }
+
+  /* request */
+  {
+    struct spa_pod_int i = SPA_POD_INIT_Int (8);
+    icipc_protocol_build_request (b, sizeof(b), "name", (struct spa_pod *)&i);
+    const char *name = NULL;
+    const struct spa_pod_int *value = NULL;
+    g_assert_true (icipc_protocol_parse_request (b, sizeof(b), &name, (const struct spa_pod **)&value));
+    g_assert_cmpstr (name, ==, "name");
+    g_assert_cmpint (value->value, ==, 8);
+  }
+
+  /* reply error */
+  {
+    icipc_protocol_build_reply_error (b, sizeof(b), "error message");
+    g_assert_true (icipc_protocol_is_reply_error (b, sizeof(b)));
+    const char *msg = NULL;
+    g_assert_true (icipc_protocol_parse_reply_error (b, sizeof(b), &msg));
+    g_assert_cmpstr (msg, ==, "error message");
+  }
+
+  /* reply ok null value */
+  {
+    icipc_protocol_build_reply_ok (b, sizeof(b), NULL);
+    g_assert_true (icipc_protocol_is_reply_ok (b, sizeof(b)));
+    const struct spa_pod *value = NULL;
+    g_assert_true (icipc_protocol_parse_reply_ok (b, sizeof(b), &value));
+    g_assert_true (spa_pod_is_none (value));
+  }
+
+  /* reply ok */
+  {
+    struct spa_pod_int i = SPA_POD_INIT_Int (3);
+    icipc_protocol_build_reply_ok (b, sizeof(b), (struct spa_pod *)&i);
+    g_assert_true (icipc_protocol_is_reply_ok (b, sizeof(b)));
+    const struct spa_pod_int *value = NULL;
+    g_assert_true (icipc_protocol_parse_reply_ok (b, sizeof(b), (const struct spa_pod **)&value));
+    g_assert_cmpint (value->value, ==, 3);
+  }
+}
+
+gint
+main (gint argc, gchar *argv[])
+{
+  g_test_init (&argc, &argv, NULL);
+
+  g_test_add_func ("/icipc/icipc-protocol", test_icipc_protocol);
+
+  return g_test_run ();
+}
diff --git a/tests/sender-receiver.c b/tests/sender-receiver.c
new file mode 100644
index 0000000..bd8721d
--- /dev/null
+++ b/tests/sender-receiver.c
@@ -0,0 +1,286 @@
+/* PipeWire AGL Cluster IPC
+ *
+ * Copyright © 2021 Collabora Ltd.
+ *    @author Julian Bouzas <julian.bouzas@collabora.com>
+ *
+ * SPDX-License-Identifier: MIT
+ */
+
+#include <glib.h>
+#include <icipc/icipc.h>
+
+#define TEST_ADDRESS "/tmp/icipc-sender-receiver"
+
+struct event_data {
+  const uint8_t * expected_data;
+  size_t expected_size;
+  int connections;
+  int n_events;
+  GMutex mutex;
+};
+
+static void
+wait_for_event (struct event_data *data, int n_events)
+{
+  while (true) {
+    g_mutex_lock (&data->mutex);
+    if (data->n_events == n_events) {
+      g_mutex_unlock (&data->mutex);
+      break;
+    }
+    g_mutex_unlock (&data->mutex);
+  }
+}
+
+static void
+sender_state_callback (struct icipc_receiver *self, int sender_fd,
+    enum icipc_receiver_sender_state sender_state, void *p)
+{
+  struct event_data *data = p;
+  g_assert_nonnull (data);
+
+  g_mutex_lock (&data->mutex);
+  switch (sender_state) {
+    case ICIPC_RECEIVER_SENDER_STATE_CONNECTED:
+      data->connections++;
+      break;
+    case ICIPC_RECEIVER_SENDER_STATE_DISCONNECTED:
+      data->connections--;
+      break;
+    default:
+      g_assert_not_reached ();
+      break;
+  }
+  data->n_events++;
+  g_mutex_unlock (&data->mutex);
+}
+
+static void
+reply_callback (struct icipc_sender *self, const uint8_t *buffer, size_t size, void *p)
+{
+  struct event_data *data = p;
+  g_assert_nonnull (data);
+  g_assert_nonnull (buffer);
+
+  g_mutex_lock (&data->mutex);
+  g_assert_cmpmem (buffer, size, data->expected_data, data->expected_size);
+  data->n_events++;
+  g_mutex_unlock (&data->mutex);
+}
+
+static void
+test_icipc_receiver_basic ()
+{
+  struct icipc_receiver *r = icipc_receiver_new (TEST_ADDRESS, 16, NULL, NULL, 0);
+  g_assert_nonnull (r);
+
+  /* start and stop */
+  g_assert_false (icipc_receiver_is_running (r));
+  g_assert_true (icipc_receiver_start (r));
+  g_assert_true (icipc_receiver_is_running (r));
+  icipc_receiver_stop (r);
+  g_assert_false (icipc_receiver_is_running (r));
+
+  /* clean up */
+  icipc_receiver_free (r);
+}
+
+static void
+test_icipc_sender_basic ()
+{
+  struct icipc_sender *s = icipc_sender_new (TEST_ADDRESS, 16, NULL, NULL, 0);
+  g_assert_nonnull (s);
+
+  /* clean up */
+  icipc_sender_free (s);
+}
+
+static void
+test_icipc_sender_connect ()
+{
+  static struct icipc_receiver_events events = {
+    .sender_state = sender_state_callback,
+    .handle_message = NULL,
+  };
+  struct event_data data;
+  g_mutex_init (&data.mutex);
+  data.n_events = 0;
+  data.connections = 0;
+  struct icipc_receiver *r = icipc_receiver_new (TEST_ADDRESS, 16, &events, &data, 0);
+  g_assert_nonnull (r);
+  struct icipc_sender *s = icipc_sender_new (TEST_ADDRESS, 16, NULL, NULL, 0);
+  g_assert_nonnull (s);
+
+  /* start receiver */
+  g_assert_true (icipc_receiver_start (r));
+
+  /* connect sender */
+  g_assert_true (icipc_sender_connect (s));
+  g_assert_true (icipc_sender_is_connected (s));
+  wait_for_event (&data, 1);
+  g_assert_cmpint (data.connections, ==, 1);
+
+  /* disconnect sender */
+  icipc_sender_disconnect (s);
+  g_assert_false (icipc_sender_is_connected (s));
+  wait_for_event (&data, 2);
+  g_assert_cmpint (data.connections, ==, 0);
+
+  /* stop receiver */
+  icipc_receiver_stop (r);
+
+  /* clean up */
+  g_mutex_clear (&data.mutex);
+  icipc_sender_free (s);
+  icipc_receiver_free (r);
+}
+
+static void
+lost_connection_handler (struct icipc_sender *self, int receiver_fd, void *p)
+{
+  struct event_data *data = p;
+  g_assert_nonnull (data);
+
+  g_mutex_lock (&data->mutex);
+  data->n_events++;
+  g_mutex_unlock (&data->mutex);
+}
+
+static void
+test_icipc_sender_lost_connection ()
+{
+  struct event_data data;
+  g_mutex_init (&data.mutex);
+  struct icipc_receiver *r = icipc_receiver_new (TEST_ADDRESS, 16, NULL, NULL, 0);
+  g_assert_nonnull (r);
+  struct icipc_sender *s = icipc_sender_new (TEST_ADDRESS, 16, lost_connection_handler, &data, 0);
+  g_assert_nonnull (s);
+
+  /* connect sender */
+  g_assert_true (icipc_sender_connect (s));
+  g_assert_true (icipc_sender_is_connected (s));
+
+  /* destroy receiver and make sure the lost connection handler is triggered */
+  data.n_events = 0;
+  icipc_receiver_free (r);
+  wait_for_event (&data, 1);
+
+  /* clean up */
+  g_mutex_clear (&data.mutex);
+  icipc_sender_free (s);
+}
+
+static void
+test_icipc_sender_send ()
+{
+  struct icipc_receiver *r = icipc_receiver_new (TEST_ADDRESS, 2, NULL, NULL, 0);
+  g_assert_nonnull (r);
+  struct icipc_sender *s = icipc_sender_new (TEST_ADDRESS, 2, NULL, NULL, 0);
+  g_assert_nonnull (s);
+  struct event_data data;
+  g_mutex_init (&data.mutex);
+  data.n_events = 0;
+
+  /* start receiver */
+  g_assert_true (icipc_receiver_start (r));
+
+  /* connect */
+  g_assert_true (icipc_sender_connect (s));
+  g_assert_true (icipc_sender_is_connected (s));
+
+  /* send 1 byte message (should not realloc) */
+  data.n_events = 0;
+  data.expected_data = (const uint8_t *)"h";
+  data.expected_size = 1;
+  g_assert_true (icipc_sender_send (s, (const uint8_t *)"h1", 1, reply_callback, &data));
+  wait_for_event (&data, 1);
+
+  /* send 2 bytes message (should realloc once to 4) */
+  data.n_events = 0;
+  data.expected_data = (const uint8_t *)"hi";
+  data.expected_size = 2;
+  g_assert_true (icipc_sender_send (s, (const uint8_t *)"hi", 2, reply_callback, &data));
+  wait_for_event (&data, 1);
+
+  /* send 3 bytes message (should not realloc) */
+  data.n_events = 0;
+  data.expected_data = (const uint8_t *)"hii";
+  data.expected_size = 3;
+  g_assert_true (icipc_sender_send (s, (const uint8_t *)"hii", 3, reply_callback, &data));
+  wait_for_event (&data, 1);
+
+  /* send 28 bytes message (should realloc 3 times: first to 8, then to 16 and finally to 32) */
+  data.n_events = 0;
+  data.expected_data = (const uint8_t *)"bigger than 16 bytes message";
+  data.expected_size = 28;
+  g_assert_true (icipc_sender_send (s, (const uint8_t *)"bigger than 16 bytes message", 28, reply_callback, &data));
+  wait_for_event (&data, 1);
+
+  /* don't allow empty messages */
+  data.n_events = 0;
+  g_assert_false (icipc_sender_send (s, (const uint8_t *)"", 0, NULL, NULL));
+
+  /* stop receiver */
+  icipc_receiver_stop (r);
+
+  /* clean up */
+  g_mutex_clear (&data.mutex);
+  icipc_sender_free (s);
+  icipc_receiver_free (r);
+}
+
+static void
+test_icipc_multiple_senders_send ()
+{
+  struct icipc_receiver *r = icipc_receiver_new (TEST_ADDRESS, 16, NULL, NULL, 0);
+  g_assert_nonnull (r);
+  struct icipc_sender *senders[50];
+  struct event_data data;
+  g_mutex_init (&data.mutex);
+  data.n_events = 0;
+
+  /* start receiver */
+  g_assert_true (icipc_receiver_start (r));
+
+  /* create and connect 50 senders */
+  for (int i = 0; i < 50; i++) {
+    senders[i] = icipc_sender_new (TEST_ADDRESS, 16, NULL, NULL, 0);
+    g_assert_nonnull (senders[i]);
+    g_assert_true (icipc_sender_connect (senders[i]));
+    g_assert_true (icipc_sender_is_connected (senders[i]));
+  }
+
+  /* send 50 messages (1 per sender) */
+  data.n_events = 0;
+  data.expected_data = (const uint8_t *)"hello";
+  data.expected_size = 5;
+  for (int i = 0; i < 50; i++)
+    g_assert_true (icipc_sender_send (senders[i], (const uint8_t *)"hello", 5, reply_callback, &data));
+  wait_for_event (&data, 50);
+
+  /* stop receiver */
+  icipc_receiver_stop (r);
+
+  /* clean up */
+  g_mutex_clear (&data.mutex);
+  for (int i = 0; i < 50; i++)
+    icipc_sender_free (senders[i]);
+  icipc_receiver_free (r);
+}
+
+gint
+main (gint argc, gchar *argv[])
+{
+  g_test_init (&argc, &argv, NULL);
+
+  g_test_add_func ("/icipc/receiver-basic", test_icipc_receiver_basic);
+  g_test_add_func ("/icipc/sender-basic", test_icipc_sender_basic);
+  g_test_add_func ("/icipc/sender-connect", test_icipc_sender_connect);
+  g_test_add_func ("/icipc/sender-lost-connection",
+      test_icipc_sender_lost_connection);
+  g_test_add_func ("/icipc/sender-send", test_icipc_sender_send);
+  g_test_add_func ("/icipc/multiple-senders-send",
+      test_icipc_multiple_senders_send);
+
+  return g_test_run ();
+}