Introduce the concept of a Node

Make an object for a processing node. Implement a sink node. Make it possible to implement Sink and Source interfaces to provide input/output from the node. Improve pinosdepay to track fds and handle format changes.
author: Wim Taymans <wtaymans@redhat.com> 2016-05-05 13:31:18 +0200
committer: Wim Taymans <wtaymans@redhat.com> 2016-05-05 13:31:18 +0200
commit: b885d40390bb2398e827602cb729aadb4022af2a (patch)
tree: db347f142b2f0d1d5d49745a06fcbbe3aa9b7d45
parent: 7597e48e020e95ef0808c94d44468530d91c098a (diff)
27 files changed, 3149 insertions, 159 deletions
diff --git a/doc/design.txt b/doc/design.txt
index b15a5f50..c9af8353 100644
--- a/doc/design.txt
+++ b/doc/design.txt
@@ -21,6 +21,26 @@ this should be handled by a separate consumer rendering the media to
 a specific output device.
 
 
+Objects
+-------
+
+Daemon1:   the main pinos daemon
+           /org/pinos/server
+Client1:   a connected client, the result object from call
+           Daemon1.ConnectClient
+           /org/pinos/client*
+Device1:   a physical device on the machine, devices can provide
+           processing nodes
+           /org/pinos/device*
+Node1:     a processing node, this can be a source, sink or transform
+           element.
+           /org/pinos/node*
+Port1:     a port on a Node1, ports can be input or output ports
+           /org/pinos/node*/port*
+Channel1:  a connection of a client on a port
+           /org/pinos/client*/channel*
+
+
 DBus protocol
 -------------
 
diff --git a/pinos/Makefile.am b/pinos/Makefile.am
index 07100082..c650e1bc 100644
--- a/pinos/Makefile.am
+++ b/pinos/Makefile.am
@@ -205,11 +205,14 @@ lib_LTLIBRARIES += libpinoscore-@PINOS_MAJORMINOR@.la
 libpinoscore_@PINOS_MAJORMINOR@_la_SOURCES = \
 		server/client.c server/client.h \
 		server/daemon.c server/daemon.h \
+		server/node.c server/node.h \
 		server/source.c server/source.h \
+		server/sink.c server/sink.h \
 		server/client-source.c server/client-source.h \
 		server/channel.c server/channel.h \
 		modules/gst/gst-manager.c modules/gst/gst-manager.h \
 		modules/gst/gst-source.c modules/gst/gst-source.h \
+		modules/gst/gst-sink.c modules/gst/gst-sink.h \
 		dbus/org-pinos.c dbus/org-pinos.h
 
 libpinoscore_@PINOS_MAJORMINOR@_la_CFLAGS = $(AM_CFLAGS) $(SERVER_CFLAGS)
diff --git a/pinos/client/pinos.h b/pinos/client/pinos.h
index 316b385d..3129c1e8 100644
--- a/pinos/client/pinos.h
+++ b/pinos/client/pinos.h
@@ -33,7 +33,7 @@
 #define PINOS_DBUS_SERVICE "org.pinos"
 #define PINOS_DBUS_OBJECT_PREFIX "/org/pinos"
 #define PINOS_DBUS_OBJECT_SERVER PINOS_DBUS_OBJECT_PREFIX "/server"
-#define PINOS_DBUS_OBJECT_SOURCE PINOS_DBUS_OBJECT_PREFIX "/source"
+#define PINOS_DBUS_OBJECT_NODE PINOS_DBUS_OBJECT_PREFIX "/node"
 #define PINOS_DBUS_OBJECT_CLIENT PINOS_DBUS_OBJECT_PREFIX "/client"
 
 void pinos_init (int *argc, char **argv[]);
diff --git a/pinos/client/stream.c b/pinos/client/stream.c
index b8cd906d..5008d513 100644
--- a/pinos/client/stream.c
+++ b/pinos/client/stream.c
@@ -39,7 +39,7 @@ struct _PinosStreamPrivate
   PinosStreamState state;
   GError *error;
 
-  gchar *source_path;
+  gchar *path;
   GBytes *possible_formats;
   gboolean provide;
 
@@ -240,7 +240,7 @@ pinos_stream_finalize (GObject * object)
   if (priv->format)
     g_bytes_unref (priv->format);
 
-  g_free (priv->source_path);
+  g_free (priv->path);
   if (priv->possible_formats)
     g_bytes_unref (priv->possible_formats);
 
@@ -502,20 +502,20 @@ on_channel_proxy (GObject      *source_object,
   GError *error = NULL;
 
   priv->channel = pinos_subscribe_get_proxy_finish (context->priv->subscribe,
-                                                       res,
-                                                       &error);
+                                                    res,
+                                                    &error);
   if (priv->channel == NULL)
     goto channel_failed;
 
   /* get the source we are connected to */
-  v = g_dbus_proxy_get_cached_property (priv->channel, "Source");
+  v = g_dbus_proxy_get_cached_property (priv->channel, "Owner");
   if (v) {
     gsize len;
     str = g_variant_dup_string (v, &len);
     g_variant_unref (v);
 
-    g_free (priv->source_path);
-    priv->source_path = str;
+    g_free (priv->path);
+    priv->path = str;
   }
 
   v = g_dbus_proxy_get_cached_property (priv->channel, "PossibleFormats");
@@ -604,7 +604,7 @@ do_connect_source (PinosStream *stream)
   g_dbus_proxy_call (context->priv->client,
                      "CreateSourceChannel",
                      g_variant_new ("(ss@a{sv})",
-                       (priv->source_path ? priv->source_path : ""),
+                       (priv->path ? priv->path : ""),
                        g_bytes_get_data (priv->possible_formats, NULL),
                        pinos_properties_to_variant (priv->properties)),
                      G_DBUS_CALL_FLAGS_NONE,
@@ -644,8 +644,8 @@ pinos_stream_connect_source (PinosStream      *stream,
   g_return_val_if_fail (pinos_context_get_state (context) == PINOS_CONTEXT_STATE_READY, FALSE);
   g_return_val_if_fail (pinos_stream_get_state (stream) == PINOS_STREAM_STATE_UNCONNECTED, FALSE);
 
-  g_free (priv->source_path);
-  priv->source_path = g_strdup (source_path);
+  g_free (priv->path);
+  priv->path = g_strdup (source_path);
   if (priv->possible_formats)
     g_bytes_unref (priv->possible_formats);
   priv->possible_formats = possible_formats;
@@ -661,6 +661,71 @@ pinos_stream_connect_source (PinosStream      *stream,
 }
 
 static gboolean
+do_connect_sink (PinosStream *stream)
+{
+  PinosStreamPrivate *priv = stream->priv;
+  PinosContext *context = priv->context;
+
+  g_dbus_proxy_call (context->priv->client,
+                     "CreateSinkChannel",
+                     g_variant_new ("(ss@a{sv})",
+                       (priv->path ? priv->path : ""),
+                       g_bytes_get_data (priv->possible_formats, NULL),
+                       pinos_properties_to_variant (priv->properties)),
+                     G_DBUS_CALL_FLAGS_NONE,
+                     -1,
+                     NULL, /* GCancellable *cancellable */
+                     on_channel_created,
+                     stream);
+
+  return FALSE;
+}
+
+/**
+ * pinos_stream_connect_sink:
+ * @stream: a #PinosStream
+ * @sink_path: the sink path to connect to
+ * @flags: a #PinosStreamFlags
+ * @possible_formats: (transfer full): a #GBytes with possible accepted formats
+ *
+ * Connect @stream for playback to @sink_path.
+ *
+ * Returns: %TRUE on success.
+ */
+gboolean
+pinos_stream_connect_sink (PinosStream      *stream,
+                           const gchar      *sink_path,
+                           PinosStreamFlags  flags,
+                           GBytes           *possible_formats)
+{
+  PinosStreamPrivate *priv;
+  PinosContext *context;
+
+  g_return_val_if_fail (PINOS_IS_STREAM (stream), FALSE);
+  g_return_val_if_fail (possible_formats != NULL, FALSE);
+
+  priv = stream->priv;
+  context = priv->context;
+  g_return_val_if_fail (pinos_context_get_state (context) == PINOS_CONTEXT_STATE_READY, FALSE);
+  g_return_val_if_fail (pinos_stream_get_state (stream) == PINOS_STREAM_STATE_UNCONNECTED, FALSE);
+
+  g_free (priv->path);
+  priv->path = g_strdup (sink_path);
+  if (priv->possible_formats)
+    g_bytes_unref (priv->possible_formats);
+  priv->possible_formats = possible_formats;
+  priv->provide = FALSE;
+
+  stream_set_state (stream, PINOS_STREAM_STATE_CONNECTING, NULL);
+
+  g_main_context_invoke (context->priv->context,
+                         (GSourceFunc) do_connect_sink,
+                         g_object_ref (stream));
+
+  return TRUE;
+}
+
+static gboolean
 do_connect_provide (PinosStream *stream)
 {
   PinosStreamPrivate *priv = stream->priv;
diff --git a/pinos/gst/gstpinosdepay.c b/pinos/gst/gstpinosdepay.c
index 8ef2963a..2cc1a764 100644
--- a/pinos/gst/gstpinosdepay.c
+++ b/pinos/gst/gstpinosdepay.c
@@ -53,8 +53,13 @@
 GST_DEBUG_CATEGORY_STATIC (gst_pinos_depay_debug_category);
 #define GST_CAT_DEFAULT gst_pinos_depay_debug_category
 
-/* prototypes */
+static GQuark fdids_quark;
 
+enum
+{
+  PROP_0,
+  PROP_CAPS,
+};
 
 /* pad templates */
 static GstStaticPadTemplate gst_pinos_depay_src_template =
@@ -71,22 +76,97 @@ GST_STATIC_PAD_TEMPLATE ("sink",
 
 
 /* class initialization */
+G_DEFINE_TYPE (GstPinosDepay, gst_pinos_depay, GST_TYPE_ELEMENT);
+
+static gboolean
+gst_pinos_depay_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
+{
+  GstPinosDepay *depay = GST_PINOS_DEPAY (parent);
+  gboolean res = FALSE;
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_SEGMENT:
+    {
+      GstSegment segment;
+
+      gst_segment_init (&segment, GST_FORMAT_TIME);
+
+      res = gst_pad_push_event (depay->srcpad, gst_event_new_segment (&segment));
+      break;
+    }
+    case GST_EVENT_CAPS:
+    {
+      GstCaps *caps;
+      GstStructure *str;
+
+      gst_event_parse_caps (event, &caps);
+      str = gst_caps_get_structure (caps, 0);
+      depay->pinos_input = gst_structure_has_name (str, "application/x-pinos");
+      gst_event_unref (event);
+
+      res = gst_pad_push_event (depay->srcpad, gst_event_new_caps (depay->caps));
+      break;
+    }
+    default:
+      res = gst_pad_event_default (pad, parent, event);
+      break;
+  }
+  return res;
+}
+
+static void
+release_fds (GstPinosDepay *this, GstBuffer *buffer)
+{
+  GArray *fdids;
+  guint i;
+  PinosBufferBuilder b;
+  PinosPacketReleaseFDPayload r;
+  PinosBuffer pbuf;
+  gsize size;
+  gpointer data;
+  GstBuffer *outbuf;
+  GstEvent *ev;
+
+  fdids = gst_mini_object_steal_qdata (GST_MINI_OBJECT_CAST (buffer),
+      fdids_quark);
+  if (fdids == NULL)
+    return;
 
-G_DEFINE_TYPE_WITH_CODE (GstPinosDepay, gst_pinos_depay, GST_TYPE_ELEMENT,
-    GST_DEBUG_CATEGORY_INIT (gst_pinos_depay_debug_category, "pinosdepay", 0,
-        "debug category for pinosdepay element"));
+  pinos_buffer_builder_init (&b);
+
+  for (i = 0; i < fdids->len; i++) {
+    r.id = g_array_index (fdids, guint32, i);
+    GST_LOG ("release fd index %d", r.id);
+    pinos_buffer_builder_add_release_fd_payload (&b, &r);
+  }
+  pinos_buffer_builder_end (&b, &pbuf);
+  g_array_unref (fdids);
+
+  data = pinos_buffer_steal (&pbuf, &size, NULL);
+
+  outbuf = gst_buffer_new_wrapped (data, size);
+  ev = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
+          gst_structure_new ("GstNetworkMessage",
+              "object", G_TYPE_OBJECT, this,
+              "buffer", GST_TYPE_BUFFER, outbuf, NULL));
+  gst_buffer_unref (outbuf);
+
+  gst_pad_push_event (this->sinkpad, ev);
+  g_object_unref (this);
+}
 
 static GstFlowReturn
 gst_pinos_depay_chain (GstPad *pad, GstObject * parent, GstBuffer * buffer)
 {
   GstPinosDepay *depay = GST_PINOS_DEPAY (parent);
-  GstBuffer *outbuf;
+  GstBuffer *outbuf = NULL;
   GstMapInfo info;
   PinosBuffer pbuf;
   PinosBufferIter it;
   GstNetControlMessageMeta * meta;
   GSocketControlMessage *msg = NULL;
   GError *err = NULL;
+  GArray *fdids = NULL;
 
   meta = ((GstNetControlMessageMeta*) gst_buffer_get_meta (
       buffer, GST_NET_CONTROL_MESSAGE_META_API_TYPE));
@@ -96,13 +176,6 @@ gst_pinos_depay_chain (GstPad *pad, GstObject * parent, GstBuffer * buffer)
     meta = NULL;
   }
 
-  if (msg == NULL) {
-    gst_buffer_unref (buffer);
-    return GST_FLOW_OK;
-  }
-
-  outbuf = gst_buffer_new ();
-
   gst_buffer_map (buffer, &info, GST_MAP_READ);
   pinos_buffer_init_data (&pbuf, info.data, info.size, msg);
 
@@ -113,9 +186,21 @@ gst_pinos_depay_chain (GstPad *pad, GstObject * parent, GstBuffer * buffer)
       {
         PinosPacketHeader hdr;
 
-        if (!pinos_buffer_iter_parse_header (&it, &hdr))
+        if (!pinos_buffer_iter_parse_header  (&it, &hdr))
           goto error;
 
+        if (outbuf == NULL)
+          outbuf = gst_buffer_new ();
+
+        GST_INFO ("pts %" G_GUINT64_FORMAT ", dts_offset %"G_GUINT64_FORMAT, hdr.pts, hdr.dts_offset);
+
+#if 0
+        if (GST_CLOCK_TIME_IS_VALID (hdr.pts)) {
+          GST_BUFFER_PTS (outbuf) = hdr.pts;
+          if (GST_BUFFER_PTS (outbuf) + hdr.dts_offset > 0)
+            GST_BUFFER_DTS (outbuf) = GST_BUFFER_PTS (outbuf) + hdr.dts_offset;
+        }
+#endif
         GST_BUFFER_OFFSET (outbuf) = hdr.seq;
         break;
       }
@@ -131,10 +216,35 @@ gst_pinos_depay_chain (GstPad *pad, GstObject * parent, GstBuffer * buffer)
         if (fd == -1)
           goto error;
 
+        if (outbuf == NULL)
+          outbuf = gst_buffer_new ();
+
         fdmem = gst_fd_allocator_alloc (depay->fd_allocator, fd,
                   p.offset + p.size, GST_FD_MEMORY_FLAG_NONE);
         gst_memory_resize (fdmem, p.offset, p.size);
         gst_buffer_append_memory (outbuf, fdmem);
+
+        if (fdids == NULL)
+          fdids = g_array_new (FALSE, FALSE, sizeof (guint32));
+
+        GST_LOG ("track fd index %d", p.id);
+        g_array_append_val (fdids, p.id);
+        break;
+      }
+      case PINOS_PACKET_TYPE_FORMAT_CHANGE:
+      {
+        PinosPacketFormatChange change;
+        GstCaps *caps;
+
+        if (!pinos_buffer_iter_parse_format_change (&it, &change))
+          goto error;
+        GST_DEBUG ("got format change %d %s", change.id, change.format);
+
+        caps = gst_caps_from_string (change.format);
+        if (caps) {
+          gst_caps_take (&depay->caps, caps);
+          gst_pad_push_event (depay->srcpad, gst_event_new_caps (depay->caps));
+        }
         break;
       }
       default:
@@ -145,7 +255,17 @@ gst_pinos_depay_chain (GstPad *pad, GstObject * parent, GstBuffer * buffer)
   gst_buffer_unmap (buffer, &info);
   gst_buffer_unref (buffer);
 
-  return gst_pad_push (depay->srcpad, outbuf);
+  if (outbuf) {
+    if (fdids != NULL) {
+      gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (outbuf),
+          fdids_quark, fdids, NULL);
+      gst_mini_object_weak_ref (GST_MINI_OBJECT_CAST (outbuf),
+          (GstMiniObjectNotify) release_fds, g_object_ref (depay));
+    }
+    return gst_pad_push (depay->srcpad, outbuf);
+  }
+  else
+    return GST_FLOW_OK;
 
 error:
   {
@@ -158,12 +278,52 @@ error:
 }
 
 static void
+gst_pinos_depay_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  GstPinosDepay *depay = GST_PINOS_DEPAY (object);
+
+  switch (prop_id) {
+    case PROP_CAPS:
+    {
+      const GstCaps *caps;
+
+      caps = gst_value_get_caps (value);
+      gst_caps_replace (&depay->caps, (GstCaps *)caps);
+      break;
+    }
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pinos_depay_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec)
+{
+  GstPinosDepay *depay = GST_PINOS_DEPAY (object);
+
+  switch (prop_id) {
+    case PROP_CAPS:
+      gst_value_set_caps (value, depay->caps);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
 gst_pinos_depay_finalize (GObject * object)
 {
   GstPinosDepay *depay = GST_PINOS_DEPAY (object);
 
   GST_DEBUG_OBJECT (depay, "finalize");
 
+  gst_caps_replace (&depay->caps, NULL);
   g_object_unref (depay->fd_allocator);
 
   G_OBJECT_CLASS (gst_pinos_depay_parent_class)->finalize (object);
@@ -176,6 +336,15 @@ gst_pinos_depay_class_init (GstPinosDepayClass * klass)
   GstElementClass *element_class =
       GST_ELEMENT_CLASS (klass);
 
+  gobject_class->finalize = gst_pinos_depay_finalize;
+  gobject_class->set_property = gst_pinos_depay_set_property;
+  gobject_class->get_property = gst_pinos_depay_get_property;
+
+  g_object_class_install_property (gobject_class, PROP_CAPS,
+      g_param_spec_boxed ("caps", "Caps",
+          "The caps of the source pad", GST_TYPE_CAPS,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
   /* Setting up pads and setting metadata should be moved to
      base_class_init if you intend to subclass this class. */
   gst_element_class_add_pad_template (element_class,
@@ -188,7 +357,10 @@ gst_pinos_depay_class_init (GstPinosDepayClass * klass)
       "Pinos Depayloader for zero-copy IPC via Pinos",
       "Wim Taymans <wim.taymans@gmail.com>");
 
-  gobject_class->finalize = gst_pinos_depay_finalize;
+  GST_DEBUG_CATEGORY_INIT (gst_pinos_depay_debug_category, "pinosdepay", 0,
+      "debug category for pinosdepay element");
+
+  fdids_quark = g_quark_from_static_string ("GstPinosDepayFDIds");
 }
 
 static void
@@ -199,6 +371,7 @@ gst_pinos_depay_init (GstPinosDepay * depay)
 
   depay->sinkpad = gst_pad_new_from_static_template (&gst_pinos_depay_sink_template, "sink");
   gst_pad_set_chain_function (depay->sinkpad, gst_pinos_depay_chain);
+  gst_pad_set_event_function (depay->sinkpad, gst_pinos_depay_sink_event);
   gst_element_add_pad (GST_ELEMENT (depay), depay->sinkpad);
 
   depay->fd_allocator = gst_fd_allocator_new ();
diff --git a/pinos/gst/gstpinosdepay.h b/pinos/gst/gstpinosdepay.h
index a5af7113..527eb2c1 100644
--- a/pinos/gst/gstpinosdepay.h
+++ b/pinos/gst/gstpinosdepay.h
@@ -38,6 +38,9 @@ struct _GstPinosDepay
 {
   GstElement parent;
 
+  GstCaps *caps;
+  gboolean pinos_input;
+
   GstPad *srcpad, *sinkpad;
   GstAllocator *fd_allocator;
 };
diff --git a/pinos/gst/gstpinosprovide.c b/pinos/gst/gstpinosprovide.c
new file mode 100644
index 00000000..3cdbe360
--- /dev/null
+++ b/pinos/gst/gstpinosprovide.c
@@ -0,0 +1,737 @@
+/* GStreamer
+ * 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.
+ */
+
+/**
+ * SECTION:element-pinossink
+ *
+ * <refsect2>
+ * <title>Example launch line</title>
+ * |[
+ * gst-launch -v videotestsrc ! pinossink
+ * ]| Sends a test video source to pinos
+ * </refsect2>
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+#include "gstpinossink.h"
+
+#include <string.h>
+#include <stdlib.h>
+#include <fcntl.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <gio/gunixfdmessage.h>
+#include <gst/allocators/gstfdmemory.h>
+#include <gst/video/video.h>
+
+#include "gsttmpfileallocator.h"
+
+
+GST_DEBUG_CATEGORY_STATIC (pinos_sink_debug);
+#define GST_CAT_DEFAULT pinos_sink_debug
+
+enum
+{
+  PROP_0,
+  PROP_CLIENT_NAME,
+  PROP_STREAM_PROPERTIES
+};
+
+
+#define PINOSS_VIDEO_CAPS GST_VIDEO_CAPS_MAKE (GST_VIDEO_FORMATS_ALL)
+
+static GstStaticPadTemplate gst_pinos_sink_template =
+GST_STATIC_PAD_TEMPLATE ("sink",
+    GST_PAD_SINK,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY
+    );
+
+#define gst_pinos_sink_parent_class parent_class
+G_DEFINE_TYPE (GstPinosSink, gst_pinos_sink, GST_TYPE_BASE_SINK);
+
+static void gst_pinos_sink_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec);
+static void gst_pinos_sink_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec);
+
+static GstStateChangeReturn
+gst_pinos_sink_change_state (GstElement * element, GstStateChange transition);
+
+static gboolean gst_pinos_sink_setcaps (GstBaseSink * bsink, GstCaps * caps);
+static GstCaps *gst_pinos_sink_sink_fixate (GstBaseSink * bsink,
+    GstCaps * caps);
+
+static GstFlowReturn gst_pinos_sink_render (GstBaseSink * psink,
+    GstBuffer * buffer);
+static gboolean gst_pinos_sink_start (GstBaseSink * basesink);
+static gboolean gst_pinos_sink_stop (GstBaseSink * basesink);
+
+static void
+gst_pinos_sink_finalize (GObject * object)
+{
+  GstPinosSink *pinossink = GST_PINOS_SINK (object);
+
+  if (pinossink->properties)
+    gst_structure_free (pinossink->properties);
+  g_hash_table_unref (pinossink->fdids);
+  g_object_unref (pinossink->allocator);
+  g_free (pinossink->client_name);
+
+  G_OBJECT_CLASS (parent_class)->finalize (object);
+}
+
+static gboolean
+gst_pinos_sink_propose_allocation (GstBaseSink * bsink, GstQuery * query)
+{
+  GstPinosSink *pinossink = GST_PINOS_SINK (bsink);
+
+  gst_query_add_allocation_param (query, pinossink->allocator, NULL);
+  return TRUE;
+}
+
+static void
+gst_pinos_sink_class_init (GstPinosSinkClass * klass)
+{
+  GObjectClass *gobject_class;
+  GstElementClass *gstelement_class;
+  GstBaseSinkClass *gstbasesink_class;
+
+  gobject_class = (GObjectClass *) klass;
+  gstelement_class = (GstElementClass *) klass;
+  gstbasesink_class = (GstBaseSinkClass *) klass;
+
+  gobject_class->finalize = gst_pinos_sink_finalize;
+  gobject_class->set_property = gst_pinos_sink_set_property;
+  gobject_class->get_property = gst_pinos_sink_get_property;
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_CLIENT_NAME,
+                                   g_param_spec_string ("client-name",
+                                                        "Client Name",
+                                                        "The client name to use (NULL = default)",
+                                                        NULL,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_STATIC_STRINGS));
+
+   g_object_class_install_property (gobject_class,
+                                    PROP_STREAM_PROPERTIES,
+                                    g_param_spec_boxed ("stream-properties",
+                                                        "stream properties",
+                                                        "list of pinos stream properties",
+                                                        GST_TYPE_STRUCTURE,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_STATIC_STRINGS));
+
+  gstelement_class->change_state = gst_pinos_sink_change_state;
+
+  gst_element_class_set_static_metadata (gstelement_class,
+      "Pinos sink", "Sink/Video",
+      "Send video to pinos", "Wim Taymans <wim.taymans@gmail.com>");
+
+  gst_element_class_add_pad_template (gstelement_class,
+      gst_static_pad_template_get (&gst_pinos_sink_template));
+
+  gstbasesink_class->set_caps = gst_pinos_sink_setcaps;
+  gstbasesink_class->fixate = gst_pinos_sink_sink_fixate;
+  gstbasesink_class->propose_allocation = gst_pinos_sink_propose_allocation;
+  gstbasesink_class->start = gst_pinos_sink_start;
+  gstbasesink_class->stop = gst_pinos_sink_stop;
+  gstbasesink_class->render = gst_pinos_sink_render;
+
+  GST_DEBUG_CATEGORY_INIT (pinos_sink_debug, "pinossink", 0,
+      "Pinos Sink");
+}
+
+static void
+gst_pinos_sink_init (GstPinosSink * sink)
+{
+  sink->allocator = gst_tmpfile_allocator_new ();
+  sink->client_name = pinos_client_name();
+
+  sink->fdids = g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL,
+      (GDestroyNotify) gst_buffer_unref);
+}
+
+static GstCaps *
+gst_pinos_sink_sink_fixate (GstBaseSink * bsink, GstCaps * caps)
+{
+  GstStructure *structure;
+
+  caps = gst_caps_make_writable (caps);
+
+  structure = gst_caps_get_structure (caps, 0);
+
+  if (gst_structure_has_name (structure, "video/x-raw")) {
+    gst_structure_fixate_field_nearest_int (structure, "width", 320);
+    gst_structure_fixate_field_nearest_int (structure, "height", 240);
+    gst_structure_fixate_field_nearest_fraction (structure, "framerate", 30, 1);
+
+    if (gst_structure_has_field (structure, "pixel-aspect-ratio"))
+      gst_structure_fixate_field_nearest_fraction (structure,
+          "pixel-aspect-ratio", 1, 1);
+    else
+      gst_structure_set (structure, "pixel-aspect-ratio", GST_TYPE_FRACTION, 1, 1,
+          NULL);
+
+    if (gst_structure_has_field (structure, "colorimetry"))
+      gst_structure_fixate_field_string (structure, "colorimetry", "bt601");
+    if (gst_structure_has_field (structure, "chroma-site"))
+      gst_structure_fixate_field_string (structure, "chroma-site", "mpeg2");
+
+    if (gst_structure_has_field (structure, "interlace-mode"))
+      gst_structure_fixate_field_string (structure, "interlace-mode",
+          "progressive");
+    else
+      gst_structure_set (structure, "interlace-mode", G_TYPE_STRING,
+          "progressive", NULL);
+  } else if (gst_structure_has_name (structure, "audio/x-raw")) {
+    gst_structure_fixate_field_string (structure, "format", "S16LE");
+    gst_structure_fixate_field_nearest_int (structure, "channels", 2);
+    gst_structure_fixate_field_nearest_int (structure, "rate", 44100);
+  }
+
+  caps = GST_BASE_SINK_CLASS (parent_class)->fixate (bsink, caps);
+
+  return caps;
+}
+
+static void
+gst_pinos_sink_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  GstPinosSink *pinossink = GST_PINOS_SINK (object);
+
+  switch (prop_id) {
+    case PROP_CLIENT_NAME:
+      g_free (pinossink->client_name);
+      pinossink->client_name = g_value_dup_string (value);
+      break;
+
+    case PROP_STREAM_PROPERTIES:
+      if (pinossink->properties)
+        gst_structure_free (pinossink->properties);
+      pinossink->properties =
+          gst_structure_copy (gst_value_get_structure (value));
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pinos_sink_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec)
+{
+  GstPinosSink *pinossink = GST_PINOS_SINK (object);
+
+  switch (prop_id) {
+    case PROP_CLIENT_NAME:
+      g_value_set_string (value, pinossink->client_name);
+      break;
+
+    case PROP_STREAM_PROPERTIES:
+      gst_value_set_structure (value, pinossink->properties);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+on_new_buffer (GObject    *gobject,
+               gpointer    user_data)
+{
+  GstPinosSink *pinossink = user_data;
+  PinosBuffer *pbuf;
+  PinosBufferIter it;
+
+  GST_LOG_OBJECT (pinossink, "got new buffer");
+  if (!pinos_stream_peek_buffer (pinossink->stream, &pbuf)) {
+    g_warning ("failed to capture buffer");
+    return;
+  }
+
+  pinos_buffer_iter_init (&it, pbuf);
+  while (pinos_buffer_iter_next (&it)) {
+    switch (pinos_buffer_iter_get_type (&it)) {
+      case PINOS_PACKET_TYPE_RELEASE_FD_PAYLOAD:
+      {
+        PinosPacketReleaseFDPayload p;
+
+        if (!pinos_buffer_iter_parse_release_fd_payload (&it, &p))
+          continue;
+
+        GST_LOG ("fd index %d is released", p.id);
+        g_hash_table_remove (pinossink->fdids, GINT_TO_POINTER (p.id));
+        break;
+      }
+      case PINOS_PACKET_TYPE_REFRESH_REQUEST:
+      {
+        PinosPacketRefreshRequest p;
+
+        if (!pinos_buffer_iter_parse_refresh_request (&it, &p))
+          continue;
+
+        GST_LOG ("refresh request");
+        gst_pad_push_event (GST_BASE_SINK_PAD (pinossink),
+            gst_video_event_new_upstream_force_key_unit (p.pts,
+            p.request_type == 1, 0));
+        break;
+      }
+      default:
+        break;
+    }
+  }
+}
+
+static void
+on_stream_notify (GObject    *gobject,
+                  GParamSpec *pspec,
+                  gpointer    user_data)
+{
+  PinosStreamState state;
+  PinosStream *stream = PINOS_STREAM (gobject);
+  GstPinosSink *pinossink = user_data;
+
+  state = pinos_stream_get_state (stream);
+  GST_DEBUG ("got stream state %d", state);
+
+  switch (state) {
+    case PINOS_STREAM_STATE_UNCONNECTED:
+    case PINOS_STREAM_STATE_CONNECTING:
+    case PINOS_STREAM_STATE_STARTING:
+    case PINOS_STREAM_STATE_STREAMING:
+    case PINOS_STREAM_STATE_READY:
+      break;
+    case PINOS_STREAM_STATE_ERROR:
+      GST_ELEMENT_ERROR (pinossink, RESOURCE, FAILED,
+          ("stream error: %s",
+            pinos_stream_get_error (stream)->message), (NULL));
+      break;
+  }
+  pinos_main_loop_signal (pinossink->loop, FALSE);
+}
+
+static gboolean
+gst_pinos_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
+{
+  GstPinosSink *pinossink;
+  gchar *str;
+  PinosStreamState state;
+  gboolean res = FALSE;
+
+  pinossink = GST_PINOS_SINK (bsink);
+
+  str = gst_caps_to_string (caps);
+
+  pinos_main_loop_lock (pinossink->loop);
+  state = pinos_stream_get_state (pinossink->stream);
+
+  if (state == PINOS_STREAM_STATE_ERROR)
+    goto start_error;
+
+  if (state == PINOS_STREAM_STATE_STREAMING) {
+    PinosBufferBuilder builder;
+    PinosPacketFormatChange change;
+    PinosBuffer pbuf;
+
+    pinos_buffer_builder_init (&builder);
+
+    change.id = 1;
+    change.format = str;
+    pinos_buffer_builder_add_format_change (&builder, &change);
+    pinos_buffer_builder_end (&builder, &pbuf);
+
+    res = pinos_stream_send_buffer (pinossink->stream, &pbuf);
+    pinos_buffer_clear (&pbuf);
+  } else {
+    GBytes *format = g_bytes_new_take (str, strlen (str) + 1);
+
+    res = pinos_stream_start (pinossink->stream, format, PINOS_STREAM_MODE_BUFFER);
+
+    while (TRUE) {
+      state = pinos_stream_get_state (pinossink->stream);
+
+      if (state == PINOS_STREAM_STATE_STREAMING)
+        break;
+
+      if (state == PINOS_STREAM_STATE_ERROR)
+        goto start_error;
+
+      pinos_main_loop_wait (pinossink->loop);
+    }
+  }
+  pinos_main_loop_unlock (pinossink->loop);
+
+  pinossink->negotiated = res;
+
+  return res;
+
+start_error:
+  {
+    GST_ERROR ("could not start stream");
+    pinos_main_loop_unlock (pinossink->loop);
+    return FALSE;
+  }
+}
+
+static GstFlowReturn
+gst_pinos_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
+{
+  GstPinosSink *pinossink;
+  PinosBuffer pbuf;
+  PinosBufferBuilder builder;
+  GstMemory *mem = NULL;
+  GstClockTime pts, dts, base;
+  PinosPacketHeader hdr;
+  PinosPacketFDPayload p;
+  gsize size;
+  GError *err = NULL;
+  gboolean tmpfile, res;
+
+  pinossink = GST_PINOS_SINK (bsink);
+
+  if (!pinossink->negotiated)
+    goto not_negotiated;
+
+  base = GST_ELEMENT_CAST (bsink)->base_time;
+
+  pts = GST_BUFFER_PTS (buffer);
+  dts = GST_BUFFER_DTS (buffer);
+  if (!GST_CLOCK_TIME_IS_VALID (pts))
+    pts = dts;
+  else if (!GST_CLOCK_TIME_IS_VALID (dts))
+    dts = pts;
+
+  hdr.flags = 0;
+  hdr.seq = GST_BUFFER_OFFSET (buffer);
+  hdr.pts = GST_CLOCK_TIME_IS_VALID (pts) ? pts + base : base;
+  hdr.dts_offset = GST_CLOCK_TIME_IS_VALID (dts) && GST_CLOCK_TIME_IS_VALID (pts) ? pts - dts : 0;
+
+  size = gst_buffer_get_size (buffer);
+
+  if (gst_buffer_n_memory (buffer) == 1
+      && gst_is_fd_memory (gst_buffer_peek_memory (buffer, 0))) {
+    mem = gst_buffer_get_memory (buffer, 0);
+    tmpfile = gst_is_tmpfile_memory (mem);
+  } else {
+    GstMapInfo minfo;
+    GstAllocationParams params = {0, 0, 0, 0, { NULL, }};
+
+    GST_INFO_OBJECT (bsink, "Buffer cannot be payloaded without copying");
+
+    mem = gst_allocator_alloc (pinossink->allocator, size, &params);
+    if (!gst_memory_map (mem, &minfo, GST_MAP_WRITE))
+      goto map_error;
+    gst_buffer_extract (buffer, 0, minfo.data, size);
+    gst_memory_unmap (mem, &minfo);
+    tmpfile = TRUE;
+  }
+
+  pinos_buffer_builder_init (&builder);
+  pinos_buffer_builder_add_header (&builder, &hdr);
+
+  p.fd_index = pinos_buffer_builder_add_fd (&builder, gst_fd_memory_get_fd (mem), &err);
+  if (p.fd_index == -1)
+    goto add_fd_failed;
+  p.id = pinossink->id_counter++;
+  p.offset = 0;
+  p.size = size;
+  pinos_buffer_builder_add_fd_payload (&builder, &p);
+  pinos_buffer_builder_end (&builder, &pbuf);
+
+  gst_memory_unref (mem);
+
+  pinos_main_loop_lock (pinossink->loop);
+  if (pinos_stream_get_state (pinossink->stream) != PINOS_STREAM_STATE_STREAMING)
+    goto streaming_error;
+  res = pinos_stream_send_buffer (pinossink->stream, &pbuf);
+  pinos_buffer_clear (&pbuf);
+  pinos_main_loop_unlock (pinossink->loop);
+
+  if (res && !tmpfile) {
+    /* keep the buffer around until we get the release fd message */
+    g_hash_table_insert (pinossink->fdids, GINT_TO_POINTER (p.id), gst_buffer_ref (buffer));
+  }
+  return GST_FLOW_OK;
+
+not_negotiated:
+  {
+    return GST_FLOW_NOT_NEGOTIATED;
+  }
+map_error:
+  {
+    GST_ELEMENT_ERROR (pinossink, RESOURCE, FAILED,
+        ("failed to map buffer"), (NULL));
+    return GST_FLOW_ERROR;
+  }
+add_fd_failed:
+  {
+    GST_ELEMENT_ERROR (pinossink, RESOURCE, FAILED,
+        ("failed to add fd: %s", err->message), (NULL));
+    pinos_buffer_builder_clear (&builder);
+    return GST_FLOW_ERROR;
+  }
+streaming_error:
+  {
+    pinos_main_loop_unlock (pinossink->loop);
+    return GST_FLOW_ERROR;
+  }
+}
+
+static gboolean
+copy_properties (GQuark field_id,
+                 const GValue *value,
+                 gpointer user_data)
+{
+  PinosProperties *properties = user_data;
+
+  if (G_VALUE_HOLDS_STRING (value))
+    pinos_properties_set (properties,
+                          g_quark_to_string (field_id),
+                          g_value_get_string (value));
+  return TRUE;
+}
+
+static gboolean
+gst_pinos_sink_start (GstBaseSink * basesink)
+{
+  GstPinosSink *pinossink = GST_PINOS_SINK (basesink);
+  PinosProperties *props;
+
+  pinossink->negotiated = FALSE;
+
+  if (pinossink->properties) {
+    props = pinos_properties_new (NULL, NULL);
+    gst_structure_foreach (pinossink->properties, copy_properties, props);
+  } else {
+    props = NULL;
+  }
+
+  pinos_main_loop_lock (pinossink->loop);
+  pinossink->stream = pinos_stream_new (pinossink->ctx, pinossink->client_name, props);
+  g_signal_connect (pinossink->stream, "notify::state", (GCallback) on_stream_notify, pinossink);
+  g_signal_connect (pinossink->stream, "new-buffer", (GCallback) on_new_buffer, pinossink);
+
+  pinos_stream_connect_provide (pinossink->stream, 0, g_bytes_new_static ("ANY", strlen ("ANY")+1));
+
+  while (TRUE) {
+    PinosStreamState state = pinos_stream_get_state (pinossink->stream);
+
+    if (state == PINOS_STREAM_STATE_READY)
+      break;
+
+    if (state == PINOS_STREAM_STATE_ERROR)
+      goto connect_error;
+
+    pinos_main_loop_wait (pinossink->loop);
+  }
+  pinos_main_loop_unlock (pinossink->loop);
+
+  pinossink->negotiated = TRUE;
+
+  return TRUE;
+
+connect_error:
+  {
+    GST_ERROR ("could not connect stream");
+    pinos_main_loop_unlock (pinossink->loop);
+    return FALSE;
+  }
+}
+
+static gboolean
+gst_pinos_sink_stop (GstBaseSink * basesink)
+{
+  GstPinosSink *pinossink = GST_PINOS_SINK (basesink);
+
+  pinos_main_loop_lock (pinossink->loop);
+  if (pinossink->stream) {
+    pinos_stream_stop (pinossink->stream);
+    pinos_stream_disconnect (pinossink->stream);
+    g_clear_object (&pinossink->stream);
+  }
+  pinos_main_loop_unlock (pinossink->loop);
+
+  pinossink->negotiated = FALSE;
+
+  return TRUE;
+}
+
+static void
+on_context_notify (GObject    *gobject,
+                   GParamSpec *pspec,
+                   gpointer    user_data)
+{
+  GstPinosSink *pinossink = user_data;
+  PinosContext *ctx = PINOS_CONTEXT (gobject);
+  PinosContextState state;
+
+  state = pinos_context_get_state (ctx);
+  GST_DEBUG ("got context state %d", state);
+
+  switch (state) {
+    case PINOS_CONTEXT_STATE_UNCONNECTED:
+    case PINOS_CONTEXT_STATE_CONNECTING:
+    case PINOS_CONTEXT_STATE_REGISTERING:
+    case PINOS_CONTEXT_STATE_READY:
+      break;
+    case PINOS_CONTEXT_STATE_ERROR:
+      GST_ELEMENT_ERROR (pinossink, RESOURCE, FAILED,
+          ("context error: %s",
+            pinos_context_get_error (pinossink->ctx)->message), (NULL));
+      break;
+  }
+  pinos_main_loop_signal (pinossink->loop, FALSE);
+}
+
+static gboolean
+gst_pinos_sink_open (GstPinosSink * pinossink)
+{
+  GError *error = NULL;
+
+  pinossink->context = g_main_context_new ();
+  GST_DEBUG ("context %p", pinossink->context);
+
+  pinossink->loop = pinos_main_loop_new (pinossink->context, "pinos-sink-loop");
+  if (!pinos_main_loop_start (pinossink->loop, &error))
+    goto mainloop_error;
+
+  pinos_main_loop_lock (pinossink->loop);
+  pinossink->ctx = pinos_context_new (pinossink->context, g_get_application_name (), NULL);
+  g_signal_connect (pinossink->ctx, "notify::state", (GCallback) on_context_notify, pinossink);
+
+  pinos_context_connect(pinossink->ctx, PINOS_CONTEXT_FLAGS_NONE);
+
+  while (TRUE) {
+    PinosContextState state = pinos_context_get_state (pinossink->ctx);
+
+    if (state == PINOS_CONTEXT_STATE_READY)
+      break;
+
+    if (state == PINOS_CONTEXT_STATE_ERROR)
+      goto connect_error;
+
+    pinos_main_loop_wait (pinossink->loop);
+  }
+  pinos_main_loop_unlock (pinossink->loop);
+
+  return TRUE;
+
+  /* ERRORS */
+mainloop_error:
+  {
+    GST_ELEMENT_ERROR (pinossink, RESOURCE, FAILED,
+        ("Failed to start mainloop: %s", error->message), (NULL));
+    return FALSE;
+  }
+connect_error:
+  {
+    pinos_main_loop_unlock (pinossink->loop);
+    return FALSE;
+  }
+}
+
+static gboolean
+gst_pinos_sink_close (GstPinosSink * pinossink)
+{
+  pinos_main_loop_lock (pinossink->loop);
+  if (pinossink->stream) {
+    pinos_stream_disconnect (pinossink->stream);
+  }
+  if (pinossink->ctx) {
+    pinos_context_disconnect (pinossink->ctx);
+
+    while (TRUE) {
+      PinosContextState state = pinos_context_get_state (pinossink->ctx);
+
+      if (state == PINOS_CONTEXT_STATE_UNCONNECTED)
+        break;
+
+      if (state == PINOS_CONTEXT_STATE_ERROR)
+        break;
+
+      pinos_main_loop_wait (pinossink->loop);
+    }
+  }
+  pinos_main_loop_unlock (pinossink->loop);
+
+  pinos_main_loop_stop (pinossink->loop);
+  g_clear_object (&pinossink->loop);
+  g_clear_object (&pinossink->stream);
+  g_clear_object (&pinossink->ctx);
+  g_main_context_unref (pinossink->context);
+
+  return TRUE;
+}
+
+static GstStateChangeReturn
+gst_pinos_sink_change_state (GstElement * element, GstStateChange transition)
+{
+  GstStateChangeReturn ret;
+  GstPinosSink *this = GST_PINOS_SINK_CAST (element);
+
+  switch (transition) {
+    case GST_STATE_CHANGE_NULL_TO_READY:
+      if (!gst_pinos_sink_open (this))
+        goto open_failed;
+      break;
+    case GST_STATE_CHANGE_READY_TO_PAUSED:
+      break;
+    case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
+      /* uncork and start recording */
+      break;
+    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
+      /* stop recording ASAP by corking */
+      break;
+    default:
+      break;
+  }
+
+  ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
+
+  switch (transition) {
+    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
+      break;
+    case GST_STATE_CHANGE_PAUSED_TO_READY:
+      g_hash_table_remove_all (this->fdids);
+      break;
+    case GST_STATE_CHANGE_READY_TO_NULL:
+      g_hash_table_remove_all (this->fdids);
+      gst_pinos_sink_close (this);
+      break;
+    default:
+      break;
+  }
+  return ret;
+
+  /* ERRORS */
+open_failed:
+  {
+    return GST_STATE_CHANGE_FAILURE;
+  }
+}
diff --git a/pinos/gst/gstpinosprovide.h b/pinos/gst/gstpinosprovide.h
new file mode 100644
index 00000000..c257f426
--- /dev/null
+++ b/pinos/gst/gstpinosprovide.h
@@ -0,0 +1,79 @@
+/* GStreamer
+ * 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 __GST_PINOS_SINK_H__
+#define __GST_PINOS_SINK_H__
+
+#include <gst/gst.h>
+#include <gst/base/gstbasesink.h>
+
+#include <client/pinos.h>
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_PINOS_SINK \
+  (gst_pinos_sink_get_type())
+#define GST_PINOS_SINK(obj) \
+  (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_PINOS_SINK,GstPinosSink))
+#define GST_PINOS_SINK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_PINOS_SINK,GstPinosSinkClass))
+#define GST_IS_PINOS_SINK(obj) \
+  (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_PINOS_SINK))
+#define GST_IS_PINOS_SINK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PINOS_SINK))
+#define GST_PINOS_SINK_CAST(obj) \
+  ((GstPinosSink *) (obj))
+
+typedef struct _GstPinosSink GstPinosSink;
+typedef struct _GstPinosSinkClass GstPinosSinkClass;
+
+/**
+ * GstPinosSink:
+ *
+ * Opaque data structure.
+ */
+struct _GstPinosSink {
+  GstBaseSink element;
+
+  /*< private >*/
+  gchar *client_name;
+
+  /* video state */
+  gboolean negotiated;
+
+  GMainContext *context;
+  PinosMainLoop *loop;
+  PinosContext *ctx;
+  PinosStream *stream;
+  GstAllocator *allocator;
+  GstStructure *properties;
+
+  guint32 id_counter;
+  GHashTable *fdids;
+};
+
+struct _GstPinosSinkClass {
+  GstBaseSinkClass parent_class;
+};
+
+GType gst_pinos_sink_get_type (void);
+
+G_END_DECLS
+
+#endif /* __GST_PINOS_SINK_H__ */
diff --git a/pinos/gst/gstpinossink.c b/pinos/gst/gstpinossink.c
index 3cdbe360..c3d163c8 100644
--- a/pinos/gst/gstpinossink.c
+++ b/pinos/gst/gstpinossink.c
@@ -52,6 +52,7 @@ GST_DEBUG_CATEGORY_STATIC (pinos_sink_debug);
 enum
 {
   PROP_0,
+  PROP_PATH,
   PROP_CLIENT_NAME,
   PROP_STREAM_PROPERTIES
 };
@@ -95,6 +96,7 @@ gst_pinos_sink_finalize (GObject * object)
     gst_structure_free (pinossink->properties);
   g_hash_table_unref (pinossink->fdids);
   g_object_unref (pinossink->allocator);
+  g_free (pinossink->path);
   g_free (pinossink->client_name);
 
   G_OBJECT_CLASS (parent_class)->finalize (object);
@@ -125,6 +127,15 @@ gst_pinos_sink_class_init (GstPinosSinkClass * klass)
   gobject_class->get_property = gst_pinos_sink_get_property;
 
   g_object_class_install_property (gobject_class,
+                                   PROP_PATH,
+                                   g_param_spec_string ("path",
+                                                        "Path",
+                                                        "The sink path to connect to (NULL = default)",
+                                                        NULL,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
                                    PROP_CLIENT_NAME,
                                    g_param_spec_string ("client-name",
                                                         "Client Name",
@@ -222,6 +233,11 @@ gst_pinos_sink_set_property (GObject * object, guint prop_id,
   GstPinosSink *pinossink = GST_PINOS_SINK (object);
 
   switch (prop_id) {
+    case PROP_PATH:
+      g_free (pinossink->path);
+      pinossink->path = g_value_dup_string (value);
+      break;
+
     case PROP_CLIENT_NAME:
       g_free (pinossink->client_name);
       pinossink->client_name = g_value_dup_string (value);
@@ -247,6 +263,10 @@ gst_pinos_sink_get_property (GObject * object, guint prop_id,
   GstPinosSink *pinossink = GST_PINOS_SINK (object);
 
   switch (prop_id) {
+    case PROP_PATH:
+      g_value_set_string (value, pinossink->path);
+      break;
+
     case PROP_CLIENT_NAME:
       g_value_set_string (value, pinossink->client_name);
       break;
@@ -270,6 +290,11 @@ on_new_buffer (GObject    *gobject,
   PinosBufferIter it;
 
   GST_LOG_OBJECT (pinossink, "got new buffer");
+  if (pinossink->stream == NULL) {
+    GST_LOG_OBJECT (pinossink, "no stream");
+    return;
+  }
+
   if (!pinos_stream_peek_buffer (pinossink->stream, &pbuf)) {
     g_warning ("failed to capture buffer");
     return;
@@ -466,6 +491,8 @@ gst_pinos_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
 
   gst_memory_unref (mem);
 
+  GST_LOG ("sending fd index %d", p.id);
+
   pinos_main_loop_lock (pinossink->loop);
   if (pinos_stream_get_state (pinossink->stream) != PINOS_STREAM_STATE_STREAMING)
     goto streaming_error;
@@ -537,7 +564,7 @@ gst_pinos_sink_start (GstBaseSink * basesink)
   g_signal_connect (pinossink->stream, "notify::state", (GCallback) on_stream_notify, pinossink);
   g_signal_connect (pinossink->stream, "new-buffer", (GCallback) on_new_buffer, pinossink);
 
-  pinos_stream_connect_provide (pinossink->stream, 0, g_bytes_new_static ("ANY", strlen ("ANY")+1));
+  pinos_stream_connect_sink (pinossink->stream, pinossink->path, 0, g_bytes_new_static ("ANY", strlen ("ANY")+1));
 
   while (TRUE) {
     PinosStreamState state = pinos_stream_get_state (pinossink->stream);
diff --git a/pinos/gst/gstpinossink.h b/pinos/gst/gstpinossink.h
index c257f426..9f2366f8 100644
--- a/pinos/gst/gstpinossink.h
+++ b/pinos/gst/gstpinossink.h
@@ -52,6 +52,7 @@ struct _GstPinosSink {
   GstBaseSink element;
 
   /*< private >*/
+  gchar *path;
   gchar *client_name;
 
   /* video state */
diff --git a/pinos/gst/gstpinossrc.c b/pinos/gst/gstpinossrc.c
index 271ef184..02b36f8e 100644
--- a/pinos/gst/gstpinossrc.c
+++ b/pinos/gst/gstpinossrc.c
@@ -435,10 +435,11 @@ on_new_buffer (GObject    *gobject,
         break;
     }
   }
-  g_queue_push_tail (&pinossrc->queue, buf);
-
-  pinos_main_loop_signal (pinossrc->loop, FALSE);
+  if (buf) {
+    g_queue_push_tail (&pinossrc->queue, buf);
 
+    pinos_main_loop_signal (pinossrc->loop, FALSE);
+  }
   return;
 
   /* ERRORS */
diff --git a/pinos/modules/gst/gst-manager.c b/pinos/modules/gst/gst-manager.c
index cb7d8676..ffa1869a 100644
--- a/pinos/modules/gst/gst-manager.c
+++ b/pinos/modules/gst/gst-manager.c
@@ -23,6 +23,7 @@
 
 #include "gst-manager.h"
 #include "gst-source.h"
+#include "gst-sink.h"
 
 #define PINOS_GST_MANAGER_GET_PRIVATE(obj)  \
      (G_TYPE_INSTANCE_GET_PRIVATE ((obj), PINOS_TYPE_GST_MANAGER, PinosGstManagerPrivate))
@@ -64,7 +65,7 @@ device_added (PinosGstManager *manager,
   PinosGstManagerPrivate *priv = manager->priv;
   gchar *name, *klass;
   GstElement *element;
-  PinosSource *source;
+  PinosNode *node;
   GstStructure *p;
   PinosProperties *properties;
   GstCaps *caps;
@@ -90,13 +91,24 @@ device_added (PinosGstManager *manager,
                         "gstreamer.device.class",
                         klass);
 
+  node = pinos_node_new (priv->daemon);
+  g_object_set_data (G_OBJECT (device), "PinosNode", node);
+
   element = gst_device_create_element (device, NULL);
-  source = pinos_gst_source_new (priv->daemon,
-                                 name,
-                                 properties,
-                                 element,
-                                 caps);
-  g_object_set_data (G_OBJECT (device), "PinosSource", source);
+
+  if (strstr (klass, "Source")) {
+    pinos_gst_source_new (node,
+                          name,
+                          properties,
+                          element,
+                          caps);
+  } else if (strstr (klass, "Sink")) {
+    pinos_gst_sink_new (node,
+                        name,
+                        properties,
+                        element,
+                        caps);
+  }
 
   pinos_properties_free (properties);
   gst_caps_unref (caps);
@@ -109,7 +121,7 @@ device_removed (PinosGstManager *manager,
                 GstDevice       *device)
 {
   gchar *name;
-  PinosSource *source;
+  PinosNode *node;
 
   name = gst_device_get_display_name (device);
   if (strcmp (name, "gst") == 0)
@@ -117,8 +129,8 @@ device_removed (PinosGstManager *manager,
 
   g_print("Device removed: %s\n", name);
 
-  source = g_object_steal_data (G_OBJECT (device), "PinosSource");
-  g_object_unref (source);
+  node = g_object_steal_data (G_OBJECT (device), "PinosNode");
+  g_object_unref (node);
   g_free (name);
 }
 
@@ -184,6 +196,7 @@ start_monitor (PinosGstManager *manager)
 
   gst_device_monitor_add_filter (priv->monitor, "Video/Source", NULL);
   gst_device_monitor_add_filter (priv->monitor, "Audio/Source", NULL);
+  gst_device_monitor_add_filter (priv->monitor, "Audio/Sink", NULL);
   gst_device_monitor_start (priv->monitor);
 
   providers = gst_device_monitor_get_providers (priv->monitor);
diff --git a/pinos/modules/gst/gst-sink.c b/pinos/modules/gst/gst-sink.c
new file mode 100644
index 00000000..b5e58ced
--- /dev/null
+++ b/pinos/modules/gst/gst-sink.c
@@ -0,0 +1,568 @@
+/* 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 <gst/gst.h>
+#include <gio/gio.h>
+
+#include <gst/net/net.h>
+
+#include "gst-sink.h"
+
+#define PINOS_GST_SINK_GET_PRIVATE(obj)  \
+     (G_TYPE_INSTANCE_GET_PRIVATE ((obj), PINOS_TYPE_GST_SINK, PinosGstSinkPrivate))
+
+struct _PinosGstSinkPrivate
+{
+  GstElement *pipeline;
+  GstElement *src;
+  GstElement *depay;
+  GstElement *element;
+
+  GstCaps *possible_formats;
+
+  GstNetTimeProvider *provider;
+
+  PinosProperties *props;
+
+  gint n_channels;
+};
+
+enum {
+  PROP_0,
+  PROP_ELEMENT,
+  PROP_POSSIBLE_FORMATS
+};
+
+G_DEFINE_TYPE (PinosGstSink, pinos_gst_sink, PINOS_TYPE_SINK);
+
+static gboolean
+bus_handler (GstBus     *bus,
+             GstMessage *message,
+             gpointer    user_data)
+{
+  PinosSink *sink = user_data;
+  PinosGstSinkPrivate *priv = PINOS_GST_SINK (sink)->priv;
+
+  switch (GST_MESSAGE_TYPE (message)) {
+    case GST_MESSAGE_ERROR:
+    {
+      GError *error;
+      gchar *debug;
+
+      gst_message_parse_error (message, &error, &debug);
+      g_warning ("got error %s (%s)\n", error->message, debug);
+      g_free (debug);
+
+      pinos_sink_report_error (sink, error);
+      gst_element_set_state (priv->pipeline, GST_STATE_NULL);
+      break;
+    }
+    case GST_MESSAGE_NEW_CLOCK:
+    {
+      GstClock *clock;
+      PinosProperties *props;
+
+      gst_message_parse_new_clock (message, &clock);
+      GST_INFO ("got new clock %s", GST_OBJECT_NAME (clock));
+
+      g_object_get (sink, "properties", &props, NULL);
+      pinos_properties_set (props, "gst.pipeline.clock", GST_OBJECT_NAME (clock));
+      g_object_set (sink, "properties", props, NULL);
+      pinos_properties_free (props);
+      break;
+    }
+    case GST_MESSAGE_CLOCK_LOST:
+    {
+      GstClock *clock;
+      PinosProperties *props;
+
+      gst_message_parse_new_clock (message, &clock);
+      GST_INFO ("clock lost %s", GST_OBJECT_NAME (clock));
+
+      g_object_get (sink, "properties", &props, NULL);
+      pinos_properties_remove (props, "gst.pipeline.clock");
+      g_object_set (sink, "properties", props, NULL);
+      pinos_properties_free (props);
+
+      gst_element_set_state (priv->pipeline, GST_STATE_PAUSED);
+      gst_element_set_state (priv->pipeline, GST_STATE_PLAYING);
+      break;
+    }
+    default:
+      break;
+  }
+  return TRUE;
+}
+
+static gboolean
+setup_pipeline (PinosGstSink *sink, GError **error)
+{
+  PinosGstSinkPrivate *priv = sink->priv;
+  GstBus *bus;
+  GstCaps *caps;
+
+  g_debug ("gst-sink %p: setup pipeline", sink);
+  priv->pipeline = gst_pipeline_new (NULL);
+
+  priv->src = gst_element_factory_make ("socketsrc", NULL);
+  caps = gst_caps_new_empty_simple ("application/x-pinos");
+  g_object_set (priv->src, "send-messages", TRUE,
+                           "caps", caps, NULL);
+  gst_caps_unref (caps);
+  gst_bin_add (GST_BIN (priv->pipeline), priv->src);
+
+  priv->depay = gst_element_factory_make ("pinosdepay", NULL);
+  gst_bin_add (GST_BIN (priv->pipeline), priv->depay);
+  gst_element_link (priv->src, priv->depay);
+
+  gst_bin_add (GST_BIN (priv->pipeline), priv->element);
+  gst_element_link (priv->depay, priv->element);
+
+  bus = gst_pipeline_get_bus (GST_PIPELINE (priv->pipeline));
+  gst_bus_add_watch (bus, bus_handler, sink);
+  gst_object_unref (bus);
+
+  return TRUE;
+}
+
+static gboolean
+start_pipeline (PinosGstSink *sink, GError **error)
+{
+  PinosGstSinkPrivate *priv = sink->priv;
+  GstCaps *caps;
+  GstQuery *query;
+  GstStateChangeReturn ret;
+  gchar *str;
+
+  g_debug ("gst-sink %p: starting pipeline", sink);
+
+  ret = gst_element_set_state (priv->pipeline, GST_STATE_READY);
+  if (ret == GST_STATE_CHANGE_FAILURE)
+    goto ready_failed;
+
+  query = gst_query_new_caps (NULL);
+  if (gst_element_query (priv->element, query)) {
+    gst_query_parse_caps_result (query, &caps);
+    gst_caps_replace (&priv->possible_formats, caps);
+    str = gst_caps_to_string (caps);
+    g_debug ("gst-sink %p: updated possible formats %s", sink, str);
+    g_free (str);
+  }
+  gst_query_unref (query);
+
+  return TRUE;
+
+  /* ERRORS */
+ready_failed:
+  {
+    GST_ERROR_OBJECT (sink, "failed state change to READY");
+    gst_element_set_state (priv->pipeline, GST_STATE_NULL);
+    if (error)
+      *error = g_error_new (G_IO_ERROR,
+                            G_IO_ERROR_FAILED,
+                            "Failed to start pipeline");
+    return FALSE;
+  }
+}
+
+static void
+stop_pipeline (PinosGstSink *sink)
+{
+  PinosGstSinkPrivate *priv = sink->priv;
+
+  g_debug ("gst-sink %p: stopping pipeline", sink);
+  gst_element_set_state (priv->pipeline, GST_STATE_NULL);
+  g_clear_object (&priv->provider);
+}
+
+static void
+destroy_pipeline (PinosGstSink *sink)
+{
+  PinosGstSinkPrivate *priv = sink->priv;
+
+  g_debug ("gst-sink %p: destroy pipeline", sink);
+  stop_pipeline (sink);
+  g_clear_object (&priv->pipeline);
+}
+
+
+static gboolean
+set_state (PinosSink      *sink,
+           PinosSinkState  state)
+{
+  PinosGstSinkPrivate *priv = PINOS_GST_SINK (sink)->priv;
+
+  g_debug ("gst-sink %p: set state %d", sink, state);
+
+  switch (state) {
+    case PINOS_SINK_STATE_SUSPENDED:
+      gst_element_set_state (priv->pipeline, GST_STATE_NULL);
+      break;
+
+    case PINOS_SINK_STATE_INITIALIZING:
+      gst_element_set_state (priv->pipeline, GST_STATE_READY);
+      break;
+
+    case PINOS_SINK_STATE_IDLE:
+      gst_element_set_state (priv->pipeline, GST_STATE_PAUSED);
+      break;
+
+    case PINOS_SINK_STATE_RUNNING:
+      gst_element_set_state (priv->pipeline, GST_STATE_PLAYING);
+      break;
+
+    case PINOS_SINK_STATE_ERROR:
+      break;
+  }
+  pinos_sink_update_state (sink, state);
+  return TRUE;
+}
+
+static GBytes *
+get_formats (PinosSink   *sink,
+             GBytes      *filter,
+             GError     **error)
+{
+  PinosGstSinkPrivate *priv = PINOS_GST_SINK (sink)->priv;
+  GstCaps *caps, *cfilter;
+  gchar *str;
+
+  if (filter) {
+    cfilter = gst_caps_from_string (g_bytes_get_data (filter, NULL));
+    if (cfilter == NULL)
+      goto invalid_filter;
+
+    caps = gst_caps_intersect (priv->possible_formats, cfilter);
+    gst_caps_unref (cfilter);
+
+    if (caps == NULL)
+      goto no_formats;
+
+  } else {
+    caps = gst_caps_ref (priv->possible_formats);
+  }
+  g_object_get (priv->depay, "caps", &cfilter, NULL);
+  if (cfilter != NULL) {
+    GstCaps *t = caps;
+
+    caps = gst_caps_intersect (t, cfilter);
+    gst_caps_unref (cfilter);
+    gst_caps_unref (t);
+  }
+  if (gst_caps_is_empty (caps)) {
+    gst_caps_unref (caps);
+    goto no_formats;
+  }
+
+  str = gst_caps_to_string (caps);
+  gst_caps_unref (caps);
+
+  return g_bytes_new_take (str, strlen (str) + 1);
+
+invalid_filter:
+  {
+    if (error)
+      *error = g_error_new (G_IO_ERROR,
+                            G_IO_ERROR_INVALID_ARGUMENT,
+                            "Invalid filter received");
+    return NULL;
+  }
+no_formats:
+  {
+    if (error)
+      *error = g_error_new (G_IO_ERROR,
+                            G_IO_ERROR_NOT_FOUND,
+                            "No compatible format found");
+    return NULL;
+  }
+}
+
+static void
+on_socket_notify (GObject    *gobject,
+                  GParamSpec *pspec,
+                  gpointer    user_data)
+{
+  PinosGstSink *sink = user_data;
+  PinosGstSinkPrivate *priv = sink->priv;
+  GSocket *socket;
+  guint num_handles;
+  GstCaps *caps;
+  GBytes *requested_format, *format = NULL;
+  gchar *str;
+  gpointer state = NULL;
+  const gchar *key, *val;
+  PinosProperties *props;
+
+  g_object_get (gobject, "socket", &socket, NULL);
+  GST_DEBUG ("got socket %p", socket);
+
+  if (socket == NULL) {
+    g_object_set (priv->src, "socket", NULL, NULL);
+    num_handles = 0;
+  } else {
+    g_object_set (priv->src, "socket", socket, NULL);
+    num_handles = 1;
+  }
+
+  if (num_handles == 0) {
+    pinos_sink_report_idle (PINOS_SINK (sink));
+    g_object_set (priv->depay, "caps", NULL, NULL);
+
+    str = gst_caps_to_string (priv->possible_formats);
+    format = g_bytes_new_take (str, strlen (str) + 1);
+  } else if (socket) {
+    /* what client requested */
+    g_object_get (gobject, "requested-format", &requested_format, NULL);
+    g_assert (requested_format != NULL);
+
+    if (num_handles == 1) {
+      /* first client, we set the requested format as the format */
+      format = requested_format;
+
+      /* set on the filter */
+      caps = gst_caps_from_string (g_bytes_get_data (format, NULL));
+      g_assert (caps != NULL);
+      g_object_set (priv->depay, "caps", caps, NULL);
+      gst_caps_unref (caps);
+    } else {
+      /* we already have a client, format is whatever is configured already */
+      g_bytes_unref (requested_format);
+
+      g_object_get (priv->depay, "caps", &caps, NULL);
+      str = gst_caps_to_string (caps);
+      format = g_bytes_new_take (str, strlen (str) + 1);
+      gst_caps_unref (caps);
+    }
+    /* this is what we use as the final format for the output */
+    g_object_set (gobject, "format", format, NULL);
+    pinos_sink_report_busy (PINOS_SINK (sink));
+    g_object_unref (socket);
+  }
+  if (format) {
+    pinos_sink_update_possible_formats (PINOS_SINK (sink), format);
+    g_bytes_unref (format);
+  }
+
+  g_object_get (gobject, "properties", &props, NULL);
+  while ((key = pinos_properties_iterate (priv->props, &state))) {
+    val = pinos_properties_get (priv->props, key);
+    pinos_properties_set (props, key, val);
+  }
+  g_object_set (gobject, "properties", props, NULL);
+  pinos_properties_free (props);
+}
+
+static PinosChannel *
+create_channel (PinosSink       *sink,
+                const gchar     *client_path,
+                GBytes          *format_filter,
+                PinosProperties *props,
+                const gchar     *prefix,
+                GError          **error)
+{
+  PinosGstSink *s = PINOS_GST_SINK (sink);
+  PinosGstSinkPrivate *priv = s->priv;
+  PinosChannel *channel;
+  gpointer state = NULL;
+  const gchar *key, *val;
+
+  if (priv->n_channels == 0) {
+    if (!start_pipeline (s, error))
+      return NULL;
+  }
+
+  while ((key = pinos_properties_iterate (priv->props, &state))) {
+    val = pinos_properties_get (priv->props, key);
+    pinos_properties_set (props, key, val);
+  }
+
+  channel = PINOS_SINK_CLASS (pinos_gst_sink_parent_class)
+                ->create_channel (sink,
+                                  client_path,
+                                  format_filter,
+                                  props,
+                                  prefix,
+                                  error);
+  if (channel == NULL)
+    goto no_channel;
+
+  g_signal_connect (channel,
+                    "notify::socket",
+                    (GCallback) on_socket_notify,
+                    sink);
+
+  priv->n_channels++;
+
+  return channel;
+
+  /* ERRORS */
+no_channel:
+  {
+    if (priv->n_channels == 0)
+       stop_pipeline (s);
+    return NULL;
+  }
+}
+
+static gboolean
+release_channel (PinosSink      *sink,
+                 PinosChannel   *channel)
+{
+  return PINOS_SINK_CLASS (pinos_gst_sink_parent_class)
+                ->release_channel (sink, channel);
+}
+
+static void
+get_property (GObject    *object,
+              guint       prop_id,
+              GValue     *value,
+              GParamSpec *pspec)
+{
+  PinosGstSink *sink = PINOS_GST_SINK (object);
+  PinosGstSinkPrivate *priv = sink->priv;
+
+  switch (prop_id) {
+    case PROP_ELEMENT:
+      g_value_set_object (value, priv->element);
+      break;
+
+    case PROP_POSSIBLE_FORMATS:
+      g_value_set_boxed (value, priv->possible_formats);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+set_property (GObject      *object,
+              guint         prop_id,
+              const GValue *value,
+              GParamSpec   *pspec)
+{
+  PinosGstSink *sink = PINOS_GST_SINK (object);
+  PinosGstSinkPrivate *priv = sink->priv;
+
+  switch (prop_id) {
+    case PROP_ELEMENT:
+      priv->element = g_value_dup_object (value);
+      break;
+
+    case PROP_POSSIBLE_FORMATS:
+      priv->possible_formats = g_value_dup_boxed (value);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+sink_constructed (GObject * object)
+{
+  PinosGstSink *sink = PINOS_GST_SINK (object);
+
+  setup_pipeline (sink, NULL);
+
+  G_OBJECT_CLASS (pinos_gst_sink_parent_class)->constructed (object);
+}
+
+static void
+sink_finalize (GObject * object)
+{
+  PinosGstSink *sink = PINOS_GST_SINK (object);
+  PinosGstSinkPrivate *priv = sink->priv;
+
+  destroy_pipeline (sink);
+  g_clear_pointer (&priv->possible_formats, gst_caps_unref);
+  pinos_properties_free (priv->props);
+
+  G_OBJECT_CLASS (pinos_gst_sink_parent_class)->finalize (object);
+}
+
+static void
+pinos_gst_sink_class_init (PinosGstSinkClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+  PinosSinkClass *sink_class = PINOS_SINK_CLASS (klass);
+
+  g_type_class_add_private (klass, sizeof (PinosGstSinkPrivate));
+
+  gobject_class->constructed = sink_constructed;
+  gobject_class->finalize = sink_finalize;
+  gobject_class->get_property = get_property;
+  gobject_class->set_property = set_property;
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_ELEMENT,
+                                   g_param_spec_object ("element",
+                                                        "Element",
+                                                        "The element",
+                                                        GST_TYPE_ELEMENT,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_CONSTRUCT_ONLY |
+                                                        G_PARAM_STATIC_STRINGS));
+  g_object_class_install_property (gobject_class,
+                                   PROP_POSSIBLE_FORMATS,
+                                   g_param_spec_boxed ("possible-formats",
+                                                       "Possible Formats",
+                                                       "The possible formats",
+                                                       GST_TYPE_CAPS,
+                                                       G_PARAM_READWRITE |
+                                                       G_PARAM_CONSTRUCT_ONLY |
+                                                       G_PARAM_STATIC_STRINGS));
+
+  sink_class->get_formats = get_formats;
+  sink_class->set_state = set_state;
+  sink_class->create_channel = create_channel;
+  sink_class->release_channel = release_channel;
+}
+
+static void
+pinos_gst_sink_init (PinosGstSink * sink)
+{
+  PinosGstSinkPrivate *priv;
+
+  priv = sink->priv = PINOS_GST_SINK_GET_PRIVATE (sink);
+  priv->props = pinos_properties_new (NULL, NULL);
+}
+
+PinosSink *
+pinos_gst_sink_new (PinosNode   *node,
+                    const gchar *name,
+                    PinosProperties *properties,
+                    GstElement  *element,
+                    GstCaps     *caps)
+{
+  PinosSink *sink;
+
+  sink = g_object_new (PINOS_TYPE_GST_SINK,
+                       "node", node,
+                       "name", name,
+                       "properties", properties,
+                       "element", element,
+                       "possible-formats", caps,
+                       NULL);
+
+  return sink;
+}
diff --git a/pinos/modules/gst/gst-sink.h b/pinos/modules/gst/gst-sink.h
new file mode 100644
index 00000000..72b4d7ce
--- /dev/null
+++ b/pinos/modules/gst/gst-sink.h
@@ -0,0 +1,64 @@
+/* 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_GST_SINK_H__
+#define __PINOS_GST_SINK_H__
+
+#include <glib-object.h>
+
+#include <client/pinos.h>
+#include <server/node.h>
+#include <server/sink.h>
+
+G_BEGIN_DECLS
+
+#define PINOS_TYPE_GST_SINK                 (pinos_gst_sink_get_type ())
+#define PINOS_IS_GST_SINK(obj)              (G_TYPE_CHECK_INSTANCE_TYPE ((obj), PINOS_TYPE_GST_SINK))
+#define PINOS_IS_GST_SINK_CLASS(klass)      (G_TYPE_CHECK_CLASS_TYPE ((klass), PINOS_TYPE_GST_SINK))
+#define PINOS_GST_SINK_GET_CLASS(obj)       (G_TYPE_INSTANCE_GET_CLASS ((obj), PINOS_TYPE_GST_SINK, PinosGstSinkClass))
+#define PINOS_GST_SINK(obj)                 (G_TYPE_CHECK_INSTANCE_CAST ((obj), PINOS_TYPE_GST_SINK, PinosGstSink))
+#define PINOS_GST_SINK_CLASS(klass)         (G_TYPE_CHECK_CLASS_CAST ((klass), PINOS_TYPE_GST_SINK, PinosGstSinkClass))
+#define PINOS_GST_SINK_CAST(obj)            ((PinosGstSink*)(obj))
+#define PINOS_GST_SINK_CLASS_CAST(klass)    ((PinosGstSinkClass*)(klass))
+
+typedef struct _PinosGstSink PinosGstSink;
+typedef struct _PinosGstSinkClass PinosGstSinkClass;
+typedef struct _PinosGstSinkPrivate PinosGstSinkPrivate;
+
+struct _PinosGstSink {
+  PinosSink object;
+
+  PinosGstSinkPrivate *priv;
+};
+
+struct _PinosGstSinkClass {
+  PinosSinkClass parent_class;
+};
+
+GType         pinos_gst_sink_get_type        (void);
+
+PinosSink *   pinos_gst_sink_new             (PinosNode       *node,
+                                              const gchar     *name,
+                                              PinosProperties *properties,
+                                              GstElement      *element,
+                                              GstCaps         *caps);
+
+G_END_DECLS
+
+#endif /* __PINOS_GST_SINK_H__ */
diff --git a/pinos/modules/gst/gst-source.c b/pinos/modules/gst/gst-source.c
index d67e525e..dea985f6 100644
--- a/pinos/modules/gst/gst-source.c
+++ b/pinos/modules/gst/gst-source.c
@@ -599,7 +599,7 @@ pinos_gst_source_init (PinosGstSource * source)
 }
 
 PinosSource *
-pinos_gst_source_new (PinosDaemon *daemon,
+pinos_gst_source_new (PinosNode   *node,
                       const gchar *name,
                       PinosProperties *properties,
                       GstElement  *element,
@@ -608,7 +608,7 @@ pinos_gst_source_new (PinosDaemon *daemon,
   PinosSource *source;
 
   source = g_object_new (PINOS_TYPE_GST_SOURCE,
-                         "daemon", daemon,
+                         "node", node,
                          "name", name,
                          "properties", properties,
                          "element", element,
diff --git a/pinos/modules/gst/gst-source.h b/pinos/modules/gst/gst-source.h
index 4c6ac92e..db1ae8a0 100644
--- a/pinos/modules/gst/gst-source.h
+++ b/pinos/modules/gst/gst-source.h
@@ -23,7 +23,7 @@
 #include <glib-object.h>
 
 #include <client/pinos.h>
-#include <server/daemon.h>
+#include <server/node.h>
 #include <server/source.h>
 
 G_BEGIN_DECLS
@@ -53,7 +53,7 @@ struct _PinosGstSourceClass {
 
 GType           pinos_gst_source_get_type        (void);
 
-PinosSource *   pinos_gst_source_new             (PinosDaemon     *daemon,
+PinosSource *   pinos_gst_source_new             (PinosNode       *node,
                                                   const gchar     *name,
                                                   PinosProperties *properties,
                                                   GstElement      *element,
diff --git a/pinos/server/channel.c b/pinos/server/channel.c
index fd5e03df..aa2ef92e 100644
--- a/pinos/server/channel.c
+++ b/pinos/server/channel.c
@@ -237,7 +237,7 @@ handle_start (PinosChannel1          *interface,
 
   socketpair (AF_UNIX, SOCK_STREAM, 0, fd);
 
-  g_debug ("channel %p: handle start, fd[%d,%d]", channel, fd[0], fd[1]);
+  g_debug ("channel %p: handle start, fd[%d,%d], format %s", channel, fd[0], fd[1], arg_requested_format);
 
   g_clear_object (&priv->socket);
   priv->socket = g_socket_new_from_fd (fd[0], NULL);
diff --git a/pinos/server/client.c b/pinos/server/client.c
index 80a26f06..01a9bdc1 100644
--- a/pinos/server/client.c
+++ b/pinos/server/client.c
@@ -148,6 +148,7 @@ handle_create_source_channel (PinosClient1           *interface,
 {
   PinosClient *client = user_data;
   PinosClientPrivate *priv = client->priv;
+  PinosNode *node;
   PinosSource *source;
   PinosChannel *channel;
   const gchar *object_path, *sender;
@@ -162,11 +163,15 @@ handle_create_source_channel (PinosClient1           *interface,
   formats = g_bytes_new (arg_accepted_formats, strlen (arg_accepted_formats) + 1);
   props = pinos_properties_from_variant (arg_properties);
 
-  source = pinos_daemon_find_source (priv->daemon,
-                                     arg_source,
-                                     props,
-                                     formats,
-                                     &error);
+  node = pinos_daemon_find_node (priv->daemon,
+                                 arg_source,
+                                 props,
+                                 formats,
+                                 &error);
+  if (node == NULL)
+    goto no_node;
+
+  source = pinos_node_get_source (node);
   if (source == NULL)
     goto no_source;
 
@@ -190,7 +195,7 @@ handle_create_source_channel (PinosClient1           *interface,
                     client);
 
   object_path = pinos_channel_get_object_path (channel);
-  g_debug ("client %p: add channel %p, %s", client, channel, object_path);
+  g_debug ("client %p: add source channel %p, %s", client, channel, object_path);
   g_dbus_method_invocation_return_value (invocation,
                                          g_variant_new ("(o)", object_path));
 
@@ -203,18 +208,124 @@ not_allowed:
                  "org.pinos.Error", "not client owner");
     return TRUE;
   }
+no_node:
+  {
+    g_debug ("client %p: could not find node %s, %s", client, arg_source, error->message);
+    g_dbus_method_invocation_return_gerror (invocation, error);
+    pinos_properties_free (props);
+    g_bytes_unref (formats);
+    g_clear_error (&error);
+    return TRUE;
+  }
 no_source:
   {
-    g_debug ("client %p: could not find source %s, %s", client, arg_source, error->message);
+    g_debug ("client %p: node %s is not a source", client, arg_source);
+    g_dbus_method_invocation_return_dbus_error (invocation,
+                 "org.pinos.Error", "not node is not a source");
+    pinos_properties_free (props);
+    g_bytes_unref (formats);
+    return TRUE;
+  }
+no_channel:
+  {
+    g_debug ("client %p: could not create source channel %s", client, error->message);
+    g_dbus_method_invocation_return_gerror (invocation, error);
+    g_clear_error (&error);
+    return TRUE;
+  }
+}
+
+static gboolean
+handle_create_sink_channel (PinosClient1           *interface,
+                            GDBusMethodInvocation  *invocation,
+                            const gchar            *arg_sink,
+                            const gchar            *arg_accepted_formats,
+                            GVariant               *arg_properties,
+                            gpointer                user_data)
+{
+  PinosClient *client = user_data;
+  PinosClientPrivate *priv = client->priv;
+  PinosNode *node;
+  PinosSink *sink;
+  PinosChannel *channel;
+  const gchar *object_path, *sender;
+  GBytes *formats;
+  PinosProperties *props;
+  GError *error = NULL;
+
+  sender = g_dbus_method_invocation_get_sender (invocation);
+  if (g_strcmp0 (pinos_client_get_sender (client), sender) != 0)
+    goto not_allowed;
+
+  formats = g_bytes_new (arg_accepted_formats, strlen (arg_accepted_formats) + 1);
+  props = pinos_properties_from_variant (arg_properties);
+
+  node = pinos_daemon_find_node (priv->daemon,
+                                 arg_sink,
+                                 props,
+                                 formats,
+                                 &error);
+  if (node == NULL)
+    goto no_node;
+
+  sink = pinos_node_get_sink (node);
+  if (sink == NULL)
+    goto no_sink;
+
+  channel = pinos_sink_create_channel (sink,
+                                       priv->object_path,
+                                       formats,
+                                       props,
+                                       priv->object_path,
+                                       &error);
+  pinos_properties_free (props);
+  g_bytes_unref (formats);
+
+  if (channel == NULL)
+    goto no_channel;
+
+  priv->channels = g_list_prepend (priv->channels, channel);
+
+  g_signal_connect (channel,
+                    "remove",
+                    (GCallback) handle_remove_channel,
+                    client);
+
+  object_path = pinos_channel_get_object_path (channel);
+  g_debug ("client %p: add sink channel %p, %s", client, channel, object_path);
+  g_dbus_method_invocation_return_value (invocation,
+                                         g_variant_new ("(o)", object_path));
+
+  return TRUE;
+
+  /* ERRORS */
+not_allowed:
+  {
+    g_dbus_method_invocation_return_dbus_error (invocation,
+                 "org.pinos.Error", "not client owner");
+    return TRUE;
+  }
+no_node:
+  {
+    g_debug ("client %p: could not find node %s, %s", client, arg_sink, error->message);
     g_dbus_method_invocation_return_gerror (invocation, error);
     pinos_properties_free (props);
     g_bytes_unref (formats);
     g_clear_error (&error);
     return TRUE;
   }
+no_sink:
+  {
+    g_debug ("client %p: node %s is not a sink", client, arg_sink);
+    g_dbus_method_invocation_return_dbus_error (invocation,
+                 "org.pinos.Error", "node is not a sink");
+    pinos_properties_free (props);
+    g_bytes_unref (formats);
+    return TRUE;
+  }
 no_channel:
   {
-    g_debug ("client %p: could not channel %s", client, error->message);
+    g_debug ("client %p: could not create sink channel %s", client, error->message);
     g_dbus_method_invocation_return_gerror (invocation, error);
     g_clear_error (&error);
     return TRUE;
@@ -298,7 +409,7 @@ no_source:
   }
 no_channel:
   {
-    g_debug ("client %p: could not create channel %s", client, error->message);
+    g_debug ("client %p: could not create upload channel %s", client, error->message);
     g_dbus_method_invocation_return_gerror (invocation, error);
     g_object_unref (source);
     g_clear_error (&error);
@@ -340,6 +451,9 @@ client_register_object (PinosClient *client,
   g_signal_connect (priv->client1, "handle-create-source-channel",
                                    (GCallback) handle_create_source_channel,
                                    client);
+  g_signal_connect (priv->client1, "handle-create-sink-channel",
+                                   (GCallback) handle_create_sink_channel,
+                                   client);
   g_signal_connect (priv->client1, "handle-create-upload-channel",
                                    (GCallback) handle_create_upload_channel,
                                    client);
diff --git a/pinos/server/daemon.c b/pinos/server/daemon.c
index ec092de2..48188cfd 100644
--- a/pinos/server/daemon.c
+++ b/pinos/server/daemon.c
@@ -37,7 +37,7 @@ struct _PinosDaemonPrivate
   GDBusConnection *connection;
   GDBusObjectManagerServer *server_manager;
 
-  GList *sources;
+  GList *nodes;
 
   GHashTable *senders;
 
@@ -340,87 +340,87 @@ pinos_daemon_unexport (PinosDaemon *daemon,
 }
 
 /**
- * pinos_daemon_add_source:
+ * pinos_daemon_add_node:
  * @daemon: a #PinosDaemon
- * @source: a #PinosSource
+ * @node: a #PinosNode
  *
- * Add @source to @daemon.
+ * Add @node to @daemon.
  */
 void
-pinos_daemon_add_source (PinosDaemon *daemon,
-                         PinosSource *source)
+pinos_daemon_add_node (PinosDaemon *daemon,
+                       PinosNode *node)
 {
   PinosDaemonPrivate *priv;
 
   g_return_if_fail (PINOS_IS_DAEMON (daemon));
-  g_return_if_fail (PINOS_IS_SOURCE (source));
+  g_return_if_fail (PINOS_IS_NODE (node));
   priv = daemon->priv;
 
-  priv->sources = g_list_prepend (priv->sources, source);
+  priv->nodes = g_list_prepend (priv->nodes, node);
 }
 
 /**
- * pinos_daemon_remove_source:
+ * pinos_daemon_remove_node:
  * @daemon: a #PinosDaemon
- * @source: a #PinosSource
+ * @node: a #PinosNode
  *
- * Remove @source from @daemon.
+ * Remove @node from @daemon.
  */
 void
-pinos_daemon_remove_source (PinosDaemon *daemon,
-                            PinosSource *source)
+pinos_daemon_remove_node (PinosDaemon *daemon,
+                          PinosNode *node)
 {
   PinosDaemonPrivate *priv;
 
   g_return_if_fail (PINOS_IS_DAEMON (daemon));
-  g_return_if_fail (PINOS_IS_SOURCE (source));
+  g_return_if_fail (PINOS_IS_NODE (node));
   priv = daemon->priv;
 
-  priv->sources = g_list_remove (priv->sources, source);
+  priv->nodes = g_list_remove (priv->nodes, node);
 }
 
 /**
- * pinos_daemon_find_source:
+ * pinos_daemon_find_node:
  * @daemon: a #PinosDaemon
- * @name: a source name
- * @props: source properties
+ * @name: a node name
+ * @props: node properties
  * @format_filter: a format filter
  * @error: location for an error
  *
- * Find the best source in @daemon that matches the given parameters.
+ * Find the best node in @daemon that matches the given parameters.
  *
- * Returns: a #PinosSource or %NULL when no source could be found.
+ * Returns: a #PinosNode or %NULL when no node could be found.
  */
-PinosSource *
-pinos_daemon_find_source (PinosDaemon     *daemon,
-                          const gchar     *name,
-                          PinosProperties *props,
-                          GBytes          *format_filter,
-                          GError         **error)
+PinosNode *
+pinos_daemon_find_node (PinosDaemon     *daemon,
+                        const gchar     *name,
+                        PinosProperties *props,
+                        GBytes          *format_filter,
+                        GError         **error)
 {
   PinosDaemonPrivate *priv;
-  PinosSource *best = NULL;
+  PinosNode *best = NULL;
   GList *walk;
 
   g_return_val_if_fail (PINOS_IS_DAEMON (daemon), NULL);
   priv = daemon->priv;
 
-  for (walk = priv->sources; walk; walk = g_list_next (walk)) {
-    PinosSource *s = walk->data;
+  for (walk = priv->nodes; walk; walk = g_list_next (walk)) {
+    PinosNode *n = walk->data;
 
     if (name == NULL) {
-      best = s;
+      best = n;
       break;
     }
-    else if (g_str_has_suffix (pinos_source_get_object_path (s), name))
-      best = s;
+    else if (g_str_has_suffix (pinos_node_get_object_path (n), name))
+      best = n;
   }
 
   if (best == NULL) {
     if (error)
       *error = g_error_new (G_IO_ERROR,
                             G_IO_ERROR_NOT_FOUND,
-                            "Source not found");
+                            "Node not found");
   }
   return best;
 }
diff --git a/pinos/server/daemon.h b/pinos/server/daemon.h
index 07879ecb..7be3e5bc 100644
--- a/pinos/server/daemon.h
+++ b/pinos/server/daemon.h
@@ -37,7 +37,7 @@ typedef struct _PinosDaemon PinosDaemon;
 typedef struct _PinosDaemonClass PinosDaemonClass;
 typedef struct _PinosDaemonPrivate PinosDaemonPrivate;
 
-#include <pinos/server/source.h>
+#include <pinos/server/node.h>
 #include <pinos/client/properties.h>
 
 /**
@@ -71,9 +71,9 @@ void              pinos_daemon_stop              (PinosDaemon *daemon);
 gchar *           pinos_daemon_export_uniquely   (PinosDaemon *daemon, GDBusObjectSkeleton *skel);
 void              pinos_daemon_unexport          (PinosDaemon *daemon, const gchar *name);
 
-void              pinos_daemon_add_source        (PinosDaemon *daemon, PinosSource *source);
-void              pinos_daemon_remove_source     (PinosDaemon *daemon, PinosSource *source);
-PinosSource *     pinos_daemon_find_source       (PinosDaemon     *daemon,
+void              pinos_daemon_add_node          (PinosDaemon *daemon, PinosNode *node);
+void              pinos_daemon_remove_node       (PinosDaemon *daemon, PinosNode *node);
+PinosNode *       pinos_daemon_find_node         (PinosDaemon     *daemon,
                                                   const gchar     *name,
                                                   PinosProperties *props,
                                                   GBytes          *format_filter,
diff --git a/pinos/server/node.c b/pinos/server/node.c
new file mode 100644
index 00000000..2ad9b4ec
--- /dev/null
+++ b/pinos/server/node.c
@@ -0,0 +1,333 @@
+/* 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 <gio/gio.h>
+
+#include "pinos/client/pinos.h"
+#include "pinos/client/enumtypes.h"
+
+#include "pinos/server/node.h"
+#include "pinos/server/source.h"
+#include "pinos/server/daemon.h"
+
+#include "pinos/dbus/org-pinos.h"
+
+
+#define PINOS_NODE_GET_PRIVATE(node)  \
+   (G_TYPE_INSTANCE_GET_PRIVATE ((node), PINOS_TYPE_NODE, PinosNodePrivate))
+
+struct _PinosNodePrivate
+{
+  PinosDaemon *daemon;
+  PinosObjectSkeleton *skeleton;
+  gchar *object_path;
+  PinosSource *source;
+  PinosSink *sink;
+};
+
+G_DEFINE_TYPE (PinosNode, pinos_node, G_TYPE_OBJECT);
+
+enum
+{
+  PROP_0,
+  PROP_DAEMON,
+  PROP_SKELETON,
+  PROP_OBJECT_PATH,
+};
+
+static void
+pinos_node_get_property (GObject    *_object,
+                         guint       prop_id,
+                         GValue     *value,
+                         GParamSpec *pspec)
+{
+  PinosNode *node = PINOS_NODE (_object);
+  PinosNodePrivate *priv = node->priv;
+
+  switch (prop_id) {
+    case PROP_DAEMON:
+      g_value_set_object (value, priv->daemon);
+      break;
+
+    case PROP_SKELETON:
+      g_value_set_object (value, priv->skeleton);
+      break;
+
+    case PROP_OBJECT_PATH:
+      g_value_set_string (value, priv->object_path);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (node, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+pinos_node_set_property (GObject      *_object,
+                         guint         prop_id,
+                         const GValue *value,
+                         GParamSpec   *pspec)
+{
+  PinosNode *node = PINOS_NODE (_object);
+  PinosNodePrivate *priv = node->priv;
+
+  switch (prop_id) {
+    case PROP_DAEMON:
+      priv->daemon = g_value_dup_object (value);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (node, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+node_register_object (PinosNode *node)
+{
+  PinosNodePrivate *priv = node->priv;
+  PinosDaemon *daemon = priv->daemon;
+
+  priv->skeleton = pinos_object_skeleton_new (PINOS_DBUS_OBJECT_NODE);
+
+  g_free (priv->object_path);
+  priv->object_path = pinos_daemon_export_uniquely (daemon, G_DBUS_OBJECT_SKELETON (priv->skeleton));
+
+  pinos_daemon_add_node (daemon, node);
+
+  return;
+}
+
+static void
+node_unregister_object (PinosNode *node)
+{
+  PinosNodePrivate *priv = node->priv;
+
+  pinos_daemon_unexport (priv->daemon, priv->object_path);
+  pinos_daemon_remove_node (priv->daemon, node);
+  g_clear_object (&priv->skeleton);
+}
+
+static void
+pinos_node_constructed (GObject * obj)
+{
+  PinosNode *node = PINOS_NODE (obj);
+
+  node_register_object (node);
+
+  G_OBJECT_CLASS (pinos_node_parent_class)->constructed (obj);
+}
+
+static void
+pinos_node_dispose (GObject * obj)
+{
+  PinosNode *node = PINOS_NODE (obj);
+
+  node_unregister_object (node);
+
+  G_OBJECT_CLASS (pinos_node_parent_class)->dispose (obj);
+}
+
+static void
+pinos_node_finalize (GObject * obj)
+{
+  PinosNode *node = PINOS_NODE (obj);
+  PinosNodePrivate *priv = node->priv;
+
+  g_free (priv->object_path);
+
+  G_OBJECT_CLASS (pinos_node_parent_class)->finalize (obj);
+}
+
+static void
+pinos_node_class_init (PinosNodeClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+
+  g_type_class_add_private (klass, sizeof (PinosNodePrivate));
+
+  gobject_class->constructed = pinos_node_constructed;
+  gobject_class->dispose = pinos_node_dispose;
+  gobject_class->finalize = pinos_node_finalize;
+  gobject_class->set_property = pinos_node_set_property;
+  gobject_class->get_property = pinos_node_get_property;
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_DAEMON,
+                                   g_param_spec_object ("daemon",
+                                                        "Daemon",
+                                                        "The Daemon",
+                                                        PINOS_TYPE_DAEMON,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_CONSTRUCT_ONLY |
+                                                        G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_OBJECT_PATH,
+                                   g_param_spec_string ("object-path",
+                                                        "Object Path",
+                                                        "The object path",
+                                                        NULL,
+                                                        G_PARAM_READABLE |
+                                                        G_PARAM_STATIC_STRINGS));
+}
+
+static void
+pinos_node_init (PinosNode * node)
+{
+  node->priv = PINOS_NODE_GET_PRIVATE (node);
+}
+
+/**
+ * pinos_node_get_daemon:
+ * @node: a #PinosNode
+ *
+ * Get the daemon of @node.
+ *
+ * Returns: the daemon of @node.
+ */
+PinosDaemon *
+pinos_node_get_daemon (PinosNode *node)
+{
+  PinosNodePrivate *priv;
+
+  g_return_val_if_fail (PINOS_IS_NODE (node), NULL);
+  priv = node->priv;
+
+ return priv->daemon;
+}
+
+/**
+ * pinos_node_get_object_path:
+ * @node: a #PinosNode
+ *
+ * Get the object path of @node.
+ *
+ * Returns: the object path of @node.
+ */
+const gchar *
+pinos_node_get_object_path (PinosNode *node)
+{
+  PinosNodePrivate *priv;
+
+  g_return_val_if_fail (PINOS_IS_NODE (node), NULL);
+  priv = node->priv;
+
+  return priv->object_path;
+}
+
+/**
+ * pinos_node_set_source:
+ * @node: a #PinosNode
+ * @source: a #PinosSource
+ *
+ * Set the #PinosSource of @node
+ */
+void
+pinos_node_set_source (PinosNode *node, PinosSource *source, GObject *iface)
+{
+  PinosNodePrivate *priv;
+
+  g_return_if_fail (PINOS_IS_NODE (node));
+  g_return_if_fail (source == NULL || PINOS_IS_SOURCE (source));
+  g_return_if_fail (iface == NULL || PINOS_IS_SOURCE1 (iface));
+  priv = node->priv;
+
+  if (source) {
+    pinos_object_skeleton_set_source1 (priv->skeleton, PINOS_SOURCE1 (iface));
+    priv->source = source;
+  } else {
+    pinos_object_skeleton_set_source1 (priv->skeleton, NULL);
+    priv->source = NULL;
+  }
+}
+
+/**
+ * pinos_node_get_source:
+ * @node: a #PinosNode
+ *
+ * Get the #PinosSource of @node
+ *
+ * Returns: the #PinosSource of @node or %NULL
+ */
+PinosSource *
+pinos_node_get_source (PinosNode *node)
+{
+  PinosNodePrivate *priv;
+
+  g_return_val_if_fail (PINOS_IS_NODE (node), NULL);
+  priv = node->priv;
+
+  return priv->source;
+}
+
+
+/**
+ * pinos_node_set_sink:
+ * @node: a #PinosNode
+ * @sink: a #PinosSink
+ *
+ * Set the #PinosSink of @node
+ */
+void
+pinos_node_set_sink (PinosNode *node, PinosSink *sink, GObject *iface)
+{
+  PinosNodePrivate *priv;
+
+  g_return_if_fail (PINOS_IS_NODE (node));
+  g_return_if_fail (sink == NULL || PINOS_IS_SINK (sink));
+  g_return_if_fail (iface == NULL || PINOS_IS_SINK1 (iface));
+  priv = node->priv;
+
+  if (sink) {
+    pinos_object_skeleton_set_sink1 (priv->skeleton, PINOS_SINK1 (iface));
+    priv->sink = sink;
+  } else {
+    pinos_object_skeleton_set_sink1 (priv->skeleton, NULL);
+    priv->sink = NULL;
+  }
+}
+
+/**
+ * pinos_node_get_sink:
+ * @node: a #PinosNode
+ *
+ * Get the #PinosSink of @node
+ *
+ * Returns: the #PinosSink of @node or %NULL
+ */
+PinosSink *
+pinos_node_get_sink (PinosNode *node)
+{
+  PinosNodePrivate *priv;
+
+  g_return_val_if_fail (PINOS_IS_NODE (node), NULL);
+  priv = node->priv;
+
+  return priv->sink;
+}
+
+PinosNode *
+pinos_node_new (PinosDaemon *daemon)
+{
+  return g_object_new (PINOS_TYPE_NODE,
+                       "daemon", daemon,
+                       NULL);
+}
diff --git a/pinos/server/node.h b/pinos/server/node.h
new file mode 100644
index 00000000..cbdb21b4
--- /dev/null
+++ b/pinos/server/node.h
@@ -0,0 +1,85 @@
+/* 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_NODE_H__
+#define __PINOS_NODE_H__
+
+#include <glib-object.h>
+
+G_BEGIN_DECLS
+
+typedef struct _PinosNode PinosNode;
+typedef struct _PinosNodeClass PinosNodeClass;
+typedef struct _PinosNodePrivate PinosNodePrivate;
+
+#include <pinos/client/introspect.h>
+#include <pinos/server/daemon.h>
+#include <pinos/server/source.h>
+#include <pinos/server/sink.h>
+
+#define PINOS_TYPE_NODE                 (pinos_node_get_type ())
+#define PINOS_IS_NODE(obj)              (G_TYPE_CHECK_INSTANCE_TYPE ((obj), PINOS_TYPE_NODE))
+#define PINOS_IS_NODE_CLASS(klass)      (G_TYPE_CHECK_CLASS_TYPE ((klass), PINOS_TYPE_NODE))
+#define PINOS_NODE_GET_CLASS(obj)       (G_TYPE_INSTANCE_GET_CLASS ((obj), PINOS_TYPE_NODE, PinosNodeClass))
+#define PINOS_NODE(obj)                 (G_TYPE_CHECK_INSTANCE_CAST ((obj), PINOS_TYPE_NODE, PinosNode))
+#define PINOS_NODE_CLASS(klass)         (G_TYPE_CHECK_CLASS_CAST ((klass), PINOS_TYPE_NODE, PinosNodeClass))
+#define PINOS_NODE_CAST(obj)            ((PinosNode*)(obj))
+#define PINOS_NODE_CLASS_CAST(klass)    ((PinosNodeClass*)(klass))
+
+/**
+ * PinosNode:
+ *
+ * Pinos node class.
+ */
+struct _PinosNode {
+  GObject object;
+
+  PinosNodePrivate *priv;
+};
+
+/**
+ * PinosNodeClass:
+ *
+ * Pinos node class.
+ */
+struct _PinosNodeClass {
+  GObjectClass parent_class;
+};
+
+/* normal GObject stuff */
+GType               pinos_node_get_type                (void);
+
+PinosNode *         pinos_node_new                     (PinosDaemon *daemon);
+
+PinosDaemon *       pinos_node_get_daemon              (PinosNode *node);
+const gchar *       pinos_node_get_object_path         (PinosNode *node);
+
+void                pinos_node_set_source              (PinosNode   *node,
+                                                        PinosSource *source,
+                                                        GObject     *iface);
+PinosSource *       pinos_node_get_source              (PinosNode   *node);
+
+void                pinos_node_set_sink                (PinosNode   *node,
+                                                        PinosSink   *sink,
+                                                        GObject     *iface);
+PinosSink *         pinos_node_get_sink                (PinosNode   *node);
+
+G_END_DECLS
+
+#endif /* __PINOS_NODE_H__ */
diff --git a/pinos/server/sink.c b/pinos/server/sink.c
new file mode 100644
index 00000000..006bfd08
--- /dev/null
+++ b/pinos/server/sink.c
@@ -0,0 +1,646 @@
+/* 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 <gio/gio.h>
+
+#include "pinos/client/pinos.h"
+#include "pinos/client/enumtypes.h"
+
+#include "pinos/server/sink.h"
+#include "pinos/server/node.h"
+
+#include "pinos/dbus/org-pinos.h"
+
+
+#define PINOS_SINK_GET_PRIVATE(obj)  \
+   (G_TYPE_INSTANCE_GET_PRIVATE ((obj), PINOS_TYPE_SINK, PinosSinkPrivate))
+
+struct _PinosSinkPrivate
+{
+  PinosNode *node;
+  PinosSink1 *iface;
+
+  gchar *name;
+  PinosProperties *properties;
+
+  PinosSinkState state;
+  GError *error;
+  guint idle_timeout;
+
+  GList *channels;
+};
+
+G_DEFINE_ABSTRACT_TYPE (PinosSink, pinos_sink, G_TYPE_OBJECT);
+
+enum
+{
+  PROP_0,
+  PROP_NODE,
+  PROP_NAME,
+  PROP_STATE,
+  PROP_PROPERTIES
+};
+
+static void
+pinos_sink_get_property (GObject    *_object,
+                           guint       prop_id,
+                           GValue     *value,
+                           GParamSpec *pspec)
+{
+  PinosSink *sink = PINOS_SINK (_object);
+  PinosSinkPrivate *priv = sink->priv;
+
+  switch (prop_id) {
+    case PROP_NODE:
+      g_value_set_object (value, priv->node);
+      break;
+
+    case PROP_NAME:
+      g_value_set_string (value, priv->name);
+      break;
+
+    case PROP_STATE:
+      g_value_set_enum (value, priv->state);
+      break;
+
+    case PROP_PROPERTIES:
+      g_value_set_boxed (value, priv->properties);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (sink, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+pinos_sink_set_property (GObject      *_object,
+                           guint         prop_id,
+                           const GValue *value,
+                           GParamSpec   *pspec)
+{
+  PinosSink *sink = PINOS_SINK (_object);
+  PinosSinkPrivate *priv = sink->priv;
+
+  switch (prop_id) {
+    case PROP_NODE:
+      priv->node = g_value_dup_object (value);
+      break;
+
+    case PROP_NAME:
+      g_free (priv->name);
+      priv->name = g_value_dup_string (value);
+      break;
+
+    case PROP_PROPERTIES:
+      if (priv->properties)
+        pinos_properties_free (priv->properties);
+      priv->properties = g_value_dup_boxed (value);
+      if (priv->iface)
+        g_object_set (priv->iface,
+            "properties", priv->properties ?
+                      pinos_properties_to_variant (priv->properties) : NULL,
+            NULL);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (sink, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+sink_register_object (PinosSink *sink)
+{
+  PinosSinkPrivate *priv = sink->priv;
+  GBytes *formats;
+  GVariant *variant;
+
+  formats = pinos_sink_get_formats (sink, NULL, NULL);
+
+  if (priv->properties)
+    variant = pinos_properties_to_variant (priv->properties);
+  else
+    variant = NULL;
+
+  priv->iface = pinos_sink1_skeleton_new ();
+  g_object_set (priv->iface, "name", priv->name,
+                             "state", priv->state,
+                             "properties", variant,
+                             "possible-formats", g_bytes_get_data (formats, NULL),
+                             NULL);
+  g_bytes_unref (formats);
+
+  pinos_node_set_sink (priv->node, sink, G_OBJECT (priv->iface));
+
+  return;
+}
+
+static void
+sink_unregister_object (PinosSink *sink)
+{
+  PinosSinkPrivate *priv = sink->priv;
+
+  pinos_node_set_sink (priv->node, NULL, NULL);
+  g_clear_object (&priv->iface);
+}
+
+static void
+pinos_sink_constructed (GObject * object)
+{
+  PinosSink *sink = PINOS_SINK (object);
+
+  sink_register_object (sink);
+
+  G_OBJECT_CLASS (pinos_sink_parent_class)->constructed (object);
+}
+
+static void
+do_remove_channel (PinosChannel *channel,
+                  gpointer       user_data)
+{
+  pinos_channel_remove (channel);
+}
+
+static void
+pinos_sink_dispose (GObject * object)
+{
+  PinosSink *sink = PINOS_SINK (object);
+  PinosSinkPrivate *priv = sink->priv;
+
+  g_list_foreach (priv->channels, (GFunc) do_remove_channel, sink);
+  sink_unregister_object (sink);
+
+  G_OBJECT_CLASS (pinos_sink_parent_class)->dispose (object);
+}
+
+static void
+pinos_sink_finalize (GObject * object)
+{
+  PinosSink *sink = PINOS_SINK (object);
+  PinosSinkPrivate *priv = sink->priv;
+
+  g_free (priv->name);
+  if (priv->properties)
+    pinos_properties_free (priv->properties);
+
+  G_OBJECT_CLASS (pinos_sink_parent_class)->finalize (object);
+}
+
+static gboolean
+default_set_state (PinosSink      *sink,
+                   PinosSinkState  state)
+{
+  pinos_sink_update_state (sink, state);
+  return TRUE;
+}
+
+static void
+handle_remove_channel (PinosChannel *channel,
+                       gpointer      user_data)
+{
+  PinosSink *sink = user_data;
+
+  pinos_sink_release_channel (sink, channel);
+}
+
+static PinosChannel *
+default_create_channel (PinosSink     *sink,
+                        const gchar     *client_path,
+                        GBytes          *format_filter,
+                        PinosProperties *props,
+                        const gchar     *prefix,
+                        GError          **error)
+{
+  PinosSinkPrivate *priv = sink->priv;
+  PinosChannel *channel;
+  GBytes *possible_formats;
+
+  possible_formats = pinos_sink_get_formats (sink, format_filter, error);
+  if (possible_formats == NULL)
+    return NULL;
+
+  channel = g_object_new (PINOS_TYPE_CHANNEL, "daemon", pinos_node_get_daemon (priv->node),
+                                              "object-path", prefix,
+                                              "client-path", client_path,
+                                              "owner-path", pinos_node_get_object_path (priv->node),
+                                              "possible-formats", possible_formats,
+                                              "properties", props,
+                                              NULL);
+  g_bytes_unref (possible_formats);
+
+  if (channel == NULL)
+    goto no_channel;
+
+  g_signal_connect (channel,
+                    "remove",
+                    (GCallback) handle_remove_channel,
+                    sink);
+
+  priv->channels = g_list_prepend (priv->channels, channel);
+
+  return g_object_ref (channel);
+
+  /* ERRORS */
+no_channel:
+  {
+    if (error)
+      *error = g_error_new (G_IO_ERROR,
+                            G_IO_ERROR_FAILED,
+                            "Could not create channel");
+    return NULL;
+  }
+}
+
+static gboolean
+default_release_channel (PinosSink  *sink,
+                         PinosChannel *channel)
+{
+  PinosSinkPrivate *priv = sink->priv;
+  GList *find;
+
+  find = g_list_find (priv->channels, channel);
+  if (find == NULL)
+    return FALSE;
+
+  priv->channels = g_list_delete_link (priv->channels, find);
+  g_object_unref (channel);
+
+  return TRUE;
+}
+
+static void
+pinos_sink_class_init (PinosSinkClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+
+  g_type_class_add_private (klass, sizeof (PinosSinkPrivate));
+
+  gobject_class->constructed = pinos_sink_constructed;
+  gobject_class->dispose = pinos_sink_dispose;
+  gobject_class->finalize = pinos_sink_finalize;
+  gobject_class->set_property = pinos_sink_set_property;
+  gobject_class->get_property = pinos_sink_get_property;
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_NODE,
+                                   g_param_spec_object ("node",
+                                                        "Node",
+                                                        "The Node",
+                                                        PINOS_TYPE_NODE,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_CONSTRUCT_ONLY |
+                                                        G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_NAME,
+                                   g_param_spec_string ("name",
+                                                        "Name",
+                                                        "The sink name",
+                                                        NULL,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_CONSTRUCT_ONLY |
+                                                        G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_STATE,
+                                   g_param_spec_enum ("state",
+                                                      "State",
+                                                      "The state of the sink",
+                                                      PINOS_TYPE_SINK_STATE,
+                                                      PINOS_SINK_STATE_SUSPENDED,
+                                                      G_PARAM_READABLE |
+                                                      G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_PROPERTIES,
+                                   g_param_spec_boxed ("properties",
+                                                       "Properties",
+                                                       "The properties of the sink",
+                                                       PINOS_TYPE_PROPERTIES,
+                                                       G_PARAM_READWRITE |
+                                                       G_PARAM_CONSTRUCT |
+                                                       G_PARAM_STATIC_STRINGS));
+
+
+  klass->set_state = default_set_state;
+  klass->create_channel = default_create_channel;
+  klass->release_channel = default_release_channel;
+}
+
+static void
+pinos_sink_init (PinosSink * sink)
+{
+  PinosSinkPrivate *priv = sink->priv = PINOS_SINK_GET_PRIVATE (sink);
+
+  priv->state = PINOS_SINK_STATE_SUSPENDED;
+}
+
+/**
+ * pinos_sink_get_formats:
+ * @sink: a #PinosSink
+ * @filter: a #GBytes
+ * @error: a #GError or %NULL
+ *
+ * Get all the currently supported formats for @sink and filter the
+ * results with @filter.
+ *
+ * Returns: the list of supported format. If %NULL is returned, @error will
+ * be set.
+ */
+GBytes *
+pinos_sink_get_formats (PinosSink  *sink,
+                          GBytes       *filter,
+                          GError      **error)
+{
+  PinosSinkClass *klass;
+  GBytes *res;
+
+  g_return_val_if_fail (PINOS_IS_SINK (sink), NULL);
+
+  klass = PINOS_SINK_GET_CLASS (sink);
+
+  if (klass->get_formats)
+    res = klass->get_formats (sink, filter, error);
+  else {
+    res = NULL;
+    if (error)
+      *error = g_error_new (G_IO_ERROR,
+                            G_IO_ERROR_NOT_SUPPORTED,
+                            "Format query is not supported");
+  }
+  return res;
+}
+
+static void
+remove_idle_timeout (PinosSink *sink)
+{
+  PinosSinkPrivate *priv = sink->priv;
+
+  if (priv->idle_timeout) {
+    g_source_remove (priv->idle_timeout);
+    priv->idle_timeout = 0;
+  }
+}
+
+/**
+ * pinos_sink_set_state:
+ * @sink: a #PinosSink
+ * @state: a #PinosSinkState
+ *
+ * Set the state of @sink to @state.
+ *
+ * Returns: %TRUE on success.
+ */
+gboolean
+pinos_sink_set_state (PinosSink      *sink,
+                        PinosSinkState  state)
+{
+  PinosSinkClass *klass;
+  gboolean res;
+
+  g_return_val_if_fail (PINOS_IS_SINK (sink), FALSE);
+
+  klass = PINOS_SINK_GET_CLASS (sink);
+
+  remove_idle_timeout (sink);
+
+  if (klass->set_state)
+    res = klass->set_state (sink, state);
+  else
+    res = FALSE;
+
+  return res;
+}
+
+/**
+ * pinos_sink_update_state:
+ * @sink: a #PinosSink
+ * @state: a #PinosSinkState
+ *
+ * Update the state of a sink. This method is used from
+ * inside @sink itself.
+ */
+void
+pinos_sink_update_state (PinosSink      *sink,
+                           PinosSinkState  state)
+{
+  PinosSinkPrivate *priv;
+
+  g_return_if_fail (PINOS_IS_SINK (sink));
+  priv = sink->priv;
+
+  if (priv->state != state) {
+    priv->state = state;
+    pinos_sink1_set_state (priv->iface, state);
+    g_object_notify (G_OBJECT (sink), "state");
+  }
+}
+
+/**
+ * pinos_sink_report_error:
+ * @sink: a #PinosSink
+ * @error: a #GError
+ *
+ * Report an error from within @sink.
+ */
+void
+pinos_sink_report_error (PinosSink *sink,
+                           GError      *error)
+{
+  PinosSinkPrivate *priv;
+
+  g_return_if_fail (PINOS_IS_SINK (sink));
+  priv = sink->priv;
+
+  g_clear_error (&priv->error);
+  remove_idle_timeout (sink);
+  priv->error = error;
+  priv->state = PINOS_SINK_STATE_ERROR;
+  g_debug ("got error state %s", error->message);
+  pinos_sink1_set_state (priv->iface, priv->state);
+  g_object_notify (G_OBJECT (sink), "state");
+}
+
+static gboolean
+idle_timeout (PinosSink *sink)
+{
+  PinosSinkPrivate *priv = sink->priv;
+
+  priv->idle_timeout = 0;
+  pinos_sink_set_state (sink, PINOS_SINK_STATE_SUSPENDED);
+
+  return G_SOURCE_REMOVE;
+}
+
+/**
+ * pinos_sink_report_idle:
+ * @sink: a #PinosSink
+ *
+ * Mark @sink as being idle. This will start a timeout that will
+ * set the sink to SUSPENDED.
+ */
+void
+pinos_sink_report_idle (PinosSink *sink)
+{
+  PinosSinkPrivate *priv;
+
+  g_return_if_fail (PINOS_IS_SINK (sink));
+  priv = sink->priv;
+
+  pinos_sink_set_state (sink, PINOS_SINK_STATE_IDLE);
+
+  priv->idle_timeout = g_timeout_add_seconds (3,
+                                              (GSourceFunc) idle_timeout,
+                                              sink);
+}
+
+/**
+ * pinos_sink_report_busy:
+ * @sink: a #PinosSink
+ *
+ * Mark @sink as being busy. This will set the state of the sink
+ * to the RUNNING state.
+ */
+void
+pinos_sink_report_busy (PinosSink *sink)
+{
+  g_return_if_fail (PINOS_IS_SINK (sink));
+
+  pinos_sink_set_state (sink, PINOS_SINK_STATE_RUNNING);
+}
+
+/**
+ * pinos_sink_update_possible_formats:
+ * @sink: a #PinosSink
+ * @formats: a #GBytes
+ *
+ * Update the possible formats in @sink to @formats. This function also
+ * updates the possible formats of the channels.
+ */
+void
+pinos_sink_update_possible_formats (PinosSink *sink, GBytes *formats)
+{
+  PinosSinkPrivate *priv;
+  GList *walk;
+
+  g_return_if_fail (PINOS_IS_SINK (sink));
+  priv = sink->priv;
+
+  if (priv->iface)
+    g_object_set (priv->iface, "possible-formats",
+                  g_bytes_get_data (formats, NULL),
+                  NULL);
+
+  for (walk = priv->channels; walk; walk = g_list_next (walk))
+    g_object_set (walk->data, "possible-formats", formats, NULL);
+}
+
+/**
+ * pinos_sink_update_format:
+ * @sink: a #PinosSink
+ * @format: a #GBytes
+ *
+ * Update the current format in @sink to @format. This function also
+ * updates the current format of the channels.
+ */
+void
+pinos_sink_update_format (PinosSink *sink, GBytes *format)
+{
+  PinosSinkPrivate *priv;
+  GList *walk;
+
+  g_return_if_fail (PINOS_IS_SINK (sink));
+  priv = sink->priv;
+
+  for (walk = priv->channels; walk; walk = g_list_next (walk))
+    g_object_set (walk->data, "format", format, NULL);
+}
+
+/**
+ * pinos_sink_create_channel:
+ * @sink: a #PinosSink
+ * @client_path: the client path
+ * @format_filter: a #GBytes
+ * @props: #PinosProperties
+ * @prefix: a prefix
+ * @error: a #GError or %NULL
+ *
+ * Create a new #PinosChannel for @sink.
+ *
+ * Returns: a new #PinosChannel or %NULL, in wich case @error will contain
+ *          more information about the error.
+ */
+PinosChannel *
+pinos_sink_create_channel (PinosSink     *sink,
+                             const gchar     *client_path,
+                             GBytes          *format_filter,
+                             PinosProperties *props,
+                             const gchar     *prefix,
+                             GError          **error)
+{
+  PinosSinkClass *klass;
+  PinosChannel *res;
+
+  g_return_val_if_fail (PINOS_IS_SINK (sink), NULL);
+
+  klass = PINOS_SINK_GET_CLASS (sink);
+
+  if (klass->create_channel) {
+    res = klass->create_channel (sink, client_path, format_filter, props, prefix, error);
+  } else {
+    if (error) {
+      *error = g_error_new (G_IO_ERROR,
+                            G_IO_ERROR_NOT_SUPPORTED,
+                            "CreateChannel not implemented");
+    }
+    res = NULL;
+  }
+
+  return res;
+}
+
+/**
+ * pinos_sink_release_channel:
+ * @sink: a #PinosSink
+ * @channel: a #PinosChannel
+ *
+ * Release the @channel in @sink.
+ *
+ * Returns: %TRUE on success.
+ */
+gboolean
+pinos_sink_release_channel (PinosSink  *sink,
+                              PinosChannel *channel)
+{
+  PinosSinkClass *klass;
+  gboolean res;
+
+  g_return_val_if_fail (PINOS_IS_SINK (sink), FALSE);
+  g_return_val_if_fail (PINOS_IS_CHANNEL (channel), FALSE);
+
+  klass = PINOS_SINK_GET_CLASS (sink);
+
+  if (klass->release_channel)
+    res = klass->release_channel (sink, channel);
+  else
+    res = FALSE;
+
+  return res;
+}
diff --git a/pinos/server/sink.h b/pinos/server/sink.h
new file mode 100644
index 00000000..0a231dd1
--- /dev/null
+++ b/pinos/server/sink.h
@@ -0,0 +1,109 @@
+/* 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_SINK_H__
+#define __PINOS_SINK_H__
+
+#include <glib-object.h>
+
+G_BEGIN_DECLS
+
+typedef struct _PinosSink PinosSink;
+typedef struct _PinosSinkClass PinosSinkClass;
+typedef struct _PinosSinkPrivate PinosSinkPrivate;
+
+#include <pinos/client/introspect.h>
+#include <pinos/server/channel.h>
+
+#define PINOS_TYPE_SINK                 (pinos_sink_get_type ())
+#define PINOS_IS_SINK(obj)              (G_TYPE_CHECK_INSTANCE_TYPE ((obj), PINOS_TYPE_SINK))
+#define PINOS_IS_SINK_CLASS(klass)      (G_TYPE_CHECK_CLASS_TYPE ((klass), PINOS_TYPE_SINK))
+#define PINOS_SINK_GET_CLASS(obj)       (G_TYPE_INSTANCE_GET_CLASS ((obj), PINOS_TYPE_SINK, PinosSinkClass))
+#define PINOS_SINK(obj)                 (G_TYPE_CHECK_INSTANCE_CAST ((obj), PINOS_TYPE_SINK, PinosSink))
+#define PINOS_SINK_CLASS(klass)         (G_TYPE_CHECK_CLASS_CAST ((klass), PINOS_TYPE_SINK, PinosSinkClass))
+#define PINOS_SINK_CAST(obj)            ((PinosSink*)(obj))
+#define PINOS_SINK_CLASS_CAST(klass)    ((PinosSinkClass*)(klass))
+
+/**
+ * PinosSink:
+ *
+ * Pinos sink object class.
+ */
+struct _PinosSink {
+  GObject object;
+
+  PinosSinkPrivate *priv;
+};
+
+/**
+ * PinosSinkClass:
+ * @get_formats: called to get a list of supported formats from the sink
+ * @set_state: called to change the current state of the sink
+ * @create_channel: called to create a new channel object
+ * @release_channel: called to release a channel object
+ *
+ * Pinos sink object class.
+ */
+struct _PinosSinkClass {
+  GObjectClass parent_class;
+
+  GBytes *            (*get_formats)  (PinosSink    *sink,
+                                       GBytes       *filter,
+                                       GError      **error);
+
+  gboolean            (*set_state)  (PinosSink *sink, PinosSinkState);
+
+  PinosChannel *      (*create_channel)  (PinosSink       *sink,
+                                          const gchar     *client_path,
+                                          GBytes          *format_filter,
+                                          PinosProperties *props,
+                                          const gchar     *prefix,
+                                          GError          **error);
+  gboolean            (*release_channel) (PinosSink       *sink,
+                                          PinosChannel    *channel);
+};
+
+/* normal GObject stuff */
+GType               pinos_sink_get_type                  (void);
+
+GBytes *            pinos_sink_get_formats               (PinosSink *sink,
+                                                          GBytes      *filter,
+                                                          GError     **error);
+
+gboolean            pinos_sink_set_state                 (PinosSink *sink, PinosSinkState state);
+void                pinos_sink_update_state              (PinosSink *sink, PinosSinkState state);
+void                pinos_sink_report_error              (PinosSink *sink, GError *error);
+void                pinos_sink_report_idle               (PinosSink *sink);
+void                pinos_sink_report_busy               (PinosSink *sink);
+
+void                pinos_sink_update_possible_formats   (PinosSink *sink, GBytes *formats);
+void                pinos_sink_update_format             (PinosSink *sink, GBytes *format);
+
+PinosChannel *      pinos_sink_create_channel            (PinosSink     *sink,
+                                                          const gchar     *client_path,
+                                                          GBytes          *format_filter,
+                                                          PinosProperties *props,
+                                                          const gchar     *prefix,
+                                                          GError          **error);
+gboolean            pinos_sink_release_channel           (PinosSink     *sink,
+                                                          PinosChannel    *channel);
+
+G_END_DECLS
+
+#endif /* __PINOS_SINK_H__ */
diff --git a/pinos/server/source.c b/pinos/server/source.c
index b779d59b..bc79d02a 100644
--- a/pinos/server/source.c
+++ b/pinos/server/source.c
@@ -23,7 +23,7 @@
 #include "pinos/client/enumtypes.h"
 
 #include "pinos/server/source.h"
-#include "pinos/server/daemon.h"
+#include "pinos/server/node.h"
 
 #include "pinos/dbus/org-pinos.h"
 
@@ -33,9 +33,8 @@
 
 struct _PinosSourcePrivate
 {
-  PinosDaemon *daemon;
+  PinosNode *node;
   PinosSource1 *iface;
-  gchar *object_path;
 
   gchar *name;
   PinosProperties *properties;
@@ -52,8 +51,7 @@ G_DEFINE_ABSTRACT_TYPE (PinosSource, pinos_source, G_TYPE_OBJECT);
 enum
 {
   PROP_0,
-  PROP_DAEMON,
-  PROP_OBJECT_PATH,
+  PROP_NODE,
   PROP_NAME,
   PROP_STATE,
   PROP_PROPERTIES
@@ -69,12 +67,8 @@ pinos_source_get_property (GObject    *_object,
   PinosSourcePrivate *priv = source->priv;
 
   switch (prop_id) {
-    case PROP_DAEMON:
-      g_value_set_object (value, priv->daemon);
-      break;
-
-    case PROP_OBJECT_PATH:
-      g_value_set_string (value, priv->object_path);
+    case PROP_NODE:
+      g_value_set_object (value, priv->node);
       break;
 
     case PROP_NAME:
@@ -105,13 +99,8 @@ pinos_source_set_property (GObject      *_object,
   PinosSourcePrivate *priv = source->priv;
 
   switch (prop_id) {
-    case PROP_DAEMON:
-      priv->daemon = g_value_dup_object (value);
-      break;
-
-    case PROP_OBJECT_PATH:
-      g_free (priv->object_path);
-      priv->object_path = g_value_dup_string (value);
+    case PROP_NODE:
+      priv->node = g_value_dup_object (value);
       break;
 
     case PROP_NAME:
@@ -140,15 +129,11 @@ static void
 source_register_object (PinosSource *source)
 {
   PinosSourcePrivate *priv = source->priv;
-  PinosDaemon *daemon = priv->daemon;
-  PinosObjectSkeleton *skel;
   GBytes *formats;
   GVariant *variant;
 
   formats = pinos_source_get_formats (source, NULL, NULL);
 
-  skel = pinos_object_skeleton_new (PINOS_DBUS_OBJECT_SOURCE);
-
   if (priv->properties)
     variant = pinos_properties_to_variant (priv->properties);
   else
@@ -160,12 +145,9 @@ source_register_object (PinosSource *source)
                              "properties", variant,
                              "possible-formats", g_bytes_get_data (formats, NULL),
                              NULL);
-  pinos_object_skeleton_set_source1 (skel, priv->iface);
   g_bytes_unref (formats);
 
-  g_free (priv->object_path);
-  priv->object_path = pinos_daemon_export_uniquely (daemon, G_DBUS_OBJECT_SKELETON (skel));
-  pinos_daemon_add_source (daemon, source);
+  pinos_node_set_source (priv->node, source, G_OBJECT (priv->iface));
 
   return;
 }
@@ -175,8 +157,7 @@ source_unregister_object (PinosSource *source)
 {
   PinosSourcePrivate *priv = source->priv;
 
-  pinos_daemon_remove_source (priv->daemon, source);
-  pinos_daemon_unexport (priv->daemon, priv->object_path);
+  pinos_node_set_source (priv->node, NULL, NULL);
   g_clear_object (&priv->iface);
 }
 
@@ -215,7 +196,6 @@ pinos_source_finalize (GObject * object)
   PinosSource *source = PINOS_SOURCE (object);
   PinosSourcePrivate *priv = source->priv;
 
-  g_free (priv->object_path);
   g_free (priv->name);
   if (priv->properties)
     pinos_properties_free (priv->properties);
@@ -256,10 +236,10 @@ default_create_channel (PinosSource     *source,
   if (possible_formats == NULL)
     return NULL;
 
-  channel = g_object_new (PINOS_TYPE_CHANNEL, "daemon", priv->daemon,
+  channel = g_object_new (PINOS_TYPE_CHANNEL, "daemon", pinos_node_get_daemon (priv->node),
                                               "object-path", prefix,
                                               "client-path", client_path,
-                                              "owner-path", priv->object_path,
+                                              "owner-path", pinos_node_get_object_path (priv->node),
                                               "possible-formats", possible_formats,
                                               "properties", props,
                                               NULL);
@@ -319,21 +299,11 @@ pinos_source_class_init (PinosSourceClass * klass)
   gobject_class->get_property = pinos_source_get_property;
 
   g_object_class_install_property (gobject_class,
-                                   PROP_DAEMON,
-                                   g_param_spec_object ("daemon",
-                                                        "Daemon",
-                                                        "The Daemon",
-                                                        PINOS_TYPE_DAEMON,
-                                                        G_PARAM_READWRITE |
-                                                        G_PARAM_CONSTRUCT_ONLY |
-                                                        G_PARAM_STATIC_STRINGS));
-
-  g_object_class_install_property (gobject_class,
-                                   PROP_OBJECT_PATH,
-                                   g_param_spec_string ("object-path",
-                                                        "Object Path",
-                                                        "The object path",
-                                                        NULL,
+                                   PROP_NODE,
+                                   g_param_spec_object ("node",
+                                                        "Node",
+                                                        "The Node",
+                                                        PINOS_TYPE_NODE,
                                                         G_PARAM_READWRITE |
                                                         G_PARAM_CONSTRUCT_ONLY |
                                                         G_PARAM_STATIC_STRINGS));
@@ -674,22 +644,3 @@ pinos_source_release_channel (PinosSource  *source,
 
   return res;
 }
-
-/**
- * pinos_source_get_object_path:
- * @source: a #PinosSource
- *
- * Get the object path of @source.
- *
- * Returns: the object path of @source.
- */
-const gchar *
-pinos_source_get_object_path (PinosSource *source)
-{
-  PinosSourcePrivate *priv;
-
-  g_return_val_if_fail (PINOS_IS_SOURCE (source), NULL);
-  priv = source->priv;
-
- return priv->object_path;
-}
diff --git a/pinos/server/source.h b/pinos/server/source.h
index 8ffbb13c..25c149bb 100644
--- a/pinos/server/source.h
+++ b/pinos/server/source.h
@@ -82,8 +82,6 @@ struct _PinosSourceClass {
 /* normal GObject stuff */
 GType               pinos_source_get_type                (void);
 
-const gchar *       pinos_source_get_object_path         (PinosSource *source);
-
 GBytes *            pinos_source_get_formats             (PinosSource *source,
                                                           GBytes      *filter,
                                                           GError     **error);
diff --git a/pinos/tools/pinos-monitor.c b/pinos/tools/pinos-monitor.c
index b13f1266..dadc26ac 100644
--- a/pinos/tools/pinos-monitor.c
+++ b/pinos/tools/pinos-monitor.c
@@ -157,7 +157,7 @@ dump_sink_info (PinosContext *c, const PinosSinkInfo *info, gpointer user_data)
   DumpData *data = user_data;
 
   g_print ("\tid: %p\n", info->id);
-  g_print ("\tsource-path: \"%s\"\n", info->sink_path);
+  g_print ("\tsink-path: \"%s\"\n", info->sink_path);
   if (data->print_all) {
     g_print ("%c\tname: \"%s\"\n", MARK_CHANGE (0), info->name);
     print_properties (info->properties, MARK_CHANGE (1));
author	Wim Taymans <wtaymans@redhat.com>	2016-05-05 13:31:18 +0200
committer	Wim Taymans <wtaymans@redhat.com>	2016-05-05 13:31:18 +0200
commit	b885d40390bb2398e827602cb729aadb4022af2a (patch)
tree	db347f142b2f0d1d5d49745a06fcbbe3aa9b7d45
parent	7597e48e020e95ef0808c94d44468530d91c098a (diff)