add client-node again

Add a client node again that uses the channel protocol but looks like
a node in the graph.

pinos/Makefile.am

@@ -207,6 +207,7 @@ lib_LTLIBRARIES += libpinoscore-@PINOS_MAJORMINOR@.la
 libpinoscore_@PINOS_MAJORMINOR@_la_SOURCES = \
 		server/channel.c server/channel.h \
 		server/client.c server/client.h \
+		server/client-node.c server/client-node.h \
 		server/daemon.c server/daemon.h \
 		server/link.c server/link.h \
 		server/node.c server/node.h \

pinos/client/buffer.h

@@ -160,6 +160,7 @@ gint               pinos_buffer_builder_add_fd     (PinosBufferBuilder *builder,
 /* add-mem packets */
 /**
  * PinosPacketAddMem:
+ * @port: the port number
  * @id: the unique id of this memory block
  * @type: the memory block type
  * @fd_index: the index of the fd with the data
@@ -169,6 +170,7 @@ gint               pinos_buffer_builder_add_fd     (PinosBufferBuilder *builder,
  * A Packet that contains a memory block used for data transfer.
  */
 typedef struct {
+  guint32 port;
   guint32 id;
   guint32 type;
   gint32 fd_index;
@@ -184,11 +186,13 @@ gboolean           pinos_buffer_builder_add_add_mem     (PinosBufferBuilder   *b
 /* remove-mem packets */
 /**
  * PinosPacketRemoveMem:
+ * @port: the port number
  * @id: the unique id of the memory block
  *
  * Remove a memory block.
  */
 typedef struct {
+  guint32 port;
   guint32 id;
 } PinosPacketRemoveMem;
 
@@ -208,6 +212,7 @@ gboolean           pinos_buffer_builder_add_remove_mem  (PinosBufferBuilder   *b
  * A Packet that contains the header.
  */
 typedef struct {
+  guint32 port;
   guint32 flags;
   guint32 seq;
   gint64 pts;
@@ -231,6 +236,7 @@ gboolean           pinos_buffer_builder_add_header     (PinosBufferBuilder *buil
  * @size.
  */
 typedef struct {
+  guint32 port;
   guint32 id;
   guint64 offset;
   guint64 size;
@@ -249,6 +255,7 @@ gboolean           pinos_buffer_builder_add_process_mem (PinosBufferBuilder    *
  * Release the payload with @id
  */
 typedef struct {
+  guint32 port;
   guint32 id;
   guint64 offset;
   guint64 size;
@@ -269,6 +276,7 @@ gboolean           pinos_buffer_builder_add_reuse_mem (PinosBufferBuilder   *bui
  * A new format.
  */
 typedef struct {
+  guint32 port;
   guint8 id;
   const gchar *format;
 } PinosPacketFormatChange;
@@ -288,6 +296,7 @@ gboolean           pinos_buffer_builder_add_format_change  (PinosBufferBuilder
  * A new property change.
  */
 typedef struct {
+  guint32 port;
   const gchar *key;
   const gchar *value;
 } PinosPacketPropertyChange;
@@ -308,6 +317,7 @@ gboolean           pinos_buffer_builder_add_property_change (PinosBufferBuilder
  * A refresh request packet. This packet is sent to trigger a new keyframe.
  */
 typedef struct {
+  guint32 port;
   guint32 last_id;
   guint32 request_type;
   gint64 pts;

pinos/client/stream.c

@@ -53,7 +53,7 @@ struct _PinosStreamPrivate
 
   GBytes *format;
 
-  GDBusProxy *channel;
+  GDBusProxy *node;
   gboolean disconnecting;
 
   PinosStreamMode mode;
@@ -215,7 +215,7 @@ subscription_cb (PinosSubscribe         *subscribe,
   switch (flags) {
     case PINOS_SUBSCRIPTION_FLAG_CHANNEL:
       if (event == PINOS_SUBSCRIPTION_EVENT_REMOVE) {
-        if (object == priv->channel && !priv->disconnecting) {
+        if (object == priv->node && !priv->disconnecting) {
           stream_set_state (stream,
                             PINOS_STREAM_STATE_ERROR,
                             g_error_new_literal (G_IO_ERROR,
@@ -252,7 +252,7 @@ pinos_stream_finalize (GObject * object)
 
   g_debug ("free stream %p", stream);
 
-  g_clear_object (&priv->channel);
+  g_clear_object (&priv->node);
 
   if (priv->possible_formats)
     g_bytes_unref (priv->possible_formats);
@@ -488,71 +488,6 @@ pinos_stream_get_error (PinosStream *stream)
 }
 
 
-static void
-on_channel_proxy (GObject      *source_object,
-                  GAsyncResult *res,
-                  gpointer      user_data)
-{
-  PinosStream *stream = user_data;
-  PinosStreamPrivate *priv = stream->priv;
-  PinosContext *context = priv->context;
-  GVariant *v;
-  gchar *str;
-  GError *error = NULL;
-
-  priv->channel = pinos_subscribe_get_proxy_finish (context->priv->subscribe,
-                                                    res,
-                                                    &error);
-  if (priv->channel == NULL)
-    goto channel_failed;
-
-  /* get the port we are connected to */
-  v = g_dbus_proxy_get_cached_property (priv->channel, "Port");
-  if (v) {
-    gsize len;
-    str = g_variant_dup_string (v, &len);
-    g_variant_unref (v);
-
-    g_free (priv->path);
-    priv->path = str;
-  }
-
-  v = g_dbus_proxy_get_cached_property (priv->channel, "PossibleFormats");
-  if (v) {
-    gsize len;
-    str = g_variant_dup_string (v, &len);
-    g_variant_unref (v);
-
-    if (priv->possible_formats)
-      g_bytes_unref (priv->possible_formats);
-    priv->possible_formats = g_bytes_new_take (str, len + 1);
-
-    g_object_notify (G_OBJECT (stream), "possible-formats");
-  }
-  v = g_dbus_proxy_get_cached_property (priv->channel, "Properties");
-  if (v) {
-    if (priv->properties)
-      pinos_properties_free (priv->properties);
-    priv->properties = pinos_properties_from_variant (v);
-    g_variant_unref (v);
-
-    g_object_notify (G_OBJECT (stream), "properties");
-  }
-
-  stream_set_state (stream, PINOS_STREAM_STATE_READY, NULL);
-  g_object_unref (stream);
-
-  return;
-
-channel_failed:
-  {
-    g_warning ("failed to get channel proxy: %s", error->message);
-    stream_set_state (stream, PINOS_STREAM_STATE_ERROR, error);
-    g_object_unref (stream);
-    return;
-  }
-}
-
 static gboolean
 parse_buffer (PinosStream *stream,
               PinosBuffer *pbuf)
@@ -745,16 +680,46 @@ unhandle_socket (PinosStream *stream)
 
 
 static void
-on_channel_created (GObject      *source_object,
+on_node_proxy (GObject      *source_object,
-                    GAsyncResult *res,
+               GAsyncResult *res,
-                    gpointer      user_data)
+               gpointer      user_data)
+{
+  PinosStream *stream = user_data;
+  PinosStreamPrivate *priv = stream->priv;
+  PinosContext *context = priv->context;
+  GError *error = NULL;
+
+  priv->node = pinos_subscribe_get_proxy_finish (context->priv->subscribe,
+                                                 res,
+                                                 &error);
+  if (priv->node == NULL)
+    goto node_failed;
+
+  stream_set_state (stream, PINOS_STREAM_STATE_READY, NULL);
+  g_object_unref (stream);
+
+  return;
+
+node_failed:
+  {
+    g_warning ("failed to get node proxy: %s", error->message);
+    stream_set_state (stream, PINOS_STREAM_STATE_ERROR, error);
+    g_object_unref (stream);
+    return;
+  }
+}
+
+static void
+on_node_created (GObject      *source_object,
+                 GAsyncResult *res,
+                 gpointer      user_data)
 {
   PinosStream *stream = user_data;
   PinosStreamPrivate *priv = stream->priv;
   PinosContext *context = priv->context;
   GVariant *ret;
   GError *error = NULL;
-  const gchar *channel_path;
+  const gchar *node_path;
   GUnixFDList *fd_list;
   gint fd_idx, fd;
 
@@ -766,7 +731,7 @@ on_channel_created (GObject      *source_object,
   if (ret == NULL)
     goto create_failed;
 
-  g_variant_get (ret, "(&oh)", &channel_path, &fd_idx);
+  g_variant_get (ret, "(&oh)", &node_path, &fd_idx);
   g_variant_unref (ret);
 
   if ((fd = g_unix_fd_list_get (fd_list, fd_idx, &error)) < 0)
@@ -778,10 +743,10 @@ on_channel_created (GObject      *source_object,
 
   pinos_subscribe_get_proxy (context->priv->subscribe,
                              PINOS_DBUS_SERVICE,
-                             channel_path,
+                             node_path,
-                             "org.pinos.Channel1",
+                             "org.pinos.Node1",
                              NULL,
-                             on_channel_proxy,
+                             on_node_proxy,
                              stream);
 
   return;
@@ -812,18 +777,29 @@ do_connect (PinosStream *stream)
 {
   PinosStreamPrivate *priv = stream->priv;
   PinosContext *context = priv->context;
+  GVariantBuilder b;
+  GVariant *ports;
+
+  g_variant_builder_init (&b, G_VARIANT_TYPE ("a(uusa{sv}s)"));
+  g_variant_builder_open (&b, G_VARIANT_TYPE ("(uusa{sv}s)"));
+  g_variant_builder_add (&b, "u", priv->direction);
+  g_variant_builder_add (&b, "u", 0);
+  g_variant_builder_add (&b, "s", g_bytes_get_data (priv->possible_formats, NULL));
+  g_variant_builder_add_value (&b, pinos_properties_to_variant (priv->properties));
+  g_variant_builder_add (&b, "s", priv->path);
+  g_variant_builder_close (&b);
+  ports = g_variant_builder_end (&b);
 
   g_dbus_proxy_call (context->priv->daemon,
-                     "CreateChannel",
+                     "CreateClientNode",
-                     g_variant_new ("(sus@a{sv})",
+                     g_variant_new ("(s@a{sv}@a(uusa{sv}s))",
-                       (priv->path ? priv->path : ""),
+                       "client-node",
-                       priv->direction,
+                       pinos_properties_to_variant (priv->properties),
-                       g_bytes_get_data (priv->possible_formats, NULL),
+                       ports),
-                       pinos_properties_to_variant (priv->properties)),
                      G_DBUS_CALL_FLAGS_NONE,
                      -1,
                      NULL, /* GCancellable *cancellable */
-                     on_channel_created,
+                     on_node_created,
                      stream);
   return FALSE;
 }
@@ -981,7 +957,7 @@ pinos_stream_stop (PinosStream *stream)
 }
 
 static void
-on_channel_removed (GObject      *source_object,
+on_node_removed (GObject      *source_object,
                           GAsyncResult *res,
                           gpointer      user_data)
 {
@@ -990,10 +966,10 @@ on_channel_removed (GObject      *source_object,
   GVariant *ret;
   GError *error = NULL;
 
-  g_assert (priv->channel == G_DBUS_PROXY (source_object));
+  g_assert (priv->node == G_DBUS_PROXY (source_object));
 
   priv->disconnecting = FALSE;
-  g_clear_object (&priv->channel);
+  g_clear_object (&priv->node);
 
   ret = g_dbus_proxy_call_finish (G_DBUS_PROXY (source_object), res, &error);
   if (ret == NULL)
@@ -1021,13 +997,13 @@ do_disconnect (PinosStream *stream)
 {
   PinosStreamPrivate *priv = stream->priv;
 
-  g_dbus_proxy_call (priv->channel,
+  g_dbus_proxy_call (priv->node,
                      "Remove",
                      g_variant_new ("()"),
                      G_DBUS_CALL_FLAGS_NONE,
                      -1,
                      NULL, /* GCancellable *cancellable */
-                     on_channel_removed,
+                     on_node_removed,
                      stream);
 
   return FALSE;
@@ -1050,7 +1026,7 @@ pinos_stream_disconnect (PinosStream *stream)
   g_return_val_if_fail (PINOS_IS_STREAM (stream), FALSE);
   priv = stream->priv;
   g_return_val_if_fail (priv->state >= PINOS_STREAM_STATE_READY, FALSE);
-  g_return_val_if_fail (priv->channel != NULL, FALSE);
+  g_return_val_if_fail (priv->node != NULL, FALSE);
   context = priv->context;
   g_return_val_if_fail (pinos_context_get_state (context) >= PINOS_CONTEXT_STATE_CONNECTED, FALSE);
   g_return_val_if_fail (!priv->disconnecting, FALSE);

pinos/dbus/org.pinos.xml

@@ -59,6 +59,14 @@
       <arg type='o' name='node' direction='out'/>
     </method>
 
+    <method name='CreateClientNode'>
+      <arg type='s' name='name' direction='in' />
+      <arg type='a{sv}' name='properties' direction='in'/>
+      <arg type='a(uusa{sv}s)' name='ports' direction='in'/>
+      <arg type='o' name='node' direction='out'/>
+      <arg type='h' name='fd' direction='out'/>
+    </method>
+
     <method name='LinkNodes'>
       <arg type='o' name='src_node' direction='in' />
       <arg type='o' name='dest_node' direction='in' />
@@ -139,6 +147,16 @@
 
     <property name='Ports' type='a(uu)' access='read' />
 
+    <method name='AddPort'>
+      <arg type='u' name='direction' direction='in'/>
+      <arg type='u' name='id' direction='in'/>
+    </method>
+
+    <method name='RemovePort'>
+      <arg type='u' name='id' direction='in'/>
+    </method>
+
+
     <!-- state: state of the node
           -1 = the node is in error
           0 = the node is suspended, this means that the node is not

pinos/modules/gst/gst-sink.c

@@ -31,6 +31,7 @@
 typedef struct {
   PinosGstSink *sink;
 
+  guint id;
   PinosPort *port;
 
   GstElement *src;
@@ -384,6 +385,7 @@ free_sink_port_data (SinkPortData *data)
 static PinosPort *
 add_port (PinosNode       *node,
           PinosDirection   direction,
+          guint            id,
           GError         **error)
 {
   PinosGstSink *sink = PINOS_GST_SINK (node);
@@ -392,9 +394,9 @@ add_port (PinosNode       *node,
 
   data = g_slice_new0 (SinkPortData);
   data->sink = sink;
-
+  data->id = id;
   data->port = PINOS_NODE_CLASS (pinos_gst_sink_parent_class)
-                ->add_port (node, direction, error);
+                ->add_port (node, direction, id, error);
 
   g_debug ("connecting signals");
   g_signal_connect (data->port, "linked", (GCallback) on_linked, data);
@@ -418,9 +420,9 @@ add_port (PinosNode       *node,
   return data->port;
 }
 
-static void
+static gboolean
 remove_port (PinosNode       *node,
-             PinosPort       *port)
+             guint            id)
 {
   PinosGstSink *sink = PINOS_GST_SINK (node);
   PinosGstSinkPrivate *priv = sink->priv;
@@ -429,7 +431,7 @@ remove_port (PinosNode       *node,
   for (walk = priv->ports; walk; walk = g_list_next (walk)) {
     SinkPortData *data = walk->data;
 
-    if (data->port == PINOS_PORT_CAST (port)) {
+    if (data->id == id) {
       free_sink_port_data (data);
       priv->ports = g_list_delete_link (priv->ports, walk);
       break;
@@ -437,6 +439,8 @@ remove_port (PinosNode       *node,
   }
   if (priv->ports == NULL)
     pinos_node_report_idle (node);
+
+  return TRUE;
 }
 
 static void

pinos/modules/gst/gst-source.c

@@ -31,6 +31,7 @@
 typedef struct {
   PinosGstSource *source;
 
+  guint id;
   PinosPort *port;
 
   GstElement *convert;
@@ -430,9 +431,9 @@ free_source_port_data (SourcePortData *data)
   g_slice_free (SourcePortData, data);
 }
 
-static void
+static gboolean
 remove_port (PinosNode       *node,
-             PinosPort       *port)
+             guint            id)
 {
   PinosGstSource *source = PINOS_GST_SOURCE (node);
   PinosGstSourcePrivate *priv = source->priv;
@@ -441,7 +442,7 @@ remove_port (PinosNode       *node,
   for (walk = priv->ports; walk; walk = g_list_next (walk)) {
     SourcePortData *data = walk->data;
 
-    if (data->port == PINOS_PORT_CAST (port)) {
+    if (data->id == id) {
       free_source_port_data (data);
       priv->ports = g_list_delete_link (priv->ports, walk);
       break;
@@ -449,6 +450,8 @@ remove_port (PinosNode       *node,
   }
   if (priv->ports == NULL)
     pinos_node_report_idle (node);
+
+  return TRUE;
 }
 
 static void
@@ -550,6 +553,7 @@ create_best_element (GstCaps *caps)
 static PinosPort *
 add_port (PinosNode       *node,
           PinosDirection   direction,
+          guint            id,
           GError         **error)
 {
   PinosGstSource *source = PINOS_GST_SOURCE (node);
@@ -573,9 +577,9 @@ add_port (PinosNode       *node,
 
   data = g_slice_new0 (SourcePortData);
   data->source = source;
-
+  data->id = id;
   data->port = PINOS_NODE_CLASS (pinos_gst_source_parent_class)
-                ->add_port (node, direction, error);
+                ->add_port (node, direction, id, error);
 
   g_debug ("connecting signals");
   g_signal_connect (data->port, "activate", (GCallback) on_activate, data);

pinos/server/channel.c

@@ -84,7 +84,6 @@ enum
   PROP_POSSIBLE_FORMATS,
   PROP_PROPERTIES,
   PROP_FORMAT,
-  PROP_SOCKET,
   PROP_STATE,
 };
 
@@ -138,10 +137,6 @@ pinos_channel_get_property (GObject    *_object,
       g_value_set_boxed (value, priv->format);
       break;
 
-    case PROP_SOCKET:
-      g_value_set_object (value, priv->socket);
-      break;
-
     case PROP_STATE:
       g_value_set_uint (value, priv->state);
       break;
@@ -530,7 +525,6 @@ pinos_channel_dispose (GObject * object)
   pinos_port_deactivate (priv->port);
   clear_formats (channel);
   unhandle_socket (channel);
-  g_clear_object (&priv->socket);
   channel_unregister_object (channel);
 
   G_OBJECT_CLASS (pinos_channel_parent_class)->dispose (object);
@@ -660,15 +654,6 @@ pinos_channel_class_init (PinosChannelClass * klass)
                                                        G_PARAM_READWRITE |
                                                        G_PARAM_STATIC_STRINGS));
 
-  g_object_class_install_property (gobject_class,
-                                   PROP_SOCKET,
-                                   g_param_spec_object ("socket",
-                                                        "Socket",
-                                                        "The socket with data",
-                                                        G_TYPE_SOCKET,
-                                                        G_PARAM_READABLE |
-                                                        G_PARAM_STATIC_STRINGS));
-
   signals[SIGNAL_REMOVE] = g_signal_new ("remove",
                                          G_TYPE_FROM_CLASS (klass),
                                          G_SIGNAL_RUN_LAST,

pinos/server/client-node.c

@@ -0,0 +1,415 @@
+/* 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 <sys/socket.h>
+#include <errno.h>
+
+#include <gio/gunixfdlist.h>
+
+#include "pinos/client/pinos.h"
+#include "pinos/client/enumtypes.h"
+#include "pinos/client/private.h"
+
+#include "pinos/server/daemon.h"
+#include "pinos/server/client-node.h"
+#include "pinos/server/utils.h"
+
+#include "pinos/dbus/org-pinos.h"
+
+
+#define MAX_BUFFER_SIZE 1024
+#define MAX_FDS         16
+
+struct _PinosClientNodePrivate
+{
+  int fd;
+  GSource *socket_source;
+  GSocket *sockets[2];
+
+  PinosBuffer recv_buffer;
+
+  guint8 recv_data[MAX_BUFFER_SIZE];
+  int recv_fds[MAX_FDS];
+
+  guint8 send_data[MAX_BUFFER_SIZE];
+  int send_fds[MAX_FDS];
+};
+
+#define PINOS_CLIENT_NODE_GET_PRIVATE(obj)  \
+   (G_TYPE_INSTANCE_GET_PRIVATE ((obj), PINOS_TYPE_CLIENT_NODE, PinosClientNodePrivate))
+
+G_DEFINE_TYPE (PinosClientNode, pinos_client_node, PINOS_TYPE_NODE);
+
+enum
+{
+  PROP_0,
+};
+
+enum
+{
+  SIGNAL_NONE,
+  LAST_SIGNAL
+};
+
+static guint signals[LAST_SIGNAL] = { 0 };
+
+static void
+pinos_client_node_get_property (GObject    *_object,
+                                  guint       prop_id,
+                                  GValue     *value,
+                                  GParamSpec *pspec)
+{
+  PinosClientNode *node = PINOS_CLIENT_NODE (_object);
+
+  switch (prop_id) {
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (node, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+pinos_client_node_set_property (GObject      *_object,
+                            guint         prop_id,
+                            const GValue *value,
+                            GParamSpec   *pspec)
+{
+  PinosClientNode *node = PINOS_CLIENT_NODE (_object);
+
+  switch (prop_id) {
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (node, prop_id, pspec);
+      break;
+  }
+}
+
+static gboolean
+parse_buffer (PinosClientNode *cnode,
+              PinosBuffer *pbuf)
+{
+  PinosNode *node = PINOS_NODE (cnode);
+  PinosBufferIter it;
+  PinosClientNodePrivate *priv = cnode->priv;
+  PinosPort *port;
+
+  pinos_buffer_iter_init (&it, pbuf);
+  while (pinos_buffer_iter_next (&it)) {
+    PinosPacketType type = pinos_buffer_iter_get_type (&it);
+
+    switch (type) {
+      case PINOS_PACKET_TYPE_FORMAT_CHANGE:
+      {
+        PinosPacketFormatChange p;
+        GBytes *format;
+
+        if (!pinos_buffer_iter_parse_format_change (&it, &p))
+          break;
+
+        if (!(port = pinos_node_find_port (node, p.port)))
+          break;
+
+        format = g_bytes_new_static (p.format, strlen (p.format) + 1);
+
+        g_object_set (port, "possible-formats", format, NULL);
+        g_object_set (port, "format", format, NULL);
+        g_debug ("client-node %p: format change %s", node, p.format);
+        break;
+      }
+      case PINOS_PACKET_TYPE_START:
+      {
+        GBytes *format;
+        PinosBufferBuilder builder;
+        PinosBuffer obuf;
+        guint8 buffer[1024];
+        GError *error = NULL;
+        GList *ports, *walk;
+
+        pinos_buffer_builder_init_into (&builder, buffer, 1024, NULL, 0);
+
+        ports = pinos_node_get_ports (node);
+        for (walk = ports; walk; walk = g_list_next (walk)) {
+          PinosPacketFormatChange fc;
+
+          port = walk->data;
+
+          pinos_port_activate (port);
+
+          g_object_get (port, "format", &format, "id", &fc.port, NULL);
+          if (format == NULL)
+            break;
+
+          fc.id = 0;
+          fc.format = g_bytes_get_data (format, NULL);
+
+          g_debug ("client-node %p: port %u we are now streaming in format \"%s\"",
+             node, fc.port, fc.format);
+
+          pinos_buffer_builder_add_format_change (&builder, &fc);
+        }
+        pinos_buffer_builder_add_empty (&builder, PINOS_PACKET_TYPE_STREAMING);
+        pinos_buffer_builder_end (&builder, &obuf);
+
+        if (!pinos_io_write_buffer (priv->fd, &obuf, &error)) {
+          g_warning ("client-node %p: error writing buffer: %s", node, error->message);
+          g_clear_error (&error);
+        }
+
+        break;
+      }
+      case PINOS_PACKET_TYPE_STOP:
+      {
+        break;
+      }
+      case PINOS_PACKET_TYPE_REUSE_MEM:
+      {
+        break;
+      }
+      default:
+        g_warning ("unhandled packet %d", type);
+        break;
+    }
+  }
+  pinos_buffer_iter_end (&it);
+
+  return TRUE;
+}
+
+static gboolean
+on_socket_condition (GSocket      *socket,
+                     GIOCondition  condition,
+                     gpointer      user_data)
+{
+  PinosClientNode *node = user_data;
+  PinosClientNodePrivate *priv = node->priv;
+
+  switch (condition) {
+    case G_IO_IN:
+    {
+      PinosBuffer *buffer = &priv->recv_buffer;
+      GError *error = NULL;
+
+      if (!pinos_io_read_buffer (priv->fd,
+                                 buffer,
+                                 priv->recv_data,
+                                 MAX_BUFFER_SIZE,
+                                 priv->recv_fds,
+                                 MAX_FDS,
+                                 &error)) {
+        g_warning ("client-node %p: failed to read buffer: %s", node, error->message);
+        g_clear_error (&error);
+        return TRUE;
+      }
+
+      parse_buffer (node, buffer);
+
+#if 0
+      if (!pinos_port_receive_buffer (priv->port, buffer, &error)) {
+        g_warning ("client-node %p: port %p failed to receive buffer: %s", node, priv->port, error->message);
+        g_clear_error (&error);
+      }
+#endif
+      g_assert (pinos_buffer_unref (buffer) == FALSE);
+      break;
+    }
+
+    case G_IO_OUT:
+      g_warning ("can do IO OUT\n");
+      break;
+
+    default:
+      break;
+  }
+  return TRUE;
+}
+
+static void
+handle_socket (PinosClientNode *node, GSocket *socket)
+{
+  PinosClientNodePrivate *priv = node->priv;
+  GMainContext *context = g_main_context_get_thread_default();
+
+  g_debug ("client-node %p: handle socket in context %p", node, context);
+  priv->fd = g_socket_get_fd (socket);
+  priv->socket_source = g_socket_create_source (socket, G_IO_IN, NULL);
+  g_source_set_callback (priv->socket_source, (GSourceFunc) on_socket_condition, node, NULL);
+  g_source_attach (priv->socket_source, context);
+}
+
+static void
+unhandle_socket (PinosClientNode *node)
+{
+  PinosClientNodePrivate *priv = node->priv;
+
+  g_debug ("client-node %p: unhandle socket", node);
+  if (priv->socket_source) {
+    g_source_destroy (priv->socket_source);
+    g_clear_pointer (&priv->socket_source, g_source_unref);
+    priv->fd = -1;
+  }
+}
+
+/**
+ * pinos_client_node_get_socket_pair:
+ * @node: a #PinosClientNode
+ * @error: a #GError
+ *
+ * Create or return a previously create socket pair for @node. The
+ * Socket for the other end is returned.
+ *
+ * Returns: a #GSocket that can be used to send/receive buffers to node.
+ */
+GSocket *
+pinos_client_node_get_socket_pair (PinosClientNode  *node,
+                               GError       **error)
+{
+  PinosClientNodePrivate *priv;
+
+  g_return_val_if_fail (PINOS_IS_CLIENT_NODE (node), FALSE);
+  priv = node->priv;
+
+  if (priv->sockets[1] == NULL) {
+    int fd[2];
+
+    if (socketpair (AF_UNIX, SOCK_STREAM, 0, fd) != 0)
+      goto no_sockets;
+
+    priv->sockets[0] = g_socket_new_from_fd (fd[0], error);
+    if (priv->sockets[0] == NULL)
+      goto create_failed;
+
+    priv->sockets[1] = g_socket_new_from_fd (fd[1], error);
+    if (priv->sockets[1] == NULL)
+      goto create_failed;
+
+    handle_socket (node, priv->sockets[0]);
+  }
+  return g_object_ref (priv->sockets[1]);
+
+  /* ERRORS */
+no_sockets:
+  {
+    g_set_error (error,
+                 G_IO_ERROR,
+                 g_io_error_from_errno (errno),
+                 "could not create socketpair: %s", strerror (errno));
+    return NULL;
+  }
+create_failed:
+  {
+    g_clear_object (&priv->sockets[0]);
+    g_clear_object (&priv->sockets[1]);
+    return NULL;
+  }
+}
+
+static gboolean
+on_received_buffer (PinosPort *port, PinosBuffer *buffer, GError **error, gpointer user_data)
+{
+  PinosClientNode *node = user_data;
+  PinosClientNodePrivate *priv = node->priv;
+
+  if (!pinos_io_write_buffer (priv->fd, buffer, error)) {
+    g_warning ("client-node %p: error writing buffer: %s", node, (*error)->message);
+    return FALSE;
+  }
+  return TRUE;
+}
+
+static PinosPort *
+add_port (PinosNode       *node,
+          PinosDirection   direction,
+          guint            id,
+          GError         **error)
+{
+  PinosPort *port;
+
+  port = PINOS_NODE_CLASS (pinos_client_node_parent_class)->add_port (node, direction, id, error);
+
+  if (port) {
+    pinos_port_set_received_buffer_cb (port, on_received_buffer, node, NULL);
+  }
+  return port;
+}
+
+static gboolean
+remove_port (PinosNode       *node,
+             guint            id)
+{
+  return PINOS_NODE_CLASS (pinos_client_node_parent_class)->remove_port (node, id);
+}
+
+static void
+pinos_client_node_dispose (GObject * object)
+{
+  PinosClientNode *node = PINOS_CLIENT_NODE (object);
+  PinosClientNodePrivate *priv = node->priv;
+
+  g_debug ("client-node %p: dispose", node);
+  unhandle_socket (node);
+
+  G_OBJECT_CLASS (pinos_client_node_parent_class)->dispose (object);
+}
+
+static void
+pinos_client_node_finalize (GObject * object)
+{
+  PinosClientNode *node = PINOS_CLIENT_NODE (object);
+  PinosClientNodePrivate *priv = node->priv;
+
+  g_debug ("client-node %p: finalize", node);
+
+  G_OBJECT_CLASS (pinos_client_node_parent_class)->finalize (object);
+}
+
+static void
+pinos_client_node_constructed (GObject * object)
+{
+  PinosClientNode *node = PINOS_CLIENT_NODE (object);
+
+  g_debug ("client-node %p: constructed", node);
+
+  G_OBJECT_CLASS (pinos_client_node_parent_class)->constructed (object);
+}
+
+static void
+pinos_client_node_class_init (PinosClientNodeClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+  PinosNodeClass *node_class = PINOS_NODE_CLASS (klass);
+
+  g_type_class_add_private (klass, sizeof (PinosClientNodePrivate));
+
+  gobject_class->constructed = pinos_client_node_constructed;
+  gobject_class->dispose = pinos_client_node_dispose;
+  gobject_class->finalize = pinos_client_node_finalize;
+  gobject_class->set_property = pinos_client_node_set_property;
+  gobject_class->get_property = pinos_client_node_get_property;
+
+  node_class->add_port = add_port;
+  node_class->remove_port = remove_port;
+}
+
+static void
+pinos_client_node_init (PinosClientNode * node)
+{
+  PinosClientNodePrivate *priv = node->priv = PINOS_CLIENT_NODE_GET_PRIVATE (node);
+
+  g_debug ("client-node %p: new", node);
+}

pinos/server/client-node.h

@@ -0,0 +1,68 @@
+/* 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.
+ */
+
+#ifndef __PINOS_CLIENT_NODE_H__
+#define __PINOS_CLIENT_NODE_H__
+
+#include <glib-object.h>
+
+G_BEGIN_DECLS
+
+#define PINOS_TYPE_CLIENT_NODE                 (pinos_client_node_get_type ())
+#define PINOS_IS_CLIENT_NODE(obj)              (G_TYPE_CHECK_INSTANCE_TYPE ((obj), PINOS_TYPE_CLIENT_NODE))
+#define PINOS_IS_CLIENT_NODE_CLASS(klass)      (G_TYPE_CHECK_CLASS_TYPE ((klass), PINOS_TYPE_CLIENT_NODE))
+#define PINOS_CLIENT_NODE_GET_CLASS(obj)       (G_TYPE_INSTANCE_GET_CLASS ((obj), PINOS_TYPE_CLIENT_NODE, PinosClientNodeClass))
+#define PINOS_CLIENT_NODE(obj)                 (G_TYPE_CHECK_INSTANCE_CAST ((obj), PINOS_TYPE_CLIENT_NODE, PinosClientNode))
+#define PINOS_CLIENT_NODE_CLASS(klass)         (G_TYPE_CHECK_CLASS_CAST ((klass), PINOS_TYPE_CLIENT_NODE, PinosClientNodeClass))
+#define PINOS_CLIENT_NODE_CAST(obj)            ((PinosClientNode*)(obj))
+#define PINOS_CLIENT_NODE_CLASS_CAST(klass)    ((PinosClientNodeClass*)(klass))
+
+typedef struct _PinosClientNode PinosClientNode;
+typedef struct _PinosClientNodeClass PinosClientNodeClass;
+typedef struct _PinosClientNodePrivate PinosClientNodePrivate;
+
+/**
+ * PinosClientNode:
+ *
+ * Pinos client node object class.
+ */
+struct _PinosClientNode {
+  PinosNode object;
+
+  PinosClientNodePrivate *priv;
+};
+
+/**
+ * PinosClientNodeClass:
+ *
+ * Pinos client node object class.
+ */
+struct _PinosClientNodeClass {
+  PinosNodeClass parent_class;
+};
+
+/* normal GObject stuff */
+GType              pinos_client_node_get_type             (void);
+
+GSocket *          pinos_client_node_get_socket_pair      (PinosClientNode  *node,
+                                                           GError          **error);
+
+G_END_DECLS
+
+#endif /* __PINOS_CLIENT_NODE_H__ */

pinos/server/daemon.c

@@ -28,8 +28,10 @@
 
 #include "pinos/server/daemon.h"
 #include "pinos/server/node.h"
+#include "pinos/server/client-node.h"
 #include "pinos/server/client.h"
 #include "pinos/server/channel.h"
+#include "pinos/server/link.h"
 
 #include "pinos/dbus/org-pinos.h"
 
@@ -291,6 +293,113 @@ no_node:
   }
 }
 
+static gboolean
+handle_create_client_node (PinosDaemon1           *interface,
+                           GDBusMethodInvocation  *invocation,
+                           const gchar            *arg_name,
+                           GVariant               *arg_properties,
+                           GVariant               *arg_ports,
+                           gpointer                user_data)
+{
+  PinosDaemon *daemon = user_data;
+  PinosNode *node;
+  PinosClient *client;
+  const gchar *sender, *object_path;
+  PinosProperties *props;
+  GError *error = NULL;
+  GUnixFDList *fdlist;
+  GSocket *socket;
+  gint fdidx;
+
+  sender = g_dbus_method_invocation_get_sender (invocation);
+
+  g_debug ("daemon %p: create client-node: %s", daemon, sender);
+  props = pinos_properties_from_variant (arg_properties);
+
+  node =  g_object_new (PINOS_TYPE_CLIENT_NODE,
+                        "daemon", daemon,
+                        "sender", sender,
+                        "name", arg_name,
+                        "properties", props,
+                        NULL);
+
+  client = sender_get_client (daemon, sender);
+
+  {
+    GVariantIter it;
+    guint direction;
+    guint id;
+    const gchar *format_str, *target_node;
+    GVariant *port_props;
+
+    g_variant_iter_init (&it, arg_ports);
+    while (g_variant_iter_loop (&it, "(uu&s@a{sv}&s)", &direction, &id, &format_str, &port_props, &target_node)) {
+      PinosPort *port, *target;
+      PinosProperties *pprops;
+      GBytes *formats;
+      PinosLink *link;
+
+      port = pinos_node_add_port (node, direction, id, &error);
+      if (port == NULL)
+        goto add_failed;
+
+      pprops = pinos_properties_from_variant (port_props);
+      formats = g_bytes_new (format_str, strlen (format_str) + 1);
+
+      g_object_set (port, "possible-formats", formats,
+                          "properties", port_props,
+                          NULL);
+
+      target = pinos_daemon_find_port (daemon,
+                                       pinos_direction_reverse (direction),
+                                       target_node,
+                                       pprops,
+                                       formats,
+                                       &error);
+      if (target == NULL)
+        continue;
+
+      link = pinos_link_new (daemon, port, target, NULL);
+      pinos_client_add_object (client, G_OBJECT (link));
+    }
+  }
+
+  socket = pinos_client_node_get_socket_pair (PINOS_CLIENT_NODE (node), &error);
+  if (socket == NULL)
+    goto no_socket;
+
+  pinos_client_add_object (client, G_OBJECT (node));
+
+  object_path = pinos_node_get_object_path (PINOS_NODE (node));
+  g_debug ("daemon %p: add client-node %p, %s", daemon, node, object_path);
+
+  fdlist = g_unix_fd_list_new ();
+  fdidx = g_unix_fd_list_append (fdlist, g_socket_get_fd (socket), &error);
+
+  g_dbus_method_invocation_return_value_with_unix_fd_list (invocation,
+           g_variant_new ("(oh)", object_path, fdidx), fdlist);
+  g_object_unref (fdlist);
+
+  return TRUE;
+
+no_socket:
+  {
+    g_debug ("daemon %p: could not create node %s", daemon, error->message);
+    goto exit_error;
+  }
+add_failed:
+  {
+    g_debug ("daemon %p: failed to add port %s", daemon, error->message);
+    goto exit_error;
+  }
+exit_error:
+  {
+    g_dbus_method_invocation_return_gerror (invocation, error);
+    g_clear_error (&error);
+    return TRUE;
+  }
+}
+
 static gboolean
 handle_link_nodes (PinosDaemon1           *interface,
                    GDBusMethodInvocation  *invocation,
@@ -544,11 +653,13 @@ pinos_daemon_find_port (PinosDaemon     *daemon,
 
   have_name = name ? strlen (name) > 0 : FALSE;
 
+  g_debug ("name %s, format %s", name, (gchar*)g_bytes_get_data (format_filter, NULL));
+
   for (nodes = priv->nodes; nodes; nodes = g_list_next (nodes)) {
     PinosNode *n = nodes->data;
     gboolean node_found = FALSE;
 
-    g_debug ("name %s, node path %s", name, pinos_node_get_object_path (n));
+    g_debug ("node path %s", pinos_node_get_object_path (n));
 
     /* we found the node */
     if (have_name && g_str_has_suffix (pinos_node_get_object_path (n), name)) {
@@ -576,7 +687,7 @@ pinos_daemon_find_port (PinosDaemon     *daemon,
     }
     if (best == NULL && node_found) {
       g_debug ("node %p: making port", n);
-      best = pinos_node_add_port (n, direction, NULL);
+      best = pinos_node_add_port (n, direction, 0, NULL);
       if (best != NULL) {
         created_port = TRUE;
         break;
@@ -730,6 +841,7 @@ pinos_daemon_init (PinosDaemon * daemon)
   g_debug ("daemon %p: new", daemon);
   priv->iface = pinos_daemon1_skeleton_new ();
   g_signal_connect (priv->iface, "handle-create-node", (GCallback) handle_create_node, daemon);
+  g_signal_connect (priv->iface, "handle-create-client-node", (GCallback) handle_create_client_node, daemon);
   g_signal_connect (priv->iface, "handle-create-channel", (GCallback) handle_create_channel, daemon);
   g_signal_connect (priv->iface, "handle-link-nodes", (GCallback) handle_link_nodes, daemon);
 

pinos/server/link.c

@@ -39,11 +39,10 @@ struct _PinosLinkPrivate
 
   gchar *object_path;
 
-  PinosPort *src;
+  PinosPort *output;
-  PinosPort *dest;
+  PinosPort *input;
   GBytes *possible_formats;
   GBytes *format;
-  PinosProperties *properties;
 };
 
 G_DEFINE_TYPE (PinosLink, pinos_link, G_TYPE_OBJECT);
@@ -52,22 +51,39 @@ enum
 {
   PROP_0,
   PROP_DAEMON,
+  PROP_OUTPUT,
+  PROP_INPUT,
   PROP_OBJECT_PATH,
   PROP_POSSIBLE_FORMATS,
   PROP_FORMAT,
-  PROP_PROPERTIES,
 };
 
 enum
 {
   SIGNAL_REMOVE,
-  SIGNAL_ACTIVATE,
-  SIGNAL_DEACTIVATE,
   LAST_SIGNAL
 };
 
 static guint signals[LAST_SIGNAL] = { 0 };
 
+static gboolean
+on_output_send (PinosPort *port, PinosBuffer *buffer, GError **error, gpointer user_data)
+{
+  PinosLink *link = user_data;
+  PinosLinkPrivate *priv = link->priv;
+
+  return pinos_port_receive_buffer (priv->input, buffer, error);
+}
+
+static gboolean
+on_input_send (PinosPort *port, PinosBuffer *buffer, GError **error, gpointer user_data)
+{
+  PinosLink *link = user_data;
+  PinosLinkPrivate *priv = link->priv;
+
+  return pinos_port_receive_buffer (priv->output, buffer, error);
+}
+
 static void
 pinos_link_get_property (GObject    *_object,
                            guint       prop_id,
@@ -82,6 +98,14 @@ pinos_link_get_property (GObject    *_object,
       g_value_set_object (value, priv->daemon);
       break;
 
+    case PROP_OUTPUT:
+      g_value_set_object (value, priv->output);
+      break;
+
+    case PROP_INPUT:
+      g_value_set_object (value, priv->input);
+      break;
+
     case PROP_OBJECT_PATH:
       g_value_set_string (value, priv->object_path);
       break;
@@ -94,10 +118,6 @@ pinos_link_get_property (GObject    *_object,
       g_value_set_boxed (value, priv->format);
       break;
 
-    case PROP_PROPERTIES:
-      g_value_set_boxed (value, priv->properties);
-      break;
-
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (link, prop_id, pspec);
       break;
@@ -118,6 +138,14 @@ pinos_link_set_property (GObject      *_object,
       priv->daemon = g_value_dup_object (value);
       break;
 
+    case PROP_OUTPUT:
+      priv->output = g_value_dup_object (value);
+      break;
+
+    case PROP_INPUT:
+      priv->input = g_value_dup_object (value);
+      break;
+
     case PROP_OBJECT_PATH:
       priv->object_path = g_value_dup_string (value);
       break;
@@ -134,12 +162,6 @@ pinos_link_set_property (GObject      *_object,
       priv->format = g_value_dup_boxed (value);
       break;
 
-    case PROP_PROPERTIES:
-      if (priv->properties)
-        pinos_properties_free (priv->properties);
-      priv->properties = g_value_dup_boxed (value);
-      break;
-
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (link, prop_id, pspec);
       break;
@@ -175,10 +197,58 @@ link_unregister_object (PinosLink *link)
   pinos_daemon_unexport (priv->daemon, priv->object_path);
 }
 
+static gboolean
+on_activate (PinosPort *port, gpointer user_data)
+{
+  PinosLink *link = user_data;
+  PinosLinkPrivate *priv = link->priv;
+
+  if (priv->input == port)
+    pinos_port_activate (priv->output);
+  else
+    pinos_port_activate (priv->input);
+
+  return TRUE;
+}
+
+static gboolean
+on_deactivate (PinosPort *port, gpointer user_data)
+{
+  PinosLink *link = user_data;
+  PinosLinkPrivate *priv = link->priv;
+
+  if (priv->input == port)
+    pinos_port_deactivate (priv->output);
+  else
+    pinos_port_deactivate (priv->input);
+
+  return TRUE;
+}
+
 static void
 pinos_link_constructed (GObject * object)
 {
   PinosLink *link = PINOS_LINK (object);
+  PinosLinkPrivate *priv = link->priv;
+  GBytes *formats;
+
+  pinos_port_add_send_buffer_cb (priv->output,
+                                 on_output_send,
+                                 link,
+                                 NULL);
+
+  pinos_port_add_send_buffer_cb (priv->input,
+                                 on_input_send,
+                                 link,
+                                 NULL);
+
+  g_object_get (priv->input, "possible-formats", &formats, NULL);
+  g_object_set (priv->output, "possible-formats", formats, NULL);
+
+  g_signal_connect (priv->input, "activate", (GCallback) on_activate, link);
+  g_signal_connect (priv->input, "deactivate", (GCallback) on_deactivate, link);
+  g_signal_connect (priv->output, "activate", (GCallback) on_activate, link);
+  g_signal_connect (priv->output, "deactivate", (GCallback) on_deactivate, link);
 
   g_debug ("link %p: constructed", link);
   link_register_object (link);
@@ -206,7 +276,6 @@ pinos_link_finalize (GObject * object)
   g_debug ("link %p: finalize", link);
   g_clear_pointer (&priv->possible_formats, g_bytes_unref);
   g_clear_pointer (&priv->format, g_bytes_unref);
-  g_clear_pointer (&priv->properties, pinos_properties_free);
   g_clear_object (&priv->daemon);
   g_clear_object (&priv->iface);
   g_free (priv->object_path);
@@ -238,6 +307,26 @@ pinos_link_class_init (PinosLinkClass * klass)
                                                         G_PARAM_CONSTRUCT_ONLY |
                                                         G_PARAM_STATIC_STRINGS));
 
+  g_object_class_install_property (gobject_class,
+                                   PROP_OUTPUT,
+                                   g_param_spec_object ("output",
+                                                        "Output",
+                                                        "The output port",
+                                                        PINOS_TYPE_PORT,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_CONSTRUCT |
+                                                        G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_INPUT,
+                                   g_param_spec_object ("input",
+                                                        "Input",
+                                                        "The input port",
+                                                        PINOS_TYPE_PORT,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_CONSTRUCT |
+                                                        G_PARAM_STATIC_STRINGS));
+
   g_object_class_install_property (gobject_class,
                                    PROP_POSSIBLE_FORMATS,
                                    g_param_spec_boxed ("possible-formats",
@@ -258,16 +347,6 @@ pinos_link_class_init (PinosLinkClass * klass)
                                                        G_PARAM_CONSTRUCT |
                                                        G_PARAM_STATIC_STRINGS));
 
-  g_object_class_install_property (gobject_class,
-                                   PROP_PROPERTIES,
-                                   g_param_spec_boxed ("properties",
-                                                       "Properties",
-                                                       "The properties of the link",
-                                                       PINOS_TYPE_PROPERTIES,
-                                                       G_PARAM_READWRITE |
-                                                       G_PARAM_CONSTRUCT_ONLY |
-                                                       G_PARAM_STATIC_STRINGS));
-
   signals[SIGNAL_REMOVE] = g_signal_new ("remove",
                                          G_TYPE_FROM_CLASS (klass),
                                          G_SIGNAL_RUN_LAST,
@@ -278,26 +357,6 @@ pinos_link_class_init (PinosLinkClass * klass)
                                          G_TYPE_NONE,
                                          0,
                                          G_TYPE_NONE);
- signals[SIGNAL_ACTIVATE] = g_signal_new ("activate",
-                                          G_TYPE_FROM_CLASS (klass),
-                                          G_SIGNAL_RUN_LAST,
-                                          0,
-                                          NULL,
-                                          NULL,
-                                          g_cclosure_marshal_generic,
-                                          G_TYPE_NONE,
-                                          0,
-                                          G_TYPE_NONE);
-  signals[SIGNAL_DEACTIVATE] = g_signal_new ("deactivate",
-                                             G_TYPE_FROM_CLASS (klass),
-                                             G_SIGNAL_RUN_LAST,
-                                             0,
-                                             NULL,
-                                             NULL,
-                                             g_cclosure_marshal_generic,
-                                             G_TYPE_NONE,
-                                             0,
-                                             G_TYPE_NONE);
 }
 
 static void
@@ -309,6 +368,31 @@ pinos_link_init (PinosLink * link)
   g_debug ("link %p: new", link);
 }
 
+PinosLink *
+pinos_link_new (PinosDaemon *daemon,
+                PinosPort   *output,
+                PinosPort   *input,
+                GBytes      *format_filter)
+{
+  PinosLink *link;
+  PinosPort *tmp;
+
+  if (pinos_port_get_direction (output) != PINOS_DIRECTION_OUTPUT) {
+    tmp = output;
+    output = input;
+    input = tmp;
+  }
+
+  link = g_object_new (PINOS_TYPE_LINK,
+                       "daemon", daemon,
+                       "output", output,
+                       "input", input,
+                       "possible-formats", format_filter,
+                       NULL);
+
+  return link;
+}
+
 /**
  * pinos_link_remove:
  * @link: a #PinosLink
@@ -380,22 +464,3 @@ pinos_link_get_format (PinosLink *link)
 
   return priv->format;
 }
-
-/**
- * pinos_link_get_properties:
- * @link: a #PinosLink
- *
- * Get the properties of @link
- *
- * Returns: the properties or %NULL
- */
-PinosProperties *
-pinos_link_get_properties (PinosLink *link)
-{
-  PinosLinkPrivate *priv;
-
-  g_return_val_if_fail (PINOS_IS_LINK (link), NULL);
-  priv = link->priv;
-
-  return priv->properties;
-}

pinos/server/link.h

@@ -62,6 +62,11 @@ struct _PinosLinkClass {
 /* normal GObject stuff */
 GType               pinos_link_get_type             (void);
 
+PinosLink *         pinos_link_new                  (PinosDaemon *daemon,
+                                                     PinosPort   *output,
+                                                     PinosPort   *input,
+                                                     GBytes      *format_filter);
+
 void                pinos_link_remove               (PinosLink *link);
 
 GBytes *            pinos_link_get_possible_formats (PinosLink *link);

pinos/server/node.c

@@ -49,7 +49,7 @@ struct _PinosNodePrivate
 
   PinosProperties *properties;
 
-  GList *ports;
+  GHashTable *ports;
 };
 
 G_DEFINE_ABSTRACT_TYPE (PinosNode, pinos_node, G_TYPE_OBJECT);
@@ -83,6 +83,7 @@ node_set_state (PinosNode       *node,
 static PinosPort *
 node_add_port (PinosNode       *node,
                PinosDirection   direction,
+               guint            id,
                GError         **error)
 {
   PinosNodePrivate *priv = node->priv;
@@ -92,20 +93,23 @@ node_add_port (PinosNode       *node,
                        "daemon", priv->daemon,
                        "node", node,
                        "direction", direction,
+                       "id", id,
                        NULL);
   return port;
 }
 
-static void
+static gboolean
 node_remove_port (PinosNode       *node,
-                  PinosPort       *port)
+                  guint            id)
 {
+  return TRUE;
 }
 
 static gboolean
 handle_add_port (PinosNode1             *interface,
                  GDBusMethodInvocation  *invocation,
                  PinosDirection          arg_direction,
+                 guint                   arg_id,
                  gpointer                user_data)
 {
   PinosNode *node = user_data;
@@ -118,13 +122,13 @@ handle_add_port (PinosNode1             *interface,
   if (g_strcmp0 (priv->sender, sender) != 0)
     goto not_allowed;
 
-  port = pinos_node_add_port (node, arg_direction, &error);
+  port = pinos_node_add_port (node, arg_direction, arg_id, &error);
   if (port == NULL)
     goto no_port;
 
   g_debug ("node %p: add port %p", node, port);
   g_dbus_method_invocation_return_value (invocation,
-                                         g_variant_new ("(u)", 1));
+                                         g_variant_new ("()"));
 
   return TRUE;
 
@@ -145,6 +149,47 @@ no_port:
   }
 }
 
+static gboolean
+handle_remove_port (PinosNode1             *interface,
+                    GDBusMethodInvocation  *invocation,
+                    guint                   arg_id,
+                    gpointer                user_data)
+{
+  PinosNode *node = user_data;
+  PinosNodePrivate *priv = node->priv;
+  const gchar *sender;
+
+  sender = g_dbus_method_invocation_get_sender (invocation);
+  if (g_strcmp0 (priv->sender, sender) != 0)
+    goto not_allowed;
+
+  if (!pinos_node_remove_port (node, arg_id))
+    goto no_port;
+
+  g_debug ("node %p: remove port %u", node, arg_id);
+  g_dbus_method_invocation_return_value (invocation,
+                                         g_variant_new ("()"));
+
+
+  return TRUE;
+
+not_allowed:
+  {
+    g_debug ("sender %s is not owner of node with sender %s", sender, priv->sender);
+    g_dbus_method_invocation_return_dbus_error (invocation,
+                 "org.pinos.Error", "not node owner");
+    return TRUE;
+  }
+no_port:
+  {
+    g_debug ("node %p: could remove port", node);
+    g_dbus_method_invocation_return_dbus_error (invocation,
+                 "org.pinos.Error", "can't remove port");
+    return TRUE;
+  }
+}
+
+
 static gboolean
 handle_remove (PinosNode1             *interface,
                GDBusMethodInvocation  *invocation,
@@ -315,8 +360,7 @@ pinos_node_dispose (GObject * obj)
   g_debug ("node %p: dispose", node);
   node_unregister_object (node);
 
-  g_list_free_full (priv->ports, (GDestroyNotify) g_object_unref);
+  g_hash_table_unref (priv->ports);
-  priv->ports = NULL;
 
   G_OBJECT_CLASS (pinos_node_parent_class)->dispose (obj);
 }
@@ -438,11 +482,16 @@ pinos_node_init (PinosNode * node)
   g_signal_connect (priv->iface, "handle-add-port",
                                  (GCallback) handle_add_port,
                                  node);
+  g_signal_connect (priv->iface, "handle-remove-port",
+                                 (GCallback) handle_remove_port,
+                                 node);
   g_signal_connect (priv->iface, "handle-remove",
                                  (GCallback) handle_remove,
                                  node);
   priv->state = PINOS_NODE_STATE_SUSPENDED;
   pinos_node1_set_state (priv->iface, PINOS_NODE_STATE_SUSPENDED);
+
+  priv->ports = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL, (GDestroyNotify) g_object_unref);
 }
 
 /**
@@ -602,9 +651,9 @@ pinos_node_remove (PinosNode *node)
 }
 
 static void
-handle_remove_port (PinosPort *port, PinosNode *node)
+do_remove_port (PinosPort *port, PinosNode *node)
 {
-  pinos_node_remove_port (node, port);
+  pinos_node_remove_port (node, pinos_port_get_id (port));
 }
 
 /**
@@ -620,6 +669,7 @@ handle_remove_port (PinosPort *port, PinosNode *node)
 PinosPort *
 pinos_node_add_port (PinosNode       *node,
                      PinosDirection   direction,
+                     guint            id,
                      GError         **error)
 {
   PinosNodeClass *klass;
@@ -636,54 +686,74 @@ pinos_node_add_port (PinosNode       *node,
   }
 
   g_debug ("node %p: add port", node);
-  port = klass->add_port (node, direction, error);
+  port = klass->add_port (node, direction, id, error);
 
   if (port) {
-    priv->ports = g_list_append (priv->ports, port);
+    g_hash_table_insert (priv->ports, GUINT_TO_POINTER (id), port);
-    g_signal_connect (port, "remove", (GCallback) handle_remove_port, node);
+    g_signal_connect (port, "remove", (GCallback) do_remove_port, node);
   }
-
   return port;
 }
 
 /**
  * pinos_node_remove_port:
  * @node: a #PinosNode
- * @port: (transfer full): a #PinosPort
+ * @id: a #PinosPort id
  *
- * Remove the #PinosPort from @node
+ * Remove the #PinosPort with @id from @node
+ *
+ * Returns: %TRUE when the port was removed
  */
-void
+gboolean
-pinos_node_remove_port (PinosNode *node, PinosPort *port)
+pinos_node_remove_port (PinosNode *node, guint id)
 {
   PinosNodePrivate *priv;
   PinosNodeClass *klass;
-  GList *find;
+  gboolean res = FALSE;
 
-  g_return_if_fail (PINOS_IS_NODE (node));
+  g_return_val_if_fail (PINOS_IS_NODE (node), FALSE);
-  g_return_if_fail (PINOS_IS_PORT (port));
   priv = node->priv;
 
-  find = g_list_find (priv->ports, port);
+  klass = PINOS_NODE_GET_CLASS (node);
-  if (find) {
-    klass = PINOS_NODE_GET_CLASS (node);
 
-    if (klass->remove_port)
+  if (!klass->remove_port)
-      klass->remove_port (node, port);
+    return FALSE;
 
-    g_debug ("node %p: removed port %p", node, port);
+  res = klass->remove_port (node, id);
-    priv->ports = g_list_delete_link (priv->ports, find);
+  if (res) {
-    g_object_unref (port);
+    g_debug ("node %p: removed port %u", node, id);
+    g_hash_table_remove (priv->ports, GUINT_TO_POINTER (id));
   }
+  return res;
+}
+
+/**
+ * pinos_node_find_port:
+ * @node: a #PinosNode
+ * @id: a #PinosPort id
+ *
+ * Get the port with @id @node.
+ *
+ * Returns: a #PinosPort with @id or %NULL when not found
+ */
+PinosPort *
+pinos_node_find_port (PinosNode *node, guint id)
+{
+  PinosNodePrivate *priv;
+
+  g_return_val_if_fail (PINOS_IS_NODE (node), NULL);
+  priv = node->priv;
+
+  return g_hash_table_lookup (priv->ports, GUINT_TO_POINTER (id));
 }
 
 /**
  * pinos_node_get_ports:
  * @node: a #PinosNode
  *
- * Get the list of ports in @node.
+ * Get the ports in @node.
  *
- * Returns: a #GList of ports owned by @node.
+ * Returns: a #GList of ports g_list_free after usage.
  */
 GList *
 pinos_node_get_ports (PinosNode *node)
@@ -693,7 +763,7 @@ pinos_node_get_ports (PinosNode *node)
   g_return_val_if_fail (PINOS_IS_NODE (node), NULL);
   priv = node->priv;
 
-  return priv->ports;
+  return g_hash_table_get_values (priv->ports);
 }
 
 static void

pinos/server/node.h

@@ -66,9 +66,10 @@ struct _PinosNodeClass {
 
   PinosPort * (*add_port)     (PinosNode       *node,
                                PinosDirection   direction,
+                               guint            id,
                                GError         **error);
-  void        (*remove_port)  (PinosNode       *node,
+  gboolean    (*remove_port)  (PinosNode       *node,
-                               PinosPort       *port);
+                               guint            id);
 };
 
 /* normal GObject stuff */
@@ -90,9 +91,12 @@ const gchar *       pinos_node_get_object_path         (PinosNode       *node);
 
 PinosPort *         pinos_node_add_port                (PinosNode       *node,
                                                         PinosDirection   direction,
+                                                        guint            id,
                                                         GError         **error);
-void                pinos_node_remove_port             (PinosNode       *node,
+gboolean            pinos_node_remove_port             (PinosNode       *node,
-                                                        PinosPort       *port);
+                                                        guint            id);
+PinosPort *         pinos_node_find_port               (PinosNode       *node,
+                                                        guint            id);
 GList *             pinos_node_get_ports               (PinosNode       *node);
 
 PinosNodeState      pinos_node_get_state               (PinosNode *node);

pinos/server/port.c

@@ -48,7 +48,7 @@ struct _PinosPortPrivate
 {
   PinosDaemon *daemon;
 
-  gulong id;
+  guint id;
 
   PinosNode *node;
   PinosDirection direction;
@@ -75,6 +75,7 @@ enum
   PROP_DAEMON,
   PROP_NODE,
   PROP_DIRECTION,
+  PROP_ID,
   PROP_POSSIBLE_FORMATS,
   PROP_FORMAT,
   PROP_PROPERTIES,
@@ -183,6 +184,10 @@ pinos_port_get_property (GObject    *_object,
       g_value_set_enum (value, priv->direction);
       break;
 
+    case PROP_ID:
+      g_value_set_uint (value, priv->id);
+      break;
+
     case PROP_POSSIBLE_FORMATS:
       g_value_set_boxed (value, priv->possible_formats);
       break;
@@ -223,6 +228,10 @@ pinos_port_set_property (GObject      *_object,
       priv->direction = g_value_get_enum (value);
       break;
 
+    case PROP_ID:
+      priv->id = g_value_get_uint (value);
+      break;
+
     case PROP_POSSIBLE_FORMATS:
       if (priv->possible_formats)
         g_bytes_unref (priv->possible_formats);
@@ -336,6 +345,17 @@ pinos_port_class_init (PinosPortClass * klass)
                                                       G_PARAM_READWRITE |
                                                       G_PARAM_CONSTRUCT_ONLY |
                                                       G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (gobject_class,
+                                   PROP_ID,
+                                   g_param_spec_uint ("id",
+                                                      "Id",
+                                                      "The id of the port",
+                                                      0,
+                                                      G_MAXUINT,
+                                                      0,
+                                                      G_PARAM_READWRITE |
+                                                      G_PARAM_CONSTRUCT_ONLY |
+                                                      G_PARAM_STATIC_STRINGS));
 
   g_object_class_install_property (gobject_class,
                                    PROP_POSSIBLE_FORMATS,
@@ -364,7 +384,7 @@ pinos_port_class_init (PinosPortClass * klass)
                                                        "The properties of the port",
                                                        PINOS_TYPE_PROPERTIES,
                                                        G_PARAM_READWRITE |
-                                                       G_PARAM_CONSTRUCT_ONLY |
+                                                       G_PARAM_CONSTRUCT |
                                                        G_PARAM_STATIC_STRINGS));
 
 
@@ -473,6 +493,25 @@ pinos_port_get_direction (PinosPort *port)
   return priv->direction;
 }
 
+/**
+ * pinos_port_get_id:
+ * @port: a #PinosPort
+ *
+ * Get the id of @port
+ *
+ * Returns: the id or %NULL
+ */
+guint
+pinos_port_get_id (PinosPort *port)
+{
+  PinosPortPrivate *priv;
+
+  g_return_val_if_fail (PINOS_IS_PORT (port), -1);
+  priv = port->priv;
+
+  return priv->id;
+}
+
 /**
  * pinos_port_get_possible_formats:
  * @port: a #PinosPort

pinos/server/port.h

@@ -83,6 +83,7 @@ void                pinos_port_remove               (PinosPort *port);
 PinosNode *         pinos_port_get_node             (PinosPort *port);
 const gchar *       pinos_port_get_name             (PinosPort *port);
 PinosDirection      pinos_port_get_direction        (PinosPort *port);
+guint               pinos_port_get_id               (PinosPort *port);
 PinosProperties *   pinos_port_get_properties       (PinosPort *port);
 
 GBytes *            pinos_port_get_possible_formats (PinosPort *port);