Work on adding pulsevideosink

Add the beginnings of a pulsevideosink
Make a new client-source object for every client that wants to provide
a stream. The client will have a handle to write the stream to.

src/Makefile.am

@@ -194,6 +194,7 @@ libpulsevideocore_@PV_MAJORMINOR@_la_SOURCES = \
 		server/pv-client.c server/pv-client.h \
 		server/pv-daemon.c server/pv-daemon.h \
 		server/pv-source.c server/pv-source.h \
+		server/pv-client-source.c server/pv-client-source.h \
 		server/pv-source-output.c server/pv-source-output.h \
 		modules/v4l2/pv-v4l2-source.c 
 
@@ -210,7 +211,11 @@ plugindir = $(libdir)/gstreamer-1.0
 plugin_LTLIBRARIES = libgstpulsevideosrc.la
 
 libgstpulsevideosrc_la_SOURCES = \
-                        gst/gstpvsrc.c 
+                        gst/gstpv.c \
+                        gst/gstfdpay.c \
+                        gst/gstfddepay.c \
+                        gst/gstpvsrc.c \
+                        gst/gstpvsink.c 
 
 libgstpulsevideosrc_la_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(GST_BASE_CFLAGS) $(GST_CFLAGS) $(GLIB_CFLAGS)
 libgstpulsevideosrc_la_LDFLAGS = $(GST_PLUGIN_LDFLAGS)
@@ -218,7 +223,7 @@ libgstpulsevideosrc_la_LIBADD =  $(GST_BASE_LIBS) $(GST_LIBS) $(GLIB_LIBS) $(LIB
 				 libpulsevideo-@PV_MAJORMINOR@.la libpulsevideocore-@PV_MAJORMINOR@.la
 libgstpulsevideosrc_la_LIBTOOLFLAGS = $(GST_PLUGIN_LIBTOOLFLAGS)
 
-noinst_HEADERS = gst/gstpvsrc.h
+noinst_HEADERS = gst/gstpvsrc.h gst/gstpvsink.h gst/gstfdpay.h gst/gstfddepay.h
 
 ###################################
 #        Some minor stuff         #

src/client/pulsevideo.c

@@ -34,7 +34,4 @@ void
 pv_init (int *argc, char **argv[])
 {
   gst_init (argc, argv);
-
-  gst_element_register (NULL, "pvfdpay", GST_RANK_NONE, GST_TYPE_FDPAY);
-  gst_element_register (NULL, "pvfddepay", GST_RANK_NONE, GST_TYPE_FDDEPAY);
 }

src/client/pv-stream.c

@@ -434,6 +434,7 @@ do_connect_capture (PvStream *stream)
  * @stream: a #PvStream
  * @source: the source name to connect to
  * @flags: a #PvStreamFlags
+ * @spec: a #GVariant
  *
  * Connect @stream for capturing from @source.
  *
@@ -465,6 +466,61 @@ pv_stream_connect_capture (PvStream      *stream,
   return TRUE;
 }
 
+static gboolean
+do_connect_provide (PvStream *stream)
+{
+  PvStreamPrivate *priv = stream->priv;
+  PvContext *context = priv->context;
+
+  g_assert (g_main_context_get_thread_default () == priv->context->priv->context);
+
+  g_dbus_proxy_call (context->priv->client,
+                     "CreateSourceInput",
+                     g_variant_new ("(@a{sv})",
+                       priv->spec),
+                     G_DBUS_CALL_FLAGS_NONE,
+                     -1,
+                     NULL, /* GCancellable *cancellable */
+                     on_source_output_created,
+                     stream);
+
+  return FALSE;
+}
+
+/**
+ * pv_stream_connect_provide:
+ * @stream: a #PvStream
+ * @flags: a #PvStreamFlags
+ * @spec: a #GVariant
+ *
+ * Connect @stream for providing data for a new source.
+ *
+ * Returns: %TRUE on success.
+ */
+gboolean
+pv_stream_connect_provide (PvStream *stream,
+                           PvStreamFlags flags,
+                           GVariant *spec)
+{
+  PvStreamPrivate *priv;
+  PvContext *context;
+
+  g_return_val_if_fail (PV_IS_STREAM (stream), FALSE);
+  g_return_val_if_fail (spec != NULL, FALSE);
+
+  priv = stream->priv;
+  context = priv->context;
+  g_return_val_if_fail (pv_context_get_state (context) == PV_CONTEXT_STATE_READY, FALSE);
+
+  priv->spec = spec;
+
+  stream_set_state (stream, PV_STREAM_STATE_CONNECTING);
+
+  g_main_context_invoke (context->priv->context, (GSourceFunc) do_connect_provide, stream);
+
+  return TRUE;
+}
+
 static void
 on_source_output_removed (GObject *source_object,
                           GAsyncResult *res,
@@ -861,3 +917,28 @@ pv_stream_capture_buffer (PvStream *stream, PvBufferInfo *info)
 
   return TRUE;
 }
+
+/**
+ * pv_stream_provide_buffer:
+ * @stream: a #PvStream
+ * @info: a #PvBufferInfo
+ *
+ * Provide the next buffer from @stream. This function should be called every
+ * time a new frame becomes available.
+ *
+ * Returns: %TRUE when @info was handled
+ */
+gboolean
+pv_stream_provide_buffer (PvStream *stream, PvBufferInfo *info)
+{
+  PvStreamPrivate *priv;
+
+  g_return_val_if_fail (PV_IS_STREAM (stream), FALSE);
+  g_return_val_if_fail (info != NULL, FALSE);
+
+  priv = stream->priv;
+  g_return_val_if_fail (priv->state == PV_STREAM_STATE_STREAMING, FALSE);
+
+  return TRUE;
+}
+

src/dbus/org.pulsevideo.xml

@@ -63,6 +63,17 @@
       <arg type='a{sv}' name='props' direction='in'/>
       <arg type='o' name='output' direction='out'/>
     </method>
+    <!-- CreateSourceInput:
+         @props: input properties
+         @source: the new Source1 object path
+         @input: the SourceInput1 object path
+
+         Create a new source and input object with given @props
+    -->
+    <method name='CreateSourceInput'>
+      <arg type='a{sv}' name='props' direction='in'/>
+      <arg type='o' name='input' direction='out'/>
+    </method>
   </interface>
 
   <!--

src/gst/gstpv.c

@@ -0,0 +1,58 @@
+/* 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-pulsevideosrc
+ *
+ * <refsect2>
+ * <title>Example launch line</title>
+ * |[
+ * gst-launch -v pulsevideosrc ! ximagesink
+ * ]| Shows pulsevideo output in an X window.
+ * </refsect2>
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include "gstpvsrc.h"
+#include "gstpvsink.h"
+#include "gstfdpay.h"
+#include "gstfddepay.h"
+
+static gboolean
+plugin_init (GstPlugin * plugin)
+{
+  gst_element_register (plugin, "pvfdpay", GST_RANK_NONE,
+      GST_TYPE_FDPAY);
+  gst_element_register (plugin, "pvfddepay", GST_RANK_NONE,
+      GST_TYPE_FDDEPAY);
+  gst_element_register (plugin, "pulsevideosrc", GST_RANK_NONE,
+      GST_TYPE_PULSEVIDEO_SRC);
+  gst_element_register (plugin, "pulsevideosink", GST_RANK_NONE,
+      GST_TYPE_PULSEVIDEO_SINK);
+  return TRUE;
+}
+
+GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
+    GST_VERSION_MINOR,
+    pulsevideo,
+    "Uses pulsevideo to handle video streams",
+    plugin_init, VERSION, "LGPL", "pulsevideo", "pulsevideo.org")

src/gst/gstpvsink.c

@@ -0,0 +1,552 @@
+/* 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-pulsevideosink
+ *
+ * <refsect2>
+ * <title>Example launch line</title>
+ * |[
+ * gst-launch -v pulsevideosink ! ximagesink
+ * ]| Shows pulsevideo output in an X window.
+ * </refsect2>
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+#include "gstpvsink.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 "gsttmpfileallocator.h"
+
+
+GST_DEBUG_CATEGORY_STATIC (pulsevideo_sink_debug);
+#define GST_CAT_DEFAULT pulsevideo_sink_debug
+
+enum
+{
+  PROP_0,
+  PROP_LAST
+};
+
+
+#define PVS_VIDEO_CAPS GST_VIDEO_CAPS_MAKE (GST_VIDEO_FORMATS_ALL)
+
+static GstStaticPadTemplate gst_pulsevideo_sink_template =
+GST_STATIC_PAD_TEMPLATE ("sink",
+    GST_PAD_SINK,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS (PVS_VIDEO_CAPS)
+    );
+
+#define gst_pulsevideo_sink_parent_class parent_class
+G_DEFINE_TYPE (GstPulsevideoSink, gst_pulsevideo_sink, GST_TYPE_BASE_SINK);
+
+static void gst_pulsevideo_sink_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec);
+static void gst_pulsevideo_sink_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec);
+
+static GstStateChangeReturn
+gst_pulsevideo_sink_change_state (GstElement * element, GstStateChange transition);
+
+static GstCaps *gst_pulsevideo_sink_getcaps (GstBaseSink * bsink, GstCaps * filter);
+static gboolean gst_pulsevideo_sink_setcaps (GstBaseSink * bsink, GstCaps * caps);
+static GstCaps *gst_pulsevideo_sink_sink_fixate (GstBaseSink * bsink,
+    GstCaps * caps);
+
+static gboolean gst_pulsevideo_sink_propose_allocation (GstBaseSink * bsink,
+    GstQuery * query);
+static GstFlowReturn gst_pulsevideo_sink_render (GstBaseSink * psink,
+    GstBuffer * buffer);
+static gboolean gst_pulsevideo_sink_start (GstBaseSink * basesink);
+static gboolean gst_pulsevideo_sink_stop (GstBaseSink * basesink);
+
+static void
+gst_pulsevideo_sink_class_init (GstPulsevideoSinkClass * klass)
+{
+  GObjectClass *gobject_class;
+  GstElementClass *gstelement_class;
+  GstBaseSinkClass *gstbasesink_class;
+
+  gobject_class = (GObjectClass *) klass;
+  gstelement_class = (GstElementClass *) klass;
+  gstbasesink_class = (GstBaseSinkClass *) klass;
+
+  gobject_class->set_property = gst_pulsevideo_sink_set_property;
+  gobject_class->get_property = gst_pulsevideo_sink_get_property;
+
+  gstelement_class->change_state = gst_pulsevideo_sink_change_state;
+
+  gst_element_class_set_static_metadata (gstelement_class,
+      "Pulsevideo sink", "Sink/Video",
+      "Send video to pulsevideo", "Wim Taymans <wim.taymans@gmail.com>");
+
+  gst_element_class_add_pad_template (gstelement_class,
+      gst_static_pad_template_get (&gst_pulsevideo_sink_template));
+
+  gstbasesink_class->get_caps = gst_pulsevideo_sink_getcaps;
+  gstbasesink_class->set_caps = gst_pulsevideo_sink_setcaps;
+  gstbasesink_class->fixate = gst_pulsevideo_sink_sink_fixate;
+  gstbasesink_class->start = gst_pulsevideo_sink_start;
+  gstbasesink_class->stop = gst_pulsevideo_sink_stop;
+  gstbasesink_class->render = gst_pulsevideo_sink_render;
+  gstbasesink_class->propose_allocation = gst_pulsevideo_sink_propose_allocation;
+
+  GST_DEBUG_CATEGORY_INIT (pulsevideo_sink_debug, "pulsevideosink", 0,
+      "Pulsevideo Sink");
+}
+
+static void
+gst_pulsevideo_sink_init (GstPulsevideoSink * sink)
+{
+  sink->allocator = gst_tmpfile_allocator_new ();
+  g_mutex_init (&sink->lock);
+  g_cond_init (&sink->cond);
+}
+
+static GstCaps *
+gst_pulsevideo_sink_sink_fixate (GstBaseSink * bsink, GstCaps * caps)
+{
+  GstStructure *structure;
+
+  caps = gst_caps_make_writable (caps);
+
+  structure = gst_caps_get_structure (caps, 0);
+
+  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);
+
+  caps = GST_BASE_SINK_CLASS (parent_class)->fixate (bsink, caps);
+
+  return caps;
+}
+
+static void
+gst_pulsevideo_sink_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  switch (prop_id) {
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pulsevideo_sink_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec)
+{
+  switch (prop_id) {
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static gboolean
+gst_pulsevideo_sink_propose_allocation (GstBaseSink * bsink, GstQuery * query)
+{
+  GstPulsevideoSink *pvsink;
+  GstBufferPool *pool;
+  gboolean update;
+  guint size, min, max;
+  GstStructure *config;
+  GstCaps *caps = NULL;
+
+  pvsink = GST_PULSEVIDEO_SINK (bsink);
+
+  if (gst_query_get_n_allocation_pools (query) > 0) {
+    gst_query_parse_nth_allocation_pool (query, 0, &pool, &size, &min, &max);
+
+    /* adjust size */
+    size = MAX (size, pvsink->info.size);
+    update = TRUE;
+  } else {
+    pool = NULL;
+    size = pvsink->info.size;
+    min = max = 0;
+    update = FALSE;
+  }
+
+  /* no downstream pool, make our own */
+  if (pool == NULL) {
+    pool = gst_video_buffer_pool_new ();
+  }
+
+  config = gst_buffer_pool_get_config (pool);
+
+  gst_query_parse_allocation (query, &caps, NULL);
+  if (caps)
+    gst_buffer_pool_config_set_params (config, caps, size, min, max);
+
+  if (gst_query_find_allocation_meta (query, GST_VIDEO_META_API_TYPE, NULL)) {
+    gst_buffer_pool_config_add_option (config,
+        GST_BUFFER_POOL_OPTION_VIDEO_META);
+  }
+  gst_buffer_pool_set_config (pool, config);
+
+  if (update)
+    gst_query_set_nth_allocation_pool (query, 0, pool, size, min, max);
+  else
+    gst_query_add_allocation_pool (query, pool, size, min, max);
+
+  if (pool)
+    gst_object_unref (pool);
+
+  return GST_BASE_SINK_CLASS (parent_class)->propose_allocation (bsink, query);
+}
+
+static void
+on_new_buffer (GObject    *gobject,
+               gpointer    user_data)
+{
+  GstPulsevideoSink *pvsink = user_data;
+
+  g_mutex_lock (&pvsink->lock);
+  g_cond_signal (&pvsink->cond);
+  g_mutex_unlock (&pvsink->lock);
+}
+
+static void
+on_stream_notify (GObject    *gobject,
+                  GParamSpec *pspec,
+                  gpointer    user_data)
+{
+  PvStreamState state;
+  GstPulsevideoSink *pvsink = user_data;
+
+  g_mutex_lock (&pvsink->lock);
+  state = pv_stream_get_state (pvsink->stream);
+  g_print ("got stream state %d\n", state);
+  if (state == PV_STREAM_STATE_ERROR) {
+    GST_ELEMENT_ERROR (pvsink, RESOURCE, FAILED,
+        ("Failed to connect stream: %s",
+          pv_stream_get_error (pvsink->stream)->message), (NULL));
+  }
+  g_cond_broadcast (&pvsink->cond);
+  g_mutex_unlock (&pvsink->lock);
+}
+
+static GstCaps *
+gst_pulsevideo_sink_getcaps (GstBaseSink * bsink, GstCaps * filter)
+{
+  return GST_BASE_SINK_CLASS (parent_class)->get_caps (bsink, filter);
+
+}
+
+static gboolean
+gst_pulsevideo_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
+{
+  const GstStructure *structure;
+  GstPulsevideoSink *pvsink;
+  GstVideoInfo info;
+  GVariantBuilder builder;
+
+  pvsink = GST_PULSEVIDEO_SINK (bsink);
+
+  structure = gst_caps_get_structure (caps, 0);
+
+  if (gst_structure_has_name (structure, "video/x-raw")) {
+    /* we can use the parsing code */
+    if (!gst_video_info_from_caps (&info, caps))
+      goto parse_failed;
+
+  } else {
+    goto unsupported_caps;
+  }
+
+  /* looks ok here */
+  pvsink->info = info;
+
+  pvsink->stream = pv_stream_new (pvsink->ctx, "test", NULL);
+  g_signal_connect (pvsink->stream, "notify::state", (GCallback) on_stream_notify, pvsink);
+  g_signal_connect (pvsink->stream, "new-buffer", (GCallback) on_new_buffer, pvsink);
+
+  g_variant_builder_init (&builder, G_VARIANT_TYPE ("a{sv}"));
+  g_variant_builder_add (&builder, "{sv}", "format.encoding", g_variant_new_string ("video/x-raw"));
+  g_variant_builder_add (&builder, "{sv}", "format.format",
+      g_variant_new_string (gst_video_format_to_string (info.finfo->format)));
+  g_variant_builder_add (&builder, "{sv}", "format.width", g_variant_new_int32 (info.width));
+  g_variant_builder_add (&builder, "{sv}", "format.height", g_variant_new_int32 (info.height));
+  g_variant_builder_add (&builder, "{sv}", "format.views", g_variant_new_int32 (info.views));
+//  g_variant_builder_add (&builder, "{sv}", "format.chroma-site",
+//      g_variant_new_string (gst_video_chroma_to_string (info.chroma_site)));
+//  g_variant_builder_add (&builder, "{sv}", "format.colorimetry",
+//      g_variant_new_take_string (gst_video_colorimetry_to_string (&info.colorimetry)));
+//  g_variant_builder_add (&builder, "{sv}", "format.interlace-mode",
+//      g_variant_new_string (gst_video_interlace_mode_to_string (info.interlace_mode)));
+
+  pv_stream_connect_provide (pvsink->stream, 0, g_variant_builder_end (&builder));
+
+  g_mutex_lock (&pvsink->lock);
+  while (TRUE) {
+    PvStreamState state = pv_stream_get_state (pvsink->stream);
+
+    if (state == PV_STREAM_STATE_READY)
+      break;
+
+    if (state == PV_STREAM_STATE_ERROR)
+      goto connect_error;
+
+    g_cond_wait (&pvsink->cond, &pvsink->lock);
+  }
+  g_mutex_unlock (&pvsink->lock);
+
+  pv_stream_start (pvsink->stream, PV_STREAM_MODE_BUFFER);
+
+  GST_DEBUG_OBJECT (pvsink, "size %dx%d, %d/%d fps",
+      info.width, info.height, info.fps_n, info.fps_d);
+
+  return TRUE;
+
+  /* ERRORS */
+parse_failed:
+  {
+    GST_DEBUG_OBJECT (bsink, "failed to parse caps");
+    return FALSE;
+  }
+unsupported_caps:
+  {
+    GST_DEBUG_OBJECT (bsink, "unsupported caps: %" GST_PTR_FORMAT, caps);
+    return FALSE;
+  }
+connect_error:
+  {
+    g_mutex_unlock (&pvsink->lock);
+    return FALSE;
+  }
+}
+
+static GstFlowReturn
+gst_pulsevideo_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
+{
+  GstPulsevideoSink *pvsink;
+  PvBufferInfo info;
+  GSocketControlMessage *mesg;
+  GstMemory *mem = NULL;
+
+  pvsink = GST_PULSEVIDEO_SINK (bsink);
+
+  if (G_UNLIKELY (GST_VIDEO_INFO_FORMAT (&pvsink->info) ==
+          GST_VIDEO_FORMAT_UNKNOWN))
+    goto not_negotiated;
+
+  info.flags = 0;
+  info.seq = 0;
+  info.pts = GST_BUFFER_TIMESTAMP (buffer);
+  info.dts_offset = 0;
+  info.offset = 0;
+  info.size = gst_buffer_get_size (buffer);
+
+  mesg = g_unix_fd_message_new ();
+  if (gst_buffer_n_memory (buffer) == 1
+      && gst_is_fd_memory (gst_buffer_peek_memory (buffer, 0))) {
+    mem = gst_buffer_get_memory (buffer, 0);
+  } else {
+    GstMapInfo minfo;
+    GstAllocationParams params = {0, 0, 0, 0, { NULL, }};
+
+    GST_INFO_OBJECT (bsink, "Buffer cannot be payloaded without copying");
+
+    mem = gst_allocator_alloc (pvsink->allocator, info.size, &params);
+    if (!gst_memory_map (mem, &minfo, GST_MAP_WRITE))
+      goto error;
+    gst_buffer_extract (buffer, 0, minfo.data, info.size);
+    gst_memory_unmap (mem, &minfo);
+  }
+  g_unix_fd_message_append_fd ((GUnixFDMessage*)mesg, gst_fd_memory_get_fd (mem), NULL);
+  info.message = mesg;
+
+  g_mutex_lock (&pvsink->lock);
+  pv_stream_provide_buffer (pvsink->stream, &info);
+  g_mutex_unlock (&pvsink->lock);
+
+  return GST_FLOW_OK;
+
+not_negotiated:
+  {
+    return GST_FLOW_NOT_NEGOTIATED;
+  }
+error:
+  {
+    return GST_FLOW_ERROR;
+  }
+}
+
+static gboolean
+gst_pulsevideo_sink_start (GstBaseSink * basesink)
+{
+  GstPulsevideoSink *sink = GST_PULSEVIDEO_SINK (basesink);
+
+  gst_video_info_init (&sink->info);
+
+  return TRUE;
+}
+
+static gboolean
+gst_pulsevideo_sink_stop (GstBaseSink * basesink)
+{
+  return TRUE;
+}
+
+static gpointer
+handle_mainloop (GstPulsevideoSink *this)
+{
+  g_main_context_push_thread_default (this->context);
+  g_print ("run mainloop\n");
+  g_main_loop_run (this->loop);
+  g_print ("quit mainloop\n");
+  g_main_context_pop_thread_default (this->context);
+
+  return NULL;
+}
+
+static void
+on_state_notify (GObject    *gobject,
+                 GParamSpec *pspec,
+                 gpointer    user_data)
+{
+  GstPulsevideoSink *pvsink = user_data;
+  PvContextState state;
+
+  g_mutex_lock (&pvsink->lock);
+  state = pv_context_get_state (pvsink->ctx);
+  g_print ("got context state %d\n", state);
+  g_cond_broadcast (&pvsink->cond);
+  g_mutex_unlock (&pvsink->lock);
+
+  if (state == PV_CONTEXT_STATE_ERROR) {
+    GST_ELEMENT_ERROR (pvsink, RESOURCE, FAILED,
+        ("Failed to connect stream: %s",
+          pv_context_get_error (pvsink->ctx)->message), (NULL));
+  }
+}
+
+static gboolean
+gst_pulsevideo_sink_open (GstPulsevideoSink * pvsink)
+{
+
+  pvsink->ctx = pv_context_new (pvsink->context, "test-client", NULL);
+  g_signal_connect (pvsink->ctx, "notify::state", (GCallback) on_state_notify, pvsink);
+
+  pv_context_connect(pvsink->ctx, PV_CONTEXT_FLAGS_NONE);
+
+  g_mutex_lock (&pvsink->lock);
+  while (TRUE) {
+    PvContextState state = pv_context_get_state (pvsink->ctx);
+
+    if (state == PV_CONTEXT_STATE_READY)
+      break;
+
+    if (state == PV_CONTEXT_STATE_ERROR)
+      goto connect_error;
+
+    g_cond_wait (&pvsink->cond, &pvsink->lock);
+  }
+  g_mutex_unlock (&pvsink->lock);
+
+  return TRUE;
+
+  /* ERRORS */
+connect_error:
+  {
+    g_mutex_unlock (&pvsink->lock);
+    return FALSE;
+  }
+}
+
+static GstStateChangeReturn
+gst_pulsevideo_sink_change_state (GstElement * element, GstStateChange transition)
+{
+  GstStateChangeReturn ret;
+  GstPulsevideoSink *this = GST_PULSEVIDEO_SINK_CAST (element);
+
+  switch (transition) {
+    case GST_STATE_CHANGE_NULL_TO_READY:
+      this->context = g_main_context_new ();
+      g_print ("context %p\n", this->context);
+      this->loop = g_main_loop_new (this->context, FALSE);
+      this->thread = g_thread_new ("pulsevideo", (GThreadFunc) handle_mainloop, this);
+      break;
+    case GST_STATE_CHANGE_READY_TO_PAUSED:
+      if (!gst_pulsevideo_sink_open (this)) {
+        ret = GST_STATE_CHANGE_FAILURE;
+        goto exit;
+      }
+      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:
+      break;
+    case GST_STATE_CHANGE_READY_TO_NULL:
+      g_main_loop_quit (this->loop);
+      g_thread_join (this->thread);
+      g_main_loop_unref (this->loop);
+      g_main_context_unref (this->context);
+      break;
+    default:
+      break;
+  }
+
+exit:
+  return ret;
+}

src/gst/gstpvsink.h

@@ -0,0 +1,82 @@
+/* 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_PULSEVIDEO_SINK_H__
+#define __GST_PULSEVIDEO_SINK_H__
+
+#include <gst/gst.h>
+#include <gst/base/gstbasesink.h>
+
+#include <gst/video/video.h>
+
+#include <client/pv-context.h>
+#include <client/pv-stream.h>
+#include <client/pv-introspect.h>
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_PULSEVIDEO_SINK \
+  (gst_pulsevideo_sink_get_type())
+#define GST_PULSEVIDEO_SINK(obj) \
+  (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_PULSEVIDEO_SINK,GstPulsevideoSink))
+#define GST_PULSEVIDEO_SINK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_PULSEVIDEO_SINK,GstPulsevideoSinkClass))
+#define GST_IS_PULSEVIDEO_SINK(obj) \
+  (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_PULSEVIDEO_SINK))
+#define GST_IS_PULSEVIDEO_SINK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PULSEVIDEO_SINK))
+#define GST_PULSEVIDEO_SINK_CAST(obj) \
+  ((GstPulsevideoSink *) (obj))
+
+typedef struct _GstPulsevideoSink GstPulsevideoSink;
+typedef struct _GstPulsevideoSinkClass GstPulsevideoSinkClass;
+
+/**
+ * GstPulsevideoSink:
+ *
+ * Opaque data structure.
+ */
+struct _GstPulsevideoSink {
+  GstBaseSink element;
+
+  /*< private >*/
+
+  /* video state */
+  GstVideoInfo info;
+
+  GMainContext *context;
+  GMainLoop *loop;
+  GThread *thread;
+  PvContext *ctx;
+  PvStream *stream;
+  GstAllocator *allocator;
+
+  GMutex lock;
+  GCond cond;
+};
+
+struct _GstPulsevideoSinkClass {
+  GstBaseSinkClass parent_class;
+};
+
+GType gst_pulsevideo_sink_get_type (void);
+
+G_END_DECLS
+
+#endif /* __GST_PULSEVIDEO_SINK_H__ */

src/gst/gstpvsrc.c

@@ -124,6 +124,9 @@ gst_pulsevideo_src_class_init (GstPulsevideoSrcClass * klass)
   gstbasesrc_class->decide_allocation = gst_pulsevideo_src_decide_allocation;
 
   gstpushsrc_class->create = gst_pulsevideo_src_create;
+
+  GST_DEBUG_CATEGORY_INIT (pulsevideo_src_debug, "pulsevideosrc", 0,
+      "Pulsevideo Source");
 }
 
 static void
@@ -641,19 +644,3 @@ gst_pulsevideo_src_change_state (GstElement * element, GstStateChange transition
 exit:
   return ret;
 }
-
-static gboolean
-plugin_init (GstPlugin * plugin)
-{
-  GST_DEBUG_CATEGORY_INIT (pulsevideo_src_debug, "pulsevideosrc", 0,
-      "Pulsevideo Source");
-
-  return gst_element_register (plugin, "pulsevideosrc", GST_RANK_NONE,
-      GST_TYPE_PULSEVIDEO_SRC);
-}
-
-GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
-    GST_VERSION_MINOR,
-    pulsevideo,
-    "Uses pulsevideo to create a video stream",
-    plugin_init, VERSION, "LGPL", "pulsevideo", "pulsevideo.org")

src/server/pv-client-source.c

@@ -0,0 +1,297 @@
+/* Pulsevideo
+ * 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 <gst/gst.h>
+#include <gio/gio.h>
+
+#include "server/pv-daemon.h"
+#include "pv-client-source.h"
+
+#define PV_CLIENT_SOURCE_GET_PRIVATE(obj)  \
+     (G_TYPE_INSTANCE_GET_PRIVATE ((obj), PV_TYPE_CLIENT_SOURCE, PvClientSourcePrivate))
+
+struct _PvClientSourcePrivate
+{
+  GstElement *pipeline;
+  GstElement *src;
+  GstElement *filter;
+  GstElement *sink;
+
+  GSocket *socket;
+
+  PvSourceOutput *input;
+};
+
+G_DEFINE_TYPE (PvClientSource, pv_client_source, PV_TYPE_SOURCE);
+
+static gboolean
+bus_handler (GstBus * bus, GstMessage * message, gpointer user_data)
+{
+  PvSource *source = user_data;
+  PvClientSourcePrivate *priv = PV_CLIENT_SOURCE (source)->priv;
+
+  switch (GST_MESSAGE_TYPE (message)) {
+    case GST_MESSAGE_ERROR:
+    {
+      GError *error;
+      gchar *debug;
+
+      gst_message_parse_error (message, &error, &debug);
+      g_print ("got error %s (%s)\n", error->message, debug);
+      g_free (debug);
+
+      pv_source_report_error (source, error);
+      gst_element_set_state (priv->pipeline, GST_STATE_NULL);
+      break;
+    }
+    default:
+      break;
+  }
+  return TRUE;
+}
+
+static void
+setup_pipeline (PvClientSource *source)
+{
+  PvClientSourcePrivate *priv = source->priv;
+  GstBus *bus;
+
+  priv->pipeline = gst_parse_launch ("socketsrc name=src ! "
+                                         "capsfilter name=filter ! "
+                                     "multisocketsink "
+                                         "buffers-max=2 "
+                                         "buffers-soft-max=1 "
+                                         "recover-policy=latest "
+                                         "sync-method=latest "
+                                         "name=sink "
+                                         "sync=true "
+                                         "enable-last-sample=false",
+                                      NULL);
+  priv->filter = gst_bin_get_by_name (GST_BIN (priv->pipeline), "filter");
+  priv->sink = gst_bin_get_by_name (GST_BIN (priv->pipeline), "sink");
+  priv->src = gst_bin_get_by_name (GST_BIN (priv->pipeline), "src");
+
+  bus = gst_pipeline_get_bus (GST_PIPELINE (priv->pipeline));
+  gst_bus_add_watch (bus, bus_handler, source);
+  gst_object_unref (bus);
+}
+
+static void
+collect_capabilities (PvSource * source)
+{
+  PvClientSourcePrivate *priv = PV_CLIENT_SOURCE (source)->priv;
+  GstCaps *res;
+  GstQuery *query;
+
+  query = gst_query_new_caps (NULL);
+  gst_element_query (priv->src, query);
+  gst_query_parse_caps_result (query, &res);
+  g_print ("%s\n", gst_caps_to_string (res));
+  gst_query_unref (query);
+}
+
+static gboolean
+client_set_state (PvSource *source, PvSourceState state)
+{
+  PvClientSourcePrivate *priv = PV_CLIENT_SOURCE (source)->priv;
+
+  switch (state) {
+    case PV_SOURCE_STATE_SUSPENDED:
+      gst_element_set_state (priv->pipeline, GST_STATE_NULL);
+      break;
+
+    case PV_SOURCE_STATE_INIT:
+      gst_element_set_state (priv->pipeline, GST_STATE_READY);
+      collect_capabilities (source);
+      break;
+
+    case PV_SOURCE_STATE_IDLE:
+      gst_element_set_state (priv->pipeline, GST_STATE_PAUSED);
+      break;
+
+    case PV_SOURCE_STATE_RUNNING:
+      gst_element_set_state (priv->pipeline, GST_STATE_PLAYING);
+      break;
+
+    case PV_SOURCE_STATE_ERROR:
+      break;
+  }
+  pv_source_update_state (source, state);
+  return TRUE;
+}
+
+static GVariant *
+client_get_capabilities (PvSource *source, GVariant *props)
+{
+  return NULL;
+}
+
+static void
+on_socket_notify (GObject    *gobject,
+                  GParamSpec *pspec,
+                  gpointer    user_data)
+{
+  PvClientSource *source = user_data;
+  PvClientSourcePrivate *priv = source->priv;
+  GSocket *socket;
+  guint num_handles;
+
+  g_object_get (gobject, "socket", &socket, NULL);
+
+  g_print ("source socket %p\n", socket);
+
+  if (socket == NULL) {
+    if (priv->socket)
+      g_signal_emit_by_name (priv->sink, "remove", priv->socket);
+  } else {
+    g_signal_emit_by_name (priv->sink, "add", socket);
+  }
+  priv->socket = socket;
+
+  g_object_get (priv->sink, "num-handles", &num_handles, NULL);
+  if (num_handles == 0) {
+    gst_element_set_state (priv->pipeline, GST_STATE_READY);
+  } else {
+    gst_element_set_state (priv->pipeline, GST_STATE_PLAYING);
+  }
+}
+
+static PvSourceOutput *
+client_create_source_output (PvSource *source, GVariant *props, const gchar *prefix)
+{
+  PvClientSourcePrivate *priv = PV_CLIENT_SOURCE (source)->priv;
+  PvSourceOutput *output;
+  GVariantDict dict;
+  GstCaps *caps;
+  const gchar *str;
+  gint32 i32;
+
+  g_variant_dict_init (&dict, props);
+  if (!g_variant_dict_lookup (&dict, "format.encoding", "&s", &str))
+    goto invalid_encoding;
+
+  caps = gst_caps_new_empty_simple (str);
+
+  if (g_variant_dict_lookup (&dict, "format.format", "&s", &str))
+    gst_caps_set_simple (caps, "format", G_TYPE_STRING, str, NULL);
+  if (g_variant_dict_lookup (&dict, "format.width", "i", &i32))
+    gst_caps_set_simple (caps, "width", G_TYPE_INT, (gint) i32, NULL);
+  if (g_variant_dict_lookup (&dict, "format.height", "i", &i32))
+    gst_caps_set_simple (caps, "height", G_TYPE_INT, (gint) i32, NULL);
+  if (g_variant_dict_lookup (&dict, "format.views", "i", &i32))
+    gst_caps_set_simple (caps, "views", G_TYPE_INT, (gint) i32, NULL);
+  if (g_variant_dict_lookup (&dict, "format.chroma-site", "&s", &str))
+    gst_caps_set_simple (caps, "chroma-site", G_TYPE_STRING, str, NULL);
+  if (g_variant_dict_lookup (&dict, "format.colorimetry", "&s", &str))
+    gst_caps_set_simple (caps, "colorimetry", G_TYPE_STRING, str, NULL);
+  if (g_variant_dict_lookup (&dict, "format.interlace-mode", "&s", &str))
+    gst_caps_set_simple (caps, "interlace-mode", G_TYPE_STRING, str, NULL);
+
+  g_print ("caps %s\n", gst_caps_to_string (caps));
+
+  g_object_set (priv->filter, "caps", caps, NULL);
+  gst_caps_unref (caps);
+
+  output = PV_SOURCE_CLASS (pv_client_source_parent_class)->create_source_output (source, props, prefix);
+
+  gst_element_set_state (priv->pipeline, GST_STATE_READY);
+
+  g_signal_connect (output, "notify::socket", (GCallback) on_socket_notify, source);
+
+  return output;
+
+  /* ERRORS */
+invalid_encoding:
+  {
+    g_variant_dict_clear (&dict);
+    return NULL;
+  }
+}
+
+static gboolean
+client_release_source_output  (PvSource *source, PvSourceOutput *output)
+{
+  return PV_SOURCE_CLASS (pv_client_source_parent_class)->release_source_output (source, output);
+}
+
+static void
+client_source_finalize (GObject * object)
+{
+  G_OBJECT_CLASS (pv_client_source_parent_class)->finalize (object);
+}
+
+
+static void
+on_input_socket_notify (GObject    *gobject,
+                        GParamSpec *pspec,
+                        gpointer    user_data)
+{
+  PvClientSource *source = user_data;
+  PvClientSourcePrivate *priv = source->priv;
+  GSocket *socket;
+
+  g_object_get (gobject, "socket", &socket, NULL);
+  g_print ("input socket %p\n", socket);
+  g_object_set (priv->src, "socket", socket, NULL);
+}
+
+PvSourceOutput *
+pv_client_source_get_source_input (PvClientSource *source, GVariant *props, const gchar *prefix)
+{
+  PvClientSourcePrivate *priv;
+
+  g_return_val_if_fail (PV_IS_CLIENT_SOURCE (source), NULL);
+  priv = source->priv;
+
+  if (priv->input == NULL) {
+    priv->input = PV_SOURCE_CLASS (pv_client_source_parent_class)->create_source_output (PV_SOURCE (source), props, prefix);
+    g_signal_connect (priv->input, "notify::socket", (GCallback) on_input_socket_notify, source);
+  }
+  return priv->input;
+}
+
+static void
+pv_client_source_class_init (PvClientSourceClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+  PvSourceClass *source_class = PV_SOURCE_CLASS (klass);
+
+  g_type_class_add_private (klass, sizeof (PvClientSourcePrivate));
+
+  gobject_class->finalize = client_source_finalize;
+
+  source_class->get_capabilities = client_get_capabilities;
+  source_class->set_state = client_set_state;
+  source_class->create_source_output = client_create_source_output;
+  source_class->release_source_output = client_release_source_output;
+}
+
+static void
+pv_client_source_init (PvClientSource * source)
+{
+  source->priv = PV_CLIENT_SOURCE_GET_PRIVATE (source);
+
+  setup_pipeline (source);
+}
+
+PvSource *
+pv_client_source_new (PvDaemon *daemon)
+{
+  return g_object_new (PV_TYPE_CLIENT_SOURCE, "daemon", daemon, "name", "client-source", NULL);
+}

src/server/pv-client-source.h

@@ -0,0 +1,75 @@
+/* Pulsevideo
+ * 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 __PV_CLIENT_SOURCE_H__
+#define __PV_CLIENT_SOURCE_H__
+
+#include <glib-object.h>
+#include <gio/gio.h>
+
+G_BEGIN_DECLS
+
+typedef struct _PvClientSource PvClientSource;
+typedef struct _PvClientSourceClass PvClientSourceClass;
+typedef struct _PvClientSourcePrivate PvClientSourcePrivate;
+
+#include "server/pv-source.h"
+
+#define PV_TYPE_CLIENT_SOURCE                 (pv_client_source_get_type ())
+#define PV_IS_CLIENT_SOURCE(obj)              (G_TYPE_CHECK_INSTANCE_TYPE ((obj), PV_TYPE_SOURCE))
+#define PV_IS_CLIENT_SOURCE_CLASS(klass)      (G_TYPE_CHECK_CLASS_TYPE ((klass), PV_TYPE_SOURCE))
+#define PV_CLIENT_SOURCE_GET_CLASS(obj)       (G_TYPE_INSTANCE_GET_CLASS ((obj), PV_TYPE_SOURCE, PvClientSourceClass))
+#define PV_CLIENT_SOURCE(obj)                 (G_TYPE_CHECK_INSTANCE_CAST ((obj), PV_TYPE_SOURCE, PvClientSource))
+#define PV_CLIENT_SOURCE_CLASS(klass)         (G_TYPE_CHECK_CLASS_CAST ((klass), PV_TYPE_SOURCE, PvClientSourceClass))
+#define PV_CLIENT_SOURCE_CAST(obj)            ((PvClientSource*)(obj))
+#define PV_CLIENT_SOURCE_CLASS_CAST(klass)    ((PvClientSourceClass*)(klass))
+
+/**
+ * PvClientSource:
+ *
+ * Pulsevideo client source object class.
+ */
+struct _PvClientSource {
+  PvSource object;
+
+  PvClientSourcePrivate *priv;
+};
+
+/**
+ * PvClientSourceClass:
+ *
+ * Pulsevideo client source object class.
+ */
+struct _PvClientSourceClass {
+  PvSourceClass parent_class;
+};
+
+/* normal GObject stuff */
+GType            pv_client_source_get_type              (void);
+
+PvSource *       pv_client_source_new                   (PvDaemon *daemon);
+
+PvSourceOutput * pv_client_source_get_source_input      (PvClientSource *source,
+                                                         GVariant *props,
+                                                         const gchar *prefix);
+
+G_END_DECLS
+
+#endif /* __PV_CLIENT_SOURCE_H__ */
+

src/server/pv-client.c

@@ -22,6 +22,7 @@
 #include "client/pv-enumtypes.h"
 
 #include "server/pv-client.h"
+#include "server/pv-client-source.h"
 
 #include "dbus/org-pulsevideo.h"
 
@@ -138,7 +139,7 @@ handle_create_source_output (PvClient1              *interface,
 no_source:
   {
     g_dbus_method_invocation_return_dbus_error (invocation,
-        "org.pulsevideo.Error", "Can't create sourc");
+        "org.pulsevideo.Error", "Can't create source");
     return TRUE;
   }
 no_output:
@@ -149,6 +150,53 @@ no_output:
   }
 }
 
+static gboolean
+handle_create_source_input (PvClient1              *interface,
+                            GDBusMethodInvocation  *invocation,
+                            GVariant               *arg_properties,
+                            gpointer                user_data)
+{
+  PvClient *client = user_data;
+  PvClientPrivate *priv = client->priv;
+  PvSource *source;
+  PvSourceOutput *input;
+  const gchar *source_input_path, *sender;
+
+  source = pv_client_source_new (priv->daemon);
+  if (source == NULL)
+    goto no_source;
+
+  sender = g_dbus_method_invocation_get_sender (invocation);
+
+  pv_daemon_track_object (priv->daemon, sender, G_OBJECT (source));
+
+  input = pv_client_source_get_source_input (PV_CLIENT_SOURCE (source), arg_properties, priv->object_path);
+  if (input == NULL)
+    goto no_input;
+
+  pv_daemon_track_object (priv->daemon, sender, G_OBJECT (input));
+
+  source_input_path = pv_source_output_get_object_path (input);
+  g_dbus_method_invocation_return_value (invocation,
+                                         g_variant_new ("(o)",
+                                           source_input_path));
+
+  return TRUE;
+
+  /* ERRORS */
+no_source:
+  {
+    g_dbus_method_invocation_return_dbus_error (invocation,
+        "org.pulsevideo.Error", "Can't create source");
+    return TRUE;
+  }
+no_input:
+  {
+    g_dbus_method_invocation_return_dbus_error (invocation,
+        "org.pulsevideo.Error", "Can't create input");
+    return TRUE;
+  }
+}
 
 static void
 client_register_object (PvClient *client, const gchar *prefix)
@@ -167,6 +215,9 @@ client_register_object (PvClient *client, const gchar *prefix)
   g_signal_connect (priv->client1, "handle-create-source-output",
                                    (GCallback) handle_create_source_output,
                                    client);
+  g_signal_connect (priv->client1, "handle-create-source-input",
+                                   (GCallback) handle_create_source_input,
+                                   client);
   pv_object_skeleton_set_client1 (skel, priv->client1);
 
   g_free (priv->object_path);

src/server/pv-source.c

@@ -401,3 +401,14 @@ pv_source_release_source_output (PvSource *source, PvSourceOutput *output)
   return res;
 }
 
+const gchar *
+pv_source_get_object_path (PvSource *source)
+{
+  PvSourcePrivate *priv;
+
+  g_return_val_if_fail (PV_IS_SOURCE (source), NULL);
+  priv = source->priv;
+
+ return priv->object_path;
+}
+

src/server/pv-source.h

@@ -75,6 +75,8 @@ struct _PvSourceClass {
 /* normal GObject stuff */
 GType            pv_source_get_type               (void);
 
+const gchar *    pv_source_get_object_path        (PvSource *source);
+
 GVariant *       pv_source_get_capabilities       (PvSource *source, GVariant *props);
 
 gboolean         pv_source_set_state              (PvSource *source, PvSourceState state);