Mainloop: use PinosLoop

Implement the main-loop with pinos_loop except for the parts that still
need to go through glib mainloop.
Start working on native protocol

13 files changed, 678 insertions, 311 deletions

diff --git a/pinos/client/loop.c b/pinos/client/loop.c
index e1d1cf5b..339d29ab 100644
--- a/pinos/client/loop.c
+++ b/pinos/client/loop.c
@@ -240,6 +240,10 @@ loop_enter (SpaLoopControl  *ctrl)
 {
   PinosLoopImpl *impl = SPA_CONTAINER_OF (ctrl, PinosLoopImpl, control);
   impl->thread = pthread_self();
+  if (impl->event == NULL)
+    impl->event = spa_loop_utils_add_event (&impl->utils,
+                                            event_func,
+                                            impl);
 }
 
 static void
@@ -599,9 +603,6 @@ pinos_loop_new (void)
 
   spa_ringbuffer_init (&impl->buffer, DATAS_SIZE);
 
-  impl->event = loop_add_event (&impl->utils,
-                                event_func,
-                                impl);
   return this;
 }
 
diff --git a/pinos/daemon/pinos.conf.in b/pinos/daemon/pinos.conf.in
index 21f19824..f0ddb414 100644
--- a/pinos/daemon/pinos.conf.in
+++ b/pinos/daemon/pinos.conf.in
@@ -1,4 +1,5 @@
 load-module libpinos-module-protocol-dbus
+load-module libpinos-module-protocol-native
 load-module libpinos-module-suspend-on-idle
 load-module libpinos-module-spa
 load-module libpinos-module-autolink
diff --git a/pinos/modules/meson.build b/pinos/modules/meson.build
index f3b728a9..59c1e91f 100644
--- a/pinos/modules/meson.build
+++ b/pinos/modules/meson.build
@@ -24,6 +24,15 @@ pinos_module_protocol_dbus = shared_library('pinos-module-protocol-dbus', [ 'mod
   dependencies : [gobject_dep, gmodule_dep, glib_dep, gio_dep, mathlib, dl_lib, pinos_dep, pinoscore_dep],
 )
 
+pinos_module_protocol_native = shared_library('pinos-module-protocol-native', [ 'module-protocol-native.c' ],
+  c_args : pinos_module_c_args,
+  include_directories : [configinc, pinosinc, spa_inc],
+  link_with : spalib,
+  install : true,
+  install_dir : '@0@/pinos-0.1'.format(get_option('libdir')),
+  dependencies : [gobject_dep, gmodule_dep, glib_dep, gio_dep, mathlib, dl_lib, pinos_dep, pinoscore_dep],
+)
+
 pinos_module_suspend_on_idle = shared_library('pinos-module-suspend-on-idle', [ 'module-suspend-on-idle.c' ],
   c_args : pinos_module_c_args,
   include_directories : [configinc, pinosinc, spa_inc],
diff --git a/pinos/modules/module-protocol-dbus.c b/pinos/modules/module-protocol-dbus.c
index 35554c36..d8c506d2 100644
--- a/pinos/modules/module-protocol-dbus.c
+++ b/pinos/modules/module-protocol-dbus.c
@@ -44,18 +44,12 @@
 #define PINOS_PROTOCOL_DBUS_URI                            "http://pinos.org/ns/protocol-dbus"
 #define PINOS_PROTOCOL_DBUS_PREFIX                         PINOS_PROTOCOL_DBUS_URI "#"
 
-typedef struct _PinosProtocolDBus PinosProtocolDBus;
-
-struct _PinosProtocolDBus {
+typedef struct {
   PinosCore   *core;
   SpaList      link;
   PinosGlobal *global;
 
   PinosProperties *properties;
-};
-
-typedef struct {
-  PinosProtocolDBus this;
 
   struct {
     uint32_t protocol_dbus;
@@ -70,16 +64,16 @@ typedef struct {
   PinosListener global_added;
   PinosListener global_removed;
   PinosListener node_state_changed;
-} PinosProtocolDBusImpl;
+} PinosProtocolDBus;
 
 typedef struct {
-  PinosProtocolDBusImpl *impl;
-  SpaList                link;
-  PinosGlobal           *global;
-  void                  *iface;
-  PinosObjectSkeleton   *skel;
-  const gchar           *object_path;
-  PinosDestroy           destroy;
+  PinosProtocolDBus   *impl;
+  SpaList              link;
+  PinosGlobal         *global;
+  void                *iface;
+  PinosObjectSkeleton *skel;
+  const gchar         *object_path;
+  PinosDestroy         destroy;
 } PinosProtocolDBusObject;
 
 typedef struct {
@@ -117,13 +111,13 @@ object_unexport (PinosProtocolDBusObject *this)
 }
 
 static void *
-object_new (size_t                 size,
-            PinosProtocolDBusImpl *impl,
-            PinosGlobal           *global,
-            void                  *iface,
-            PinosObjectSkeleton   *skel,
-            bool                   export,
-            PinosDestroy           destroy)
+object_new (size_t               size,
+            PinosProtocolDBus   *impl,
+            PinosGlobal         *global,
+            void                *iface,
+            PinosObjectSkeleton *skel,
+            bool                 export,
+            PinosDestroy         destroy)
 {
   PinosProtocolDBusObject *this;
 
@@ -157,8 +151,8 @@ object_destroy (PinosProtocolDBusObject *this)
 }
 
 static PinosProtocolDBusObject *
-find_object (PinosProtocolDBusImpl *impl,
-             void                  *object)
+find_object (PinosProtocolDBus *impl,
+             void              *object)
 {
   PinosProtocolDBusObject *obj;
   spa_list_for_each (obj, &impl->object_list, link) {
@@ -201,14 +195,13 @@ client_name_vanished_handler (GDBusConnection *connection,
 
 
 static PinosProtocolDBusClient *
-client_new (PinosProtocolDBusImpl *impl,
-            const char            *sender)
+client_new (PinosProtocolDBus *impl,
+            const char        *sender)
 {
-  PinosProtocolDBus *proto = &impl->this;
   PinosProtocolDBusClient *this;
   PinosClient *client;
 
-  client = pinos_client_new (proto->core, NULL);
+  client = pinos_client_new (impl->core, NULL);
 
   if ((this = (PinosProtocolDBusClient *) find_object (impl, client))) {
     pinos_client1_set_sender (this->parent.iface, sender);
@@ -228,11 +221,10 @@ client_new (PinosProtocolDBusImpl *impl,
 }
 
 static PinosClient *
-sender_get_client (PinosProtocolDBus *proto,
+sender_get_client (PinosProtocolDBus *impl,
                    const char        *sender,
                    bool               create)
 {
-  PinosProtocolDBusImpl *impl = SPA_CONTAINER_OF (proto, PinosProtocolDBusImpl, this);
   PinosProtocolDBusClient *client;
 
   spa_list_for_each (client, &impl->client_list, link) {
@@ -247,19 +239,6 @@ sender_get_client (PinosProtocolDBus *proto,
   return client->parent.global->object;
 }
 
-static PinosNodeFactory *
-find_factory_by_name (PinosProtocolDBus *proto,
-                      const char  *name)
-{
-  PinosNodeFactory *factory;
-
-  spa_list_for_each (factory, &proto->core->node_factory_list, link) {
-    if (strcmp (factory->name, name) == 0)
-      return factory;
-  }
-  return NULL;
-}
-
 static bool
 handle_create_node (PinosDaemon1           *interface,
                     GDBusMethodInvocation  *invocation,
@@ -268,8 +247,7 @@ handle_create_node (PinosDaemon1           *interface,
                     GVariant               *arg_properties,
                     gpointer                user_data)
 {
-  PinosProtocolDBusImpl *impl = user_data;
-  PinosProtocolDBus *this = &impl->this;
+  PinosProtocolDBus *impl = user_data;
   PinosNodeFactory *factory;
   PinosNode *node;
   PinosClient *client;
@@ -278,13 +256,13 @@ handle_create_node (PinosDaemon1           *interface,
   PinosProtocolDBusObject *object;
 
   sender = g_dbus_method_invocation_get_sender (invocation);
-  client = sender_get_client (this, sender, TRUE);
+  client = sender_get_client (impl, sender, TRUE);
 
   pinos_log_debug ("protocol-dbus %p: create node: %s", impl, sender);
 
   props = pinos_properties_from_variant (arg_properties);
 
-  factory = find_factory_by_name (this, arg_factory_name);
+  factory = pinos_core_find_node_factory (impl->core, arg_factory_name);
   if (factory == NULL)
     goto no_factory;
 
@@ -302,7 +280,7 @@ handle_create_node (PinosDaemon1           *interface,
     goto object_failed;
 
   pinos_client_add_resource (client,
-                             this->core->uri.node,
+                             impl->core->uri.node,
                              node,
                              (PinosDestroy) pinos_node_destroy);
 
@@ -342,7 +320,7 @@ on_node_state_changed (PinosListener  *listener,
                        PinosNodeState  old,
                        PinosNodeState  state)
 {
-  PinosProtocolDBusImpl *impl = SPA_CONTAINER_OF (listener, PinosProtocolDBusImpl, node_state_changed);
+  PinosProtocolDBus *impl = SPA_CONTAINER_OF (listener, PinosProtocolDBus, node_state_changed);
   PinosProtocolDBusObject *object;
 
   pinos_log_debug ("protocol-dbus %p: node %p state change %s -> %s", impl, node,
@@ -361,8 +339,7 @@ handle_create_client_node (PinosDaemon1           *interface,
                            GVariant               *arg_properties,
                            gpointer                user_data)
 {
-  PinosProtocolDBusImpl *impl = user_data;
-  PinosProtocolDBus *this = &impl->this;
+  PinosProtocolDBus *impl = user_data;
   PinosClientNode *node;
   PinosClient *client;
   SpaResult res;
@@ -375,7 +352,7 @@ handle_create_client_node (PinosDaemon1           *interface,
   PinosProtocolDBusObject *object;
 
   sender = g_dbus_method_invocation_get_sender (invocation);
-  client = sender_get_client (this, sender, TRUE);
+  client = sender_get_client (impl, sender, TRUE);
 
   pinos_log_debug ("protocol-dbus %p: create client-node: %s", impl, sender);
   props = pinos_properties_from_variant (arg_properties);
@@ -388,7 +365,7 @@ handle_create_client_node (PinosDaemon1           *interface,
   }
 
 
-  node = pinos_client_node_new (this->core,
+  node = pinos_client_node_new (impl->core,
                                 arg_name,
                                 props);
 
@@ -403,7 +380,7 @@ handle_create_client_node (PinosDaemon1           *interface,
     goto no_socket;
 
   pinos_client_add_resource (client,
-                             this->core->uri.client_node,
+                             impl->core->uri.client_node,
                              node,
                              (PinosDestroy) pinos_client_node_destroy);
 
@@ -444,7 +421,7 @@ bus_acquired_handler (GDBusConnection *connection,
                       const char      *name,
                       gpointer         user_data)
 {
-  PinosProtocolDBusImpl *impl = user_data;
+  PinosProtocolDBus *impl = user_data;
   GDBusObjectManagerServer *manager = impl->server_manager;
 
   impl->connection = connection;
@@ -463,7 +440,7 @@ name_lost_handler (GDBusConnection *connection,
                    const char      *name,
                    gpointer         user_data)
 {
-  PinosProtocolDBusImpl *impl = user_data;
+  PinosProtocolDBus *impl = user_data;
   GDBusObjectManagerServer *manager = impl->server_manager;
 
   g_dbus_object_manager_server_set_connection (manager, connection);
@@ -489,11 +466,10 @@ on_global_added (PinosListener *listener,
                  PinosCore     *core,
                  PinosGlobal   *global)
 {
-  PinosProtocolDBusImpl *impl = SPA_CONTAINER_OF (listener, PinosProtocolDBusImpl, global_added);
-  PinosProtocolDBus *this = &impl->this;
+  PinosProtocolDBus *impl = SPA_CONTAINER_OF (listener, PinosProtocolDBus, global_added);
   PinosObjectSkeleton *skel;
 
-  if (global->type == this->core->uri.client) {
+  if (global->type == impl->core->uri.client) {
     PinosClient1 *iface;
     PinosClient *client = global->object;
     PinosProperties *props = client->properties;
@@ -507,15 +483,15 @@ on_global_added (PinosListener *listener,
     pinos_client1_set_properties (iface, props ? pinos_properties_to_variant (props) : NULL);
     pinos_object_skeleton_set_client1 (skel, iface);
 
-    this = object_new (sizeof (PinosProtocolDBusClient),
-                       impl,
-                       global,
-                       iface,
-                       skel,
-                       false,
-                       (PinosDestroy) client_destroy);
+    object_new (sizeof (PinosProtocolDBusClient),
+                impl,
+                global,
+                iface,
+                skel,
+                false,
+                (PinosDestroy) client_destroy);
 
-  } else if (global->type == this->core->uri.node) {
+  } else if (global->type == impl->core->uri.node) {
     PinosNode1 *iface;
     PinosNode *node = global->object;
     PinosProperties *props = node->properties;
@@ -582,7 +558,7 @@ on_global_added (PinosListener *listener,
                                  name_lost_handler,
                                  proto,
                                  NULL);
-  } else if (global->type == this->core->uri.link) {
+  } else if (global->type == impl->core->uri.link) {
     PinosLink1 *iface;
     PinosLink *link = global->object;
     PinosProtocolDBusObject *obj;
@@ -627,7 +603,7 @@ on_global_removed (PinosListener *listener,
                    PinosCore     *core,
                    PinosGlobal   *global)
 {
-  PinosProtocolDBusImpl *impl = SPA_CONTAINER_OF (listener, PinosProtocolDBusImpl, global_removed);
+  PinosProtocolDBus *impl = SPA_CONTAINER_OF (listener, PinosProtocolDBus, global_removed);
   PinosProtocolDBusObject *object;
 
   if ((object = find_object (impl, global->object)))
@@ -638,15 +614,13 @@ static PinosProtocolDBus *
 pinos_protocol_dbus_new (PinosCore       *core,
                          PinosProperties *properties)
 {
-  PinosProtocolDBusImpl *impl;
-  PinosProtocolDBus *this;
+  PinosProtocolDBus *impl;
 
-  impl = calloc (1, sizeof (PinosProtocolDBusImpl));
-  this = &impl->this;
+  impl = calloc (1, sizeof (PinosProtocolDBus));
   pinos_log_debug ("protocol-dbus %p: new", impl);
 
-  this->core = core;
-  this->properties = properties;
+  impl->core = core;
+  impl->properties = properties;
 
   spa_list_init (&impl->client_list);
   spa_list_init (&impl->object_list);
@@ -659,22 +633,21 @@ pinos_protocol_dbus_new (PinosCore       *core,
 
   impl->uri.protocol_dbus = spa_id_map_get_id (core->uri.map, PINOS_PROTOCOL_DBUS_URI);
 
-  this->global = pinos_core_add_global (core,
+  impl->global = pinos_core_add_global (core,
                                         impl->uri.protocol_dbus,
-                                        this);
-  return this;
+                                        impl);
+  return impl;
 }
 
 #if 0
 static void
-pinos_protocol_dbus_destroy (PinosProtocolDBus *proto)
+pinos_protocol_dbus_destroy (PinosProtocolDBus *impl)
 {
-  PinosProtocolDBusImpl *impl = SPA_CONTAINER_OF (proto, PinosProtocolDBusImpl, this);
   PinosProtocolDBusObject *object, *tmp;
 
   pinos_log_debug ("protocol-dbus %p: destroy", impl);
 
-  pinos_global_destroy (proto->global);
+  pinos_global_destroy (impl->global);
 
   spa_list_for_each_safe (object, tmp, &impl->object_list, link)
     object_destroy (object);
diff --git a/pinos/modules/module-protocol-native.c b/pinos/modules/module-protocol-native.c
new file mode 100644
index 00000000..2c7b6fd0
--- /dev/null
+++ b/pinos/modules/module-protocol-native.c
@@ -0,0 +1,476 @@
+/* Pinos
+ * Copyright (C) 2015 Wim Taymans <wim.taymans@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#include <string.h>
+#include <stdio.h>
+#include <errno.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <sys/file.h>
+
+#include "config.h"
+
+#include "pinos/client/pinos.h"
+#include "pinos/client/log.h"
+
+#include "pinos/server/core.h"
+#include "pinos/server/node.h"
+#include "pinos/server/module.h"
+#include "pinos/server/client-node.h"
+#include "pinos/server/client.h"
+#include "pinos/server/link.h"
+#include "pinos/server/node-factory.h"
+#include "pinos/server/data-loop.h"
+#include "pinos/server/main-loop.h"
+
+#define PINOS_PROTOCOL_NATIVE_URI                            "http://pinos.org/ns/protocol-native"
+#define PINOS_PROTOCOL_NATIVE_PREFIX                         PINOS_PROTOCOL_NATIVE_URI "#"
+
+#ifndef UNIX_PATH_MAX
+#define UNIX_PATH_MAX   108
+#endif
+
+#define LOCK_SUFFIX     ".lock"
+#define LOCK_SUFFIXLEN  5
+
+typedef struct {
+  int        fd;
+  int        fd_lock;
+  struct     sockaddr_un addr;
+  char       lock_addr[UNIX_PATH_MAX + LOCK_SUFFIXLEN];
+  PinosLoop *loop;
+  SpaSource *source;
+  char      *core_name;
+  SpaList    link;
+} Socket;
+
+typedef struct {
+  PinosCore   *core;
+  SpaList      link;
+  PinosGlobal *global;
+
+  PinosProperties *properties;
+
+  struct {
+    uint32_t protocol_native;
+  } uri;
+
+  SpaList socket_list;
+
+  SpaList client_list;
+  SpaList object_list;
+
+  PinosListener global_added;
+  PinosListener global_removed;
+  PinosListener node_state_changed;
+} PinosProtocolNative;
+
+typedef struct {
+  PinosProtocolNative *impl;
+  SpaList              link;
+  PinosGlobal         *global;
+  PinosDestroy         destroy;
+} PinosProtocolNativeObject;
+
+typedef struct {
+  PinosProtocolNativeObject parent;
+  SpaList                   link;
+} PinosProtocolNativeServer;
+
+typedef struct {
+  PinosProtocolNativeObject parent;
+  SpaList                   link;
+  int                       fd;
+  struct ucred              ucred;
+  SpaSource                *source;
+  PinosConnection          *connection;
+} PinosProtocolNativeClient;
+
+typedef struct {
+  PinosProtocolNativeObject parent;
+  PinosListener             state_changed;
+} PinosProtocolNativeNode;
+
+static void *
+object_new (size_t               size,
+            PinosProtocolNative *impl,
+            PinosGlobal         *global,
+            PinosDestroy         destroy)
+{
+  PinosProtocolNativeObject *this;
+
+  this = calloc (1, size);
+  this->impl = impl;
+  this->global = global;
+  this->destroy = destroy;
+
+  spa_list_insert (impl->object_list.prev, &this->link);
+
+  return this;
+}
+
+static void
+object_destroy (PinosProtocolNativeObject *this)
+{
+  spa_list_remove (&this->link);
+
+  if (this->destroy)
+    this->destroy (this);
+  free (this);
+}
+
+static PinosProtocolNativeObject *
+find_object (PinosProtocolNative *impl,
+             void                *object)
+{
+  PinosProtocolNativeObject *obj;
+  spa_list_for_each (obj, &impl->object_list, link) {
+    if (obj->global->object == object)
+      return obj;
+  }
+  return NULL;
+}
+
+static void
+client_destroy (PinosProtocolNativeClient *this)
+{
+  spa_list_remove (&this->link);
+  close (this->fd);
+}
+
+static void
+connection_data (SpaSource *source,
+                 int        fd,
+                 SpaIO      mask,
+                 void      *data)
+{
+  PinosProtocolNativeClient *client = data;
+  PinosConnection *conn = client->connection;
+
+  while (pinos_connection_has_next (conn)) {
+    PinosMessageType type = pinos_connection_get_type (conn);
+
+    switch (type) {
+      default:
+        pinos_log_error ("unhandled message %d", type);
+        break;
+    }
+  }
+}
+
+static PinosProtocolNativeClient *
+client_new (PinosProtocolNative *impl,
+            int                  fd)
+{
+  PinosProtocolNativeClient *this;
+  PinosClient *client;
+  socklen_t len;
+
+  client = pinos_client_new (impl->core, NULL);
+
+  if ((this = (PinosProtocolNativeClient *) find_object (impl, client)) == NULL) {
+    close (fd);
+    return NULL;
+  }
+
+  this->fd = fd;
+  this->source = pinos_loop_add_io (impl->core->main_loop->loop,
+                                    this->fd,
+                                    SPA_IO_IN,
+                                    false,
+                                    connection_data,
+                                    this);
+
+  len = sizeof (this->ucred);
+  if (getsockopt (fd, SOL_SOCKET, SO_PEERCRED, &this->ucred, &len) < 0) {
+    pinos_log_error ("no peercred: %m");
+  }
+  this->connection = pinos_connection_new (fd);
+
+  spa_list_insert (impl->client_list.prev, &this->link);
+
+  return this;
+
+}
+
+static void
+on_global_added (PinosListener *listener,
+                 PinosCore     *core,
+                 PinosGlobal   *global)
+{
+  PinosProtocolNative *impl = SPA_CONTAINER_OF (listener, PinosProtocolNative, global_added);
+
+  if (global->type == impl->core->uri.client) {
+    object_new (sizeof (PinosProtocolNativeClient),
+                impl,
+                global,
+                (PinosDestroy) client_destroy);
+  } else if (global->type == impl->core->uri.node) {
+  } else if (global->type == impl->uri.protocol_native) {
+  } else if (global->type == impl->core->uri.link) {
+  }
+}
+
+static void
+on_global_removed (PinosListener *listener,
+                   PinosCore     *core,
+                   PinosGlobal   *global)
+{
+  PinosProtocolNative *impl = SPA_CONTAINER_OF (listener, PinosProtocolNative, global_added);
+  PinosProtocolNativeObject *object;
+
+  if ((object = find_object (impl, global->object))) {
+    object_destroy (object);
+  }
+}
+
+static Socket *
+create_socket (void)
+{
+  Socket *s;
+
+  if ((s = calloc(1, sizeof (Socket))) == NULL)
+    return NULL;
+
+  s->fd = -1;
+  s->fd_lock = -1;
+  return s;
+}
+
+static void
+destroy_socket (Socket *s)
+{
+  if (s->source)
+    pinos_loop_destroy_source (s->loop, s->source);
+  if (s->addr.sun_path[0])
+    unlink (s->addr.sun_path);
+  if (s->fd >= 0)
+    close (s->fd);
+  if (s->lock_addr[0])
+    unlink (s->lock_addr);
+  if (s->fd_lock >= 0)
+    close (s->fd_lock);
+  free (s);
+}
+
+static bool
+init_socket_name (Socket *s, const char *name)
+{
+  int name_size;
+  const char *runtime_dir;
+
+  if ((runtime_dir = getenv ("XDG_RUNTIME_DIR")) == NULL) {
+    pinos_log_error ("XDG_RUNTIME_DIR not set in the environment");
+    return false;
+  }
+
+  s->addr.sun_family = AF_LOCAL;
+  name_size = snprintf (s->addr.sun_path, sizeof (s->addr.sun_path),
+                             "%s/%s", runtime_dir, name) + 1;
+
+  s->core_name = (s->addr.sun_path + name_size - 1) - strlen (name);
+
+  if (name_size > (int)sizeof (s->addr.sun_path)) {
+    pinos_log_error ("socket path \"%s/%s\" plus null terminator exceeds 108 bytes",
+                     runtime_dir, name);
+    *s->addr.sun_path = 0;
+    return false;
+  }
+  return true;
+}
+
+static bool
+lock_socket (Socket *s)
+{
+  struct stat socket_stat;
+
+  snprintf (s->lock_addr, sizeof (s->lock_addr),
+            "%s%s", s->addr.sun_path, LOCK_SUFFIX);
+
+  s->fd_lock = open (s->lock_addr, O_CREAT | O_CLOEXEC,
+                     (S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP));
+
+  if (s->fd_lock < 0) {
+    pinos_log_error ("unable to open lockfile %s check permissions", s->lock_addr);
+    goto err;
+  }
+
+  if (flock (s->fd_lock, LOCK_EX | LOCK_NB) < 0) {
+    pinos_log_error ("unable to lock lockfile %s, maybe another daemon is running", s->lock_addr);
+    goto err_fd;
+  }
+
+  if (stat (s->addr.sun_path, &socket_stat) < 0 ) {
+    if (errno != ENOENT) {
+      pinos_log_error ("did not manage to stat file %s\n", s->addr.sun_path);
+      goto err_fd;
+    }
+  } else if (socket_stat.st_mode & S_IWUSR ||
+             socket_stat.st_mode & S_IWGRP) {
+    unlink (s->addr.sun_path);
+  }
+  return true;
+
+err_fd:
+  close (s->fd_lock);
+  s->fd_lock = -1;
+err:
+  *s->lock_addr = 0;
+  *s->addr.sun_path = 0;
+  return false;
+}
+
+static void
+socket_data (SpaSource *source,
+             int        fd,
+             SpaIO      mask,
+             void      *data)
+{
+  PinosProtocolNative *impl = data;
+  struct sockaddr_un name;
+  socklen_t length;
+  int client_fd;
+
+  length = sizeof (name);
+  client_fd = accept4 (fd, (struct sockaddr *) &name, &length, SOCK_CLOEXEC);
+  if (client_fd < 0) {
+    pinos_log_error ("failed to accept: %m");
+    return;
+  }
+
+  client_new (impl, client_fd);
+}
+
+static bool
+add_socket (PinosProtocolNative *impl, Socket *s)
+{
+  socklen_t size;
+
+  if ((s->fd = socket (PF_LOCAL, SOCK_STREAM | SOCK_CLOEXEC, 0)) < 0)
+    return false;
+
+  size = offsetof (struct sockaddr_un, sun_path) + strlen (s->addr.sun_path);
+  if (bind (s->fd, (struct sockaddr *) &s->addr, size) < 0) {
+    pinos_log_error ("bind() failed with error: %m");
+    return false;
+  }
+
+  if (listen (s->fd, 128) < 0) {
+    pinos_log_error ("listen() failed with error: %m");
+    return false;
+  }
+
+  s->loop = impl->core->main_loop->loop;
+  s->source = pinos_loop_add_io (s->loop,
+                                 s->fd,
+                                 SPA_IO_IN,
+                                 false,
+                                 socket_data,
+                                 impl);
+  if (s->source == NULL)
+    return false;
+
+  spa_list_insert (impl->socket_list.prev, &s->link);
+
+  return true;
+}
+
+
+static PinosProtocolNative *
+pinos_protocol_native_new (PinosCore       *core,
+                           PinosProperties *properties)
+{
+  PinosProtocolNative *impl;
+  Socket *s;
+  const char *name;
+
+  impl = calloc (1, sizeof (PinosProtocolNative));
+  pinos_log_debug ("protocol-native %p: new", impl);
+
+  impl->core = core;
+  impl->properties = properties;
+
+  name = NULL;
+  if (impl->properties)
+    name = pinos_properties_get (impl->properties, "pinos.core.name");
+  if (name == NULL)
+    name = getenv ("PINOS_CORE");
+  if (name == NULL)
+    name = "pinos-0";
+
+  s = create_socket ();
+
+  spa_list_init (&impl->socket_list);
+  spa_list_init (&impl->client_list);
+  spa_list_init (&impl->object_list);
+
+  if (!init_socket_name (s, name))
+    goto error;
+
+  if (!lock_socket (s))
+    goto error;
+
+  if (!add_socket (impl, s))
+    goto error;
+
+  pinos_signal_add (&core->global_added, &impl->global_added, on_global_added);
+  pinos_signal_add (&core->global_removed, &impl->global_removed, on_global_removed);
+
+  impl->uri.protocol_native = spa_id_map_get_id (core->uri.map, PINOS_PROTOCOL_NATIVE_URI);
+
+  impl->global = pinos_core_add_global (core,
+                                        impl->uri.protocol_native,
+                                        impl);
+  return impl;
+
+error:
+  destroy_socket (s);
+  free (impl);
+  return NULL;
+}
+
+#if 0
+static void
+pinos_protocol_native_destroy (PinosProtocolNative *impl)
+{
+  PinosProtocolNativeObject *object, *tmp;
+
+  pinos_log_debug ("protocol-native %p: destroy", impl);
+
+  pinos_global_destroy (impl->global);
+
+  spa_list_for_each_safe (object, tmp, &impl->object_list, link)
+    object_destroy (object);
+
+  pinos_signal_remove (&impl->global_added);
+  pinos_signal_remove (&impl->global_removed);
+  pinos_signal_remove (&impl->node_state_changed);
+
+  free (impl);
+}
+#endif
+
+bool
+pinos__module_init (PinosModule * module, const char * args)
+{
+  pinos_protocol_native_new (module->core, NULL);
+  return TRUE;
+}
diff --git a/pinos/modules/module-suspend-on-idle.c b/pinos/modules/module-suspend-on-idle.c
index e90afb62..11090d8b 100644
--- a/pinos/modules/module-suspend-on-idle.c
+++ b/pinos/modules/module-suspend-on-idle.c
@@ -50,8 +50,7 @@ typedef struct {
   ModuleImpl  *impl;
   PinosNode   *node;
   SpaList      link;
-  SpaSource   *timeout;
-  guint        idle_timeout;
+  SpaSource   *idle_timeout;
 } NodeInfo;
 
 static NodeInfo *
@@ -70,18 +69,20 @@ static void
 remove_idle_timeout (NodeInfo *info)
 {
   if (info->idle_timeout) {
-    g_source_remove (info->idle_timeout);
-    info->idle_timeout = 0;
+    pinos_loop_destroy_source (info->impl->core->main_loop->loop, info->idle_timeout);
+    info->idle_timeout = NULL;
   }
 }
 
-static bool
-idle_timeout (NodeInfo *info)
+static void
+idle_timeout (SpaSource *source,
+              void      *data)
 {
-  info->idle_timeout = 0;
+  NodeInfo *info = data;
+
   pinos_log_debug ("module %p: node %p idle timeout", info->impl, info->node);
+  remove_idle_timeout (info);
   pinos_node_set_state (info->node, PINOS_NODE_STATE_SUSPENDED);
-  return G_SOURCE_REMOVE;
 }
 
 static void
@@ -113,10 +114,19 @@ on_node_state_changed (PinosListener  *listener,
   if (state != PINOS_NODE_STATE_IDLE) {
     remove_idle_timeout (info);
   } else {
+    struct timespec value;
+
     pinos_log_debug ("module %p: node %p became idle", impl, node);
-    info->idle_timeout = g_timeout_add_seconds (3,
-                                                (GSourceFunc) idle_timeout,
-                                                info);
+    info->idle_timeout = pinos_loop_add_timer (impl->core->main_loop->loop,
+                                              idle_timeout,
+                                              info);
+    value.tv_sec = 3;
+    value.tv_nsec = 0;
+    pinos_loop_update_timer (impl->core->main_loop->loop,
+                             info->idle_timeout,
+                             &value,
+                             NULL,
+                             false);
   }
 }
 
@@ -134,7 +144,6 @@ on_global_added (PinosListener *listener,
     info = calloc (1, sizeof (NodeInfo));
     info->impl = impl;
     info->node = node;
-    info->timeout = NULL;
     spa_list_insert (impl->node_list.prev, &info->link);
   }
 }
diff --git a/pinos/server/core.c b/pinos/server/core.c
index 6c0732cd..c4a3e4a3 100644
--- a/pinos/server/core.c
+++ b/pinos/server/core.c
@@ -50,7 +50,7 @@ pinos_core_new (PinosMainLoop *main_loop)
   impl->support[2].uri = SPA_LOOP__DataLoop;
   impl->support[2].data = this->data_loop->loop->loop;
   impl->support[3].uri = SPA_LOOP__MainLoop;
-  impl->support[3].data = this->main_loop->loop;
+  impl->support[3].data = this->main_loop->loop->loop;
   this->support = impl->support;
   this->n_support = 4;
 
@@ -162,3 +162,16 @@ pinos_core_find_port (PinosCore       *core,
   }
   return best;
 }
+
+PinosNodeFactory *
+pinos_core_find_node_factory (PinosCore  *core,
+                              const char *name)
+{
+  PinosNodeFactory *factory;
+
+  spa_list_for_each (factory, &core->node_factory_list, link) {
+    if (strcmp (factory->name, name) == 0)
+      return factory;
+  }
+  return NULL;
+}
diff --git a/pinos/server/core.h b/pinos/server/core.h
index f1b16e63..03406f40 100644
--- a/pinos/server/core.h
+++ b/pinos/server/core.h
@@ -34,6 +34,7 @@ typedef struct _PinosGlobal PinosGlobal;
 #include <pinos/server/uri.h>
 #include <pinos/server/node.h>
 #include <pinos/server/link.h>
+#include <pinos/server/node-factory.h>
 
 struct _PinosGlobal {
   PinosCore *core;
@@ -119,6 +120,9 @@ PinosPort *     pinos_core_find_port     (PinosCore        *core,
                                           SpaFormat       **format_filter,
                                           char            **error);
 
+PinosNodeFactory * pinos_core_find_node_factory (PinosCore  *core,
+                                                 const char *name);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/pinos/server/link.c b/pinos/server/link.c
index 15a46620..29450fc0 100644
--- a/pinos/server/link.c
+++ b/pinos/server/link.c
@@ -795,12 +795,12 @@ do_link_remove (SpaLoop        *loop,
     this->rt.output = NULL;
   }
 
-  res = spa_loop_invoke (this->core->main_loop->loop,
-                         do_link_remove_done,
-                         seq,
-                         0,
-                         NULL,
-                         this);
+  res = pinos_loop_invoke (this->core->main_loop->loop,
+                           do_link_remove_done,
+                           seq,
+                           0,
+                           NULL,
+                           this);
   return res;
 }
 
@@ -826,23 +826,23 @@ pinos_link_destroy (PinosLink * this)
     pinos_signal_remove (&impl->input_port_destroy);
     pinos_signal_remove (&impl->input_async_complete);
 
-    res = spa_loop_invoke (this->input->node->data_loop->loop->loop,
-                           do_link_remove,
-                           impl->seq++,
-                           0,
-                           NULL,
-                           this);
+    res = pinos_loop_invoke (this->input->node->data_loop->loop,
+                             do_link_remove,
+                             impl->seq++,
+                             0,
+                             NULL,
+                             this);
   }
   if (this->output) {
     pinos_signal_remove (&impl->output_port_destroy);
     pinos_signal_remove (&impl->output_async_complete);
 
-    res = spa_loop_invoke (this->output->node->data_loop->loop->loop,
-                           do_link_remove,
-                           impl->seq++,
-                           0,
-                           NULL,
-                           this);
+    res = pinos_loop_invoke (this->output->node->data_loop->loop,
+                             do_link_remove,
+                             impl->seq++,
+                             0,
+                             NULL,
+                             this);
   }
   return res;
 }
diff --git a/pinos/server/main-loop.c b/pinos/server/main-loop.c
index cb3e5f8a..2b2af039 100644
--- a/pinos/server/main-loop.c
+++ b/pinos/server/main-loop.c
@@ -30,17 +30,6 @@
 #include "pinos/client/object.h"
 #include "pinos/server/main-loop.h"
 
-#define DATAS_SIZE (4096 * 8)
-
-typedef struct {
-  size_t         item_size;
-  SpaInvokeFunc  func;
-  uint32_t       seq;
-  size_t         size;
-  void          *data;
-  void          *user_data;
-} InvokeItem;
-
 typedef struct _WorkItem WorkItem;
 
 struct _WorkItem {
@@ -50,7 +39,6 @@ struct _WorkItem {
   SpaResult       res;
   PinosDeferFunc  func;
   void           *data;
-  bool            sync;
   SpaList         link;
 };
 
@@ -58,18 +46,13 @@ typedef struct
 {
   PinosMainLoop this;
 
-  SpaLoop loop;
+  PinosLoop *loop;
 
   GMainContext *context;
   GMainLoop *main_loop;
 
   uint32_t counter;
 
-  SpaRingbuffer buffer;
-  uint8_t       buffer_data[DATAS_SIZE];
-
-  SpaSource wakeup;
-
   SpaList  work_list;
   SpaList  free_list;
 
@@ -152,79 +135,6 @@ do_remove_source (SpaSource *source)
   g_source_destroy (gsource);
 }
 
-static void
-main_loop_dispatch (SpaSource *source)
-{
-  LoopData *data = source->loop_private;
-  PinosMainLoopImpl *impl = data->impl;
-  uint64_t u;
-  size_t offset;
-  InvokeItem *item;
-
-  if (read (impl->wakeup.fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
-    pinos_log_warn ("main-loop %p: failed to read fd", strerror (errno));
-
-  while (spa_ringbuffer_get_read_offset (&impl->buffer, &offset) > 0) {
-    item = SPA_MEMBER (impl->buffer_data, offset, InvokeItem);
-    item->func (&impl->loop, true, item->seq, item->size, item->data, item->user_data);
-    spa_ringbuffer_read_advance (&impl->buffer, item->item_size);
-  }
-}
-
-static SpaResult
-do_invoke (SpaLoop       *loop,
-           SpaInvokeFunc  func,
-           uint32_t       seq,
-           size_t         size,
-           void          *data,
-           void          *user_data)
-{
-  PinosMainLoopImpl *impl = SPA_CONTAINER_OF (loop, PinosMainLoopImpl, loop);
-  bool in_thread = false;
-  SpaRingbufferArea areas[2];
-  InvokeItem *item;
-  uint64_t u = 1;
-  SpaResult res;
-
-  if (in_thread) {
-    res = func (loop, false, seq, size, data, user_data);
-  } else {
-    spa_ringbuffer_get_write_areas (&impl->buffer, areas);
-    if (areas[0].len < sizeof (InvokeItem)) {
-      pinos_log_warn ("queue full");
-      return SPA_RESULT_ERROR;
-    }
-    item = SPA_MEMBER (impl->buffer_data, areas[0].offset, InvokeItem);
-    item->func = func;
-    item->seq = seq;
-    item->size = size;
-    item->user_data = user_data;
-
-    if (areas[0].len > sizeof (InvokeItem) + size) {
-      item->data = SPA_MEMBER (item, sizeof (InvokeItem), void);
-      item->item_size = sizeof (InvokeItem) + size;
-      if (areas[0].len < sizeof (InvokeItem) + item->item_size)
-        item->item_size = areas[0].len;
-    } else {
-      item->data = SPA_MEMBER (impl->buffer_data, areas[1].offset, void);
-      item->item_size = areas[0].len + 1 + size;
-    }
-    memcpy (item->data, data, size);
-
-    spa_ringbuffer_write_advance (&impl->buffer, item->item_size);
-
-    if (write (impl->wakeup.fd, &u, sizeof(uint64_t)) != sizeof(uint64_t))
-      pinos_log_warn ("data-loop %p: failed to write fd", strerror (errno));
-
-    if (seq != SPA_ID_INVALID)
-      res = SPA_RESULT_RETURN_ASYNC (seq);
-    else
-      res = SPA_RESULT_OK;
-
-  }
-  return res;
-}
-
 static bool
 process_work_queue (PinosMainLoop *this)
 {
@@ -234,19 +144,13 @@ process_work_queue (PinosMainLoop *this)
   impl->work_id = 0;
 
   spa_list_for_each_safe (item, tmp, &impl->work_list, link) {
-    if (item->sync) {
-      if (&item->link == impl->work_list.next) {
-        pinos_log_debug ("main-loop %p: found sync item %p", this, item->obj);
-      } else {
-        continue;
-      }
-    } else if (item->seq != SPA_ID_INVALID)
+    if (item->seq != SPA_ID_INVALID)
       continue;
 
     spa_list_remove (&item->link);
 
     if (item->func) {
-      pinos_log_debug ("main-loop %p: process work item %p %d %d", this, item->obj, item->sync, item->seq);
+      pinos_log_debug ("main-loop %p: process work item %p %d", this, item->obj, item->seq);
       item->func (item->obj, item->data, item->res, item->id);
     }
     spa_list_insert (impl->free_list.prev, &item->link);
@@ -259,8 +163,7 @@ do_add_work (PinosMainLoop  *loop,
              void           *obj,
              SpaResult       res,
              PinosDeferFunc  func,
-             void           *data,
-             bool            sync)
+             void           *data)
 {
   PinosMainLoopImpl *impl = SPA_CONTAINER_OF (loop, PinosMainLoopImpl, this);
   WorkItem *item;
@@ -276,7 +179,6 @@ do_add_work (PinosMainLoop  *loop,
   item->obj = obj;
   item->func = func;
   item->data = data;
-  item->sync = sync;
 
   if (SPA_RESULT_IS_ASYNC (res)) {
     item->seq = SPA_RESULT_ASYNC_SEQ (res);
@@ -286,7 +188,7 @@ do_add_work (PinosMainLoop  *loop,
     item->seq = SPA_ID_INVALID;
     item->res = res;
     have_work = TRUE;
-    pinos_log_debug ("main-loop %p: defer object %p %d", loop, obj, sync);
+    pinos_log_debug ("main-loop %p: defer object %p", loop, obj);
   }
   spa_list_insert (impl->work_list.prev, &item->link);
 
@@ -303,7 +205,7 @@ main_loop_defer (PinosMainLoop  *loop,
                  PinosDeferFunc  func,
                  void           *data)
 {
-  return do_add_work (loop, obj, res, func, data, false);
+  return do_add_work (loop, obj, res, func, data);
 }
 
 static void
@@ -354,15 +256,6 @@ main_loop_defer_complete (PinosMainLoop  *loop,
   return have_work;
 }
 
-static uint32_t
-main_loop_sync (PinosMainLoop  *loop,
-                void           *obj,
-                PinosDeferFunc  func,
-                void           *data)
-{
-  return do_add_work (loop, obj, SPA_RESULT_OK, func, data, true);
-}
-
 static void
 main_loop_quit (PinosMainLoop *loop)
 {
@@ -376,7 +269,9 @@ main_loop_run (PinosMainLoop *loop)
 {
   PinosMainLoopImpl *impl = SPA_CONTAINER_OF (loop, PinosMainLoopImpl, this);
   pinos_log_debug ("main-loop %p: run", impl);
+  pinos_loop_enter (loop->loop);
   g_main_loop_run (impl->main_loop);
+  pinos_loop_leave (loop->loop);
 }
 
 /**
@@ -394,35 +289,24 @@ pinos_main_loop_new (void)
 
   impl = calloc (1, sizeof (PinosMainLoopImpl));
   pinos_log_debug ("main-loop %p: new", impl);
+  this = &impl->this;
+
+  this->loop = pinos_loop_new ();
+  this->loop->loop->add_source = do_add_source;
+  this->loop->loop->update_source = do_update_source;
+  this->loop->loop->remove_source = do_remove_source;
 
   impl->context = g_main_context_default ();
-  this = &impl->this;
+  impl->main_loop = g_main_loop_new (impl->context, false);
+
   this->run = main_loop_run;
   this->quit = main_loop_quit;
   this->defer = main_loop_defer;
   this->defer_cancel = main_loop_defer_cancel;
   this->defer_complete = main_loop_defer_complete;
-  this->sync = main_loop_sync;
-
-  impl->loop.size = sizeof (SpaLoop);
-  impl->loop.add_source = do_add_source;
-  impl->loop.update_source = do_update_source;
-  impl->loop.remove_source = do_remove_source;
-  impl->loop.invoke = do_invoke;
-  this->loop = &impl->loop;
 
   spa_list_init (&impl->work_list);
   spa_list_init (&impl->free_list);
-  spa_ringbuffer_init (&impl->buffer, DATAS_SIZE);
-
-  impl->main_loop = g_main_loop_new (impl->context, false);
-
-  impl->wakeup.func = main_loop_dispatch;
-  impl->wakeup.data = impl;
-  impl->wakeup.fd = eventfd (0, 0);
-  impl->wakeup.mask = SPA_IO_IN | SPA_IO_ERR;
-  impl->wakeup.rmask = 0;
-  do_add_source (&impl->loop, &impl->wakeup);
 
   return this;
 }
@@ -437,7 +321,7 @@ pinos_main_loop_destroy (PinosMainLoop *loop)
 
   g_main_loop_unref (impl->main_loop);
 
-  close (impl->wakeup.fd);
+  pinos_loop_destroy (loop->loop);
 
   spa_list_for_each_safe (item, tmp, &impl->free_list, link)
     free (item);
diff --git a/pinos/server/main-loop.h b/pinos/server/main-loop.h
index a0df7188..f73f6633 100644
--- a/pinos/server/main-loop.h
+++ b/pinos/server/main-loop.h
@@ -25,6 +25,7 @@ extern "C" {
 #endif
 
 #include <spa/include/spa/loop.h>
+#include <pinos/client/loop.h>
 
 typedef struct _PinosMainLoop PinosMainLoop;
 
@@ -39,7 +40,7 @@ typedef void (*PinosDeferFunc) (void      *obj,
  * Pinos main-loop interface.
  */
 struct _PinosMainLoop {
-  SpaLoop *loop;
+  PinosLoop    *loop;
 
   void         (*run)             (PinosMainLoop  *loop);
   void         (*quit)            (PinosMainLoop  *loop);
@@ -56,10 +57,6 @@ struct _PinosMainLoop {
                                    void           *obj,
                                    uint32_t        seq,
                                    SpaResult       res);
-  uint32_t     (*sync)            (PinosMainLoop  *loop,
-                                   void           *obj,
-                                   PinosDeferFunc  func,
-                                   void           *data);
 };
 
 PinosMainLoop *     pinos_main_loop_new                     (void);
diff --git a/pinos/server/node.c b/pinos/server/node.c
index 56c4ef2c..d6d2c29f 100644
--- a/pinos/server/node.c
+++ b/pinos/server/node.c
@@ -311,12 +311,12 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
           continue;
 
         link->rt.in_ready++;
-        spa_loop_invoke (link->rt.input->node->data_loop->loop->loop,
-                         do_read_link,
-                         SPA_ID_INVALID,
-                         sizeof (PinosLink *),
-                         &link,
-                         link->rt.input->node);
+        pinos_loop_invoke (link->rt.input->node->data_loop->loop,
+                           do_read_link,
+                           SPA_ID_INVALID,
+                           sizeof (PinosLink *),
+                           &link,
+                           link->rt.input->node);
       }
       break;
     }
@@ -344,12 +344,12 @@ on_node_event (SpaNode *node, SpaNodeEvent *event, void *user_data)
           link->queue[offset] = po->buffer_id;
           spa_ringbuffer_write_advance (&link->ringbuffer, 1);
 
-          spa_loop_invoke (link->rt.input->node->data_loop->loop->loop,
-                           do_read_link,
-                           SPA_ID_INVALID,
-                           sizeof (PinosLink *),
-                           &link,
-                           link->rt.input->node);
+          pinos_loop_invoke (link->rt.input->node->data_loop->loop,
+                             do_read_link,
+                             SPA_ID_INVALID,
+                             sizeof (PinosLink *),
+                             &link,
+                             link->rt.input->node);
           pushed = true;
         }
       }
@@ -526,12 +526,12 @@ do_node_remove (SpaLoop        *loop,
     }
   }
 
-  res = spa_loop_invoke (this->core->main_loop->loop,
-                         do_node_remove_done,
-                         seq,
-                         0,
-                         NULL,
-                         this);
+  res = pinos_loop_invoke (this->core->main_loop->loop,
+                           do_node_remove_done,
+                           seq,
+                           0,
+                           NULL,
+                           this);
 
   return res;
 }
@@ -555,12 +555,12 @@ pinos_node_destroy (PinosNode * this)
   spa_list_remove (&this->link);
   pinos_global_destroy (this->global);
 
-  res = spa_loop_invoke (this->data_loop->loop->loop,
-                         do_node_remove,
-                         impl->seq++,
-                         0,
-                         NULL,
-                         this);
+  res = pinos_loop_invoke (this->data_loop->loop,
+                           do_node_remove,
+                           impl->seq++,
+                           0,
+                           NULL,
+                           this);
   return res;
 }
 
diff --git a/pinos/server/port.c b/pinos/server/port.c
index 8700f077..1b20bde3 100644
--- a/pinos/server/port.c
+++ b/pinos/server/port.c
@@ -169,18 +169,18 @@ pinos_port_link (PinosPort       *output_port,
     output_node->n_used_output_links++;
     input_node->n_used_input_links++;
 
-    spa_loop_invoke (output_node->data_loop->loop->loop,
-                     do_add_link,
-                     SPA_ID_INVALID,
-                     sizeof (PinosLink *),
-                     &link,
-                     output_port);
-    spa_loop_invoke (input_node->data_loop->loop->loop,
-                     do_add_link,
-                     SPA_ID_INVALID,
-                     sizeof (PinosLink *),
-                     &link,
-                     input_port);
+    pinos_loop_invoke (output_node->data_loop->loop,
+                       do_add_link,
+                       SPA_ID_INVALID,
+                       sizeof (PinosLink *),
+                       &link,
+                       output_port);
+    pinos_loop_invoke (input_node->data_loop->loop,
+                       do_add_link,
+                       SPA_ID_INVALID,
+                       sizeof (PinosLink *),
+                       &link,
+                       input_port);
   }
   return link;
 
@@ -284,12 +284,12 @@ do_remove_link (SpaLoop        *loop,
     pinos_port_pause (port);
 #endif
 
-  res = spa_loop_invoke (this->core->main_loop->loop,
-                         do_remove_link_done,
-                         seq,
-                         sizeof (PinosLink *),
-                         &link,
-                         port);
+  res = pinos_loop_invoke (this->core->main_loop->loop,
+                           do_remove_link_done,
+                           seq,
+                           sizeof (PinosLink *),
+                           &link,
+                           port);
   return res;
 }
 
@@ -301,12 +301,12 @@ pinos_port_unlink (PinosPort *port, PinosLink *link)
 
   pinos_log_debug ("port %p: start unlink %p", port, link);
 
-  res = spa_loop_invoke (port->node->data_loop->loop->loop,
-                         do_remove_link,
-                         impl->seq++,
-                         sizeof (PinosLink *),
-                         &link,
-                         port);
+  res = pinos_loop_invoke (port->node->data_loop->loop,
+                           do_remove_link,
+                           impl->seq++,
+                           sizeof (PinosLink *),
+                           &link,
+                           port);
   return res;
 }
 
@@ -352,11 +352,11 @@ do_clear_buffers (SpaLoop        *loop,
 
   pinos_port_pause (port);
 
-  res = spa_loop_invoke (node->core->main_loop->loop,
-                         do_clear_buffers_done,
-                         seq,
-                         0, NULL,
-                         port);
+  res = pinos_loop_invoke (node->core->main_loop->loop,
+                           do_clear_buffers_done,
+                           seq,
+                           0, NULL,
+                           port);
   return res;
 }
 
@@ -367,10 +367,10 @@ pinos_port_clear_buffers (PinosPort *port)
   PinosPortImpl *impl = SPA_CONTAINER_OF (port, PinosPortImpl, this);
 
   pinos_log_debug ("port %p: clear buffers", port);
-  res = spa_loop_invoke (port->node->data_loop->loop->loop,
-                         do_clear_buffers,
-                         impl->seq++,
-                         0, NULL,
-                         port);
+  res = pinos_loop_invoke (port->node->data_loop->loop,
+                           do_clear_buffers,
+                           impl->seq++,
+                           0, NULL,
+                           port);
   return res;
 }