Adding upstream version 0.3.65.upstream/0.3.65upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

17 files changed, 5407 insertions, 0 deletions

diff --git a/src/gst/.editorconfig b/src/gst/.editorconfig
new file mode 100644
index 0000000..b6330b2
--- /dev/null
+++ b/src/gst/.editorconfig
@@ -0,0 +1,7 @@
+[*]
+indent_style = space
+indent_size = 2
+end_of_line = lf
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true
diff --git a/src/gst/gstpipewire.c b/src/gst/gstpipewire.c
new file mode 100644
index 0000000..b87f8a3
--- /dev/null
+++ b/src/gst/gstpipewire.c
@@ -0,0 +1,69 @@
+/* PipeWire GStreamer Elements
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+/**
+ * SECTION:element-pipewiresrc
+ *
+ * <refsect2>
+ * <title>Example launch line</title>
+ * |[
+ * gst-launch -v pipewiresrc ! ximagesink
+ * ]| Shows PipeWire output in an X window.
+ * </refsect2>
+ */
+
+#include "config.h"
+
+#include "gstpipewiresrc.h"
+#include "gstpipewiresink.h"
+#include "gstpipewiredeviceprovider.h"
+
+GST_DEBUG_CATEGORY (pipewire_debug);
+
+static gboolean
+plugin_init (GstPlugin *plugin)
+{
+  pw_init (NULL, NULL);
+
+  gst_element_register (plugin, "pipewiresrc", GST_RANK_PRIMARY + 1,
+      GST_TYPE_PIPEWIRE_SRC);
+  gst_element_register (plugin, "pipewiresink", GST_RANK_NONE,
+      GST_TYPE_PIPEWIRE_SINK);
+
+#ifdef HAVE_GSTREAMER_DEVICE_PROVIDER
+  if (!gst_device_provider_register (plugin, "pipewiredeviceprovider",
+       GST_RANK_PRIMARY + 1, GST_TYPE_PIPEWIRE_DEVICE_PROVIDER))
+    return FALSE;
+#endif
+
+  GST_DEBUG_CATEGORY_INIT (pipewire_debug, "pipewire", 0, "PipeWire elements");
+
+  return TRUE;
+}
+
+GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
+    GST_VERSION_MINOR,
+    pipewire,
+    "Uses PipeWire to handle media streams",
+    plugin_init, PACKAGE_VERSION, "MIT/X11", "pipewire", "pipewire.org")
diff --git a/src/gst/gstpipewireclock.c b/src/gst/gstpipewireclock.c
new file mode 100644
index 0000000..44b1158
--- /dev/null
+++ b/src/gst/gstpipewireclock.c
@@ -0,0 +1,127 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include "config.h"
+
+#include <gst/gst.h>
+
+#include "gstpipewireclock.h"
+
+GST_DEBUG_CATEGORY_STATIC (gst_pipewire_clock_debug_category);
+#define GST_CAT_DEFAULT gst_pipewire_clock_debug_category
+
+G_DEFINE_TYPE (GstPipeWireClock, gst_pipewire_clock, GST_TYPE_SYSTEM_CLOCK);
+
+GstClock *
+gst_pipewire_clock_new (struct pw_stream *stream, GstClockTime last_time)
+{
+  GstPipeWireClock *clock;
+
+  clock = g_object_new (GST_TYPE_PIPEWIRE_CLOCK, NULL);
+  clock->stream = stream;
+  clock->last_time = last_time;
+  clock->time_offset = last_time;
+
+  return GST_CLOCK_CAST (clock);
+}
+
+static GstClockTime
+gst_pipewire_clock_get_internal_time (GstClock * clock)
+{
+  GstPipeWireClock *pclock = (GstPipeWireClock *) clock;
+  GstClockTime result;
+  struct timespec ts;
+
+  clock_gettime(CLOCK_MONOTONIC, &ts);
+#if 0
+  struct pw_time t;
+  if (pclock->stream == NULL ||
+      pw_stream_get_time (pclock->stream, &t) < 0 ||
+      t.rate.denom == 0)
+    return pclock->last_time;
+
+  result = gst_util_uint64_scale_int (t.ticks, GST_SECOND * t.rate.num, t.rate.denom);
+  result += SPA_TIMESPEC_TO_NSEC(&ts) - t.now;
+
+  result += pclock->time_offset;
+  pclock->last_time = result;
+
+  GST_DEBUG ("%"PRId64", %d/%d %"PRId64" %"PRId64,
+                t.ticks, t.rate.num, t.rate.denom, t.now, result);
+#else
+  result = SPA_TIMESPEC_TO_NSEC(&ts);
+  result += pclock->time_offset;
+  pclock->last_time = result;
+#endif
+
+  return result;
+}
+
+static void
+gst_pipewire_clock_finalize (GObject * object)
+{
+  GstPipeWireClock *clock = GST_PIPEWIRE_CLOCK (object);
+
+  GST_DEBUG_OBJECT (clock, "finalize");
+
+  G_OBJECT_CLASS (gst_pipewire_clock_parent_class)->finalize (object);
+}
+
+static void
+gst_pipewire_clock_class_init (GstPipeWireClockClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+  GstClockClass *gstclock_class = GST_CLOCK_CLASS (klass);
+
+  gobject_class->finalize = gst_pipewire_clock_finalize;
+
+  gstclock_class->get_internal_time = gst_pipewire_clock_get_internal_time;
+
+  GST_DEBUG_CATEGORY_INIT (gst_pipewire_clock_debug_category, "pipewireclock", 0,
+      "debug category for pipewireclock object");
+}
+
+static void
+gst_pipewire_clock_init (GstPipeWireClock * clock)
+{
+  GST_OBJECT_FLAG_SET (clock, GST_CLOCK_FLAG_CAN_SET_MASTER);
+}
+
+void
+gst_pipewire_clock_reset (GstPipeWireClock * clock, GstClockTime time)
+{
+  GstClockTimeDiff time_offset;
+
+  if (clock->last_time >= time)
+    time_offset = clock->last_time - time;
+  else
+    time_offset = -(time - clock->last_time);
+
+  clock->time_offset = time_offset;
+
+  GST_DEBUG_OBJECT (clock,
+      "reset clock to %" GST_TIME_FORMAT ", last %" GST_TIME_FORMAT
+      ", offset %" GST_STIME_FORMAT, GST_TIME_ARGS (time),
+      GST_TIME_ARGS (clock->last_time), GST_STIME_ARGS (time_offset));
+}
diff --git a/src/gst/gstpipewireclock.h b/src/gst/gstpipewireclock.h
new file mode 100644
index 0000000..0a3e447
--- /dev/null
+++ b/src/gst/gstpipewireclock.h
@@ -0,0 +1,71 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef __GST_PIPEWIRE_CLOCK_H__
+#define __GST_PIPEWIRE_CLOCK_H__
+
+#include <gst/gst.h>
+
+#include <pipewire/pipewire.h>
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_PIPEWIRE_CLOCK \
+  (gst_pipewire_clock_get_type())
+#define GST_PIPEWIRE_CLOCK(obj) \
+  (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_PIPEWIRE_CLOCK,GstPipeWireClock))
+#define GST_PIPEWIRE_CLOCK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_PIPEWIRE_CLOCK,GstPipeWireClockClass))
+#define GST_IS_PIPEWIRE_CLOCK(obj) \
+  (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_PIPEWIRE_CLOCK))
+#define GST_IS_PIPEWIRE_CLOCK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PIPEWIRE_CLOCK))
+#define GST_PIPEWIRE_CLOCK_GET_CLASS(klass) \
+  (G_TYPE_INSTANCE_GET_CLASS ((klass), GST_TYPE_PIPEWIRE_CLOCK, GstPipeWireClockClass))
+
+typedef struct _GstPipeWireClock GstPipeWireClock;
+typedef struct _GstPipeWireClockClass GstPipeWireClockClass;
+
+struct _GstPipeWireClock {
+  GstSystemClock parent;
+
+  struct pw_stream *stream;
+  GstClockTime last_time;
+  GstClockTimeDiff time_offset;
+};
+
+struct _GstPipeWireClockClass {
+  GstSystemClockClass parent_class;
+};
+
+GType gst_pipewire_clock_get_type (void);
+
+GstClock *      gst_pipewire_clock_new           (struct pw_stream *stream,
+                                                  GstClockTime last_time);
+void            gst_pipewire_clock_reset         (GstPipeWireClock *clock,
+                                                  GstClockTime time);
+
+G_END_DECLS
+
+#endif /* __GST_PIPEWIRE_CLOCK_H__ */
diff --git a/src/gst/gstpipewirecore.c b/src/gst/gstpipewirecore.c
new file mode 100644
index 0000000..6910f4e
--- /dev/null
+++ b/src/gst/gstpipewirecore.c
@@ -0,0 +1,202 @@
+/* GStreamer
+ *
+ * Copyright © 2020 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include "config.h"
+#include <unistd.h>
+#include <fcntl.h>
+
+#include <spa/utils/result.h>
+
+#include "gstpipewirecore.h"
+
+/* a list of global cores indexed by fd. */
+G_LOCK_DEFINE_STATIC (cores_lock);
+static GList *cores;
+
+static void
+on_core_error(void *data, uint32_t id, int seq, int res, const char *message)
+{
+  GstPipeWireCore *core = data;
+
+  pw_log_warn("error id:%u seq:%d res:%d (%s): %s",
+          id, seq, res, spa_strerror(res), message);
+
+  if (id == PW_ID_CORE) {
+    core->last_error = res;
+  }
+  pw_thread_loop_signal(core->loop, FALSE);
+}
+
+static void on_core_done (void *data, uint32_t id, int seq)
+{
+  GstPipeWireCore * core = data;
+  if (id == PW_ID_CORE) {
+    core->last_seq = seq;
+    pw_thread_loop_signal (core->loop, FALSE);
+  }
+}
+
+static const struct pw_core_events core_events = {
+  PW_VERSION_CORE_EVENTS,
+  .done = on_core_done,
+  .error = on_core_error,
+};
+
+static GstPipeWireCore *make_core (int fd)
+{
+  GstPipeWireCore *core;
+
+  core = g_new (GstPipeWireCore, 1);
+  core->refcount = 1;
+  core->fd = fd;
+  core->loop = pw_thread_loop_new ("pipewire-main-loop", NULL);
+  core->context = pw_context_new (pw_thread_loop_get_loop(core->loop), NULL, 0);
+  core->last_seq = -1;
+  core->last_error = 0;
+  GST_DEBUG ("loop %p context %p", core->loop, core->context);
+
+  if (pw_thread_loop_start (core->loop) < 0)
+    goto mainloop_failed;
+
+  pw_thread_loop_lock (core->loop);
+
+  if (fd == -1)
+    core->core = pw_context_connect (core->context, NULL, 0);
+  else
+    core->core = pw_context_connect_fd (core->context, fcntl(fd, F_DUPFD_CLOEXEC, 3), NULL, 0);
+
+  if (core->core == NULL)
+    goto connection_failed;
+
+  pw_core_add_listener(core->core,
+                       &core->core_listener,
+                       &core_events,
+                       core);
+
+  pw_thread_loop_unlock (core->loop);
+
+  return core;
+
+mainloop_failed:
+  {
+    GST_ERROR ("error starting mainloop");
+    pw_context_destroy (core->context);
+    pw_thread_loop_destroy (core->loop);
+    g_free (core);
+    return NULL;
+  }
+connection_failed:
+  {
+    GST_ERROR ("error connect: %m");
+    pw_thread_loop_unlock (core->loop);
+    pw_context_destroy (core->context);
+    pw_thread_loop_destroy (core->loop);
+    g_free (core);
+    return NULL;
+  }
+}
+
+typedef struct {
+  int fd;
+} FindData;
+
+static gint
+core_find (GstPipeWireCore * core, FindData * data)
+{
+  /* fd's must match */
+  if (core->fd == data->fd)
+    return 0;
+  return 1;
+}
+
+GstPipeWireCore *gst_pipewire_core_get (int fd)
+{
+  GstPipeWireCore *core;
+  FindData data;
+  GList *found;
+
+  data.fd = fd;
+
+  G_LOCK (cores_lock);
+  found = g_list_find_custom (cores, &data, (GCompareFunc) core_find);
+  if (found != NULL) {
+    core = (GstPipeWireCore *) found->data;
+    core->refcount++;
+    GST_DEBUG ("found core %p", core);
+  } else {
+    core = make_core(fd);
+    if (core != NULL) {
+      GST_DEBUG ("created core %p", core);
+      /* add to list on success */
+      cores = g_list_prepend (cores, core);
+    } else {
+      GST_WARNING ("could not create core");
+    }
+  }
+  G_UNLOCK (cores_lock);
+
+  return core;
+}
+
+static void do_sync(GstPipeWireCore * core)
+{
+  struct timespec abstime;
+  core->pending_seq = pw_core_sync(core->core, 0, core->pending_seq);
+  pw_thread_loop_get_time (core->loop, &abstime,
+              GST_PIPEWIRE_DEFAULT_TIMEOUT * SPA_NSEC_PER_SEC);
+  while (true) {
+    if (core->last_seq == core->pending_seq || core->last_error < 0)
+      break;
+    if (pw_thread_loop_timed_wait_full (core->loop, &abstime) < 0)
+      break;
+  }
+}
+
+void gst_pipewire_core_release (GstPipeWireCore *core)
+{
+  gboolean zero;
+
+  G_LOCK (cores_lock);
+  core->refcount--;
+  if ((zero = (core->refcount == 0))) {
+    GST_DEBUG ("closing core %p", core);
+    /* remove from list, we can release the mutex after removing the connection
+     * from the list because after that, nobody can access the connection anymore. */
+    cores = g_list_remove (cores, core);
+  }
+  G_UNLOCK (cores_lock);
+
+  if (zero) {
+    pw_thread_loop_lock (core->loop);
+    do_sync(core);
+
+    pw_core_disconnect (core->core);
+    pw_thread_loop_unlock (core->loop);
+    pw_thread_loop_stop (core->loop);
+    pw_context_destroy (core->context);
+    pw_thread_loop_destroy (core->loop);
+
+    free(core);
+  }
+}
diff --git a/src/gst/gstpipewirecore.h b/src/gst/gstpipewirecore.h
new file mode 100644
index 0000000..c64d466
--- /dev/null
+++ b/src/gst/gstpipewirecore.h
@@ -0,0 +1,60 @@
+/* GStreamer
+ *
+ * Copyright © 2020 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef __GST_PIPEWIRE_CORE_H__
+#define __GST_PIPEWIRE_CORE_H__
+
+#include <gst/gst.h>
+
+#include <pipewire/pipewire.h>
+
+G_BEGIN_DECLS
+
+typedef struct _GstPipeWireCore GstPipeWireCore;
+
+#define GST_PIPEWIRE_DEFAULT_TIMEOUT 30
+
+/**
+ * GstPipeWireCore:
+ *
+ * Opaque data structure.
+ */
+struct _GstPipeWireCore {
+  gint refcount;
+  int fd;
+  struct pw_thread_loop *loop;
+  struct pw_context *context;
+  struct pw_core *core;
+  struct spa_hook core_listener;
+  int last_error;
+  int last_seq;
+  int pending_seq;
+};
+
+GstPipeWireCore *gst_pipewire_core_get     (int fd);
+void             gst_pipewire_core_release (GstPipeWireCore *core);
+
+G_END_DECLS
+
+#endif /* __GST_PIPEWIRE_CORE_H__ */
diff --git a/src/gst/gstpipewiredeviceprovider.c b/src/gst/gstpipewiredeviceprovider.c
new file mode 100644
index 0000000..598a7c5
--- /dev/null
+++ b/src/gst/gstpipewiredeviceprovider.c
@@ -0,0 +1,754 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include "config.h"
+
+#include <string.h>
+
+#include <spa/utils/result.h>
+#include <spa/utils/string.h>
+
+#include <gst/gst.h>
+
+#include "gstpipewireformat.h"
+#include "gstpipewiredeviceprovider.h"
+#include "gstpipewiresrc.h"
+#include "gstpipewiresink.h"
+
+GST_DEBUG_CATEGORY_EXTERN (pipewire_debug);
+#define GST_CAT_DEFAULT pipewire_debug
+
+G_DEFINE_TYPE (GstPipeWireDevice, gst_pipewire_device, GST_TYPE_DEVICE);
+
+enum
+{
+  PROP_ID = 1,
+  PROP_SERIAL,
+  PROP_FD_DEVICE,
+};
+
+static GstElement *
+gst_pipewire_device_create_element (GstDevice * device, const gchar * name)
+{
+  GstPipeWireDevice *pipewire_dev = GST_PIPEWIRE_DEVICE (device);
+  GstElement *elem;
+  gchar *serial_str;
+
+  elem = gst_element_factory_make (pipewire_dev->element, name);
+
+  serial_str = g_strdup_printf ("%"PRIu64, pipewire_dev->serial);
+  g_object_set (elem, "target-object", serial_str,
+                "fd", pipewire_dev->fd, NULL);
+  g_free (serial_str);
+
+  return elem;
+}
+
+static gboolean
+gst_pipewire_device_reconfigure_element (GstDevice * device, GstElement * element)
+{
+  GstPipeWireDevice *pipewire_dev = GST_PIPEWIRE_DEVICE (device);
+  gchar *serial_str;
+
+  if (spa_streq(pipewire_dev->element, "pipewiresrc")) {
+    if (!GST_IS_PIPEWIRE_SRC (element))
+      return FALSE;
+  } else if (spa_streq(pipewire_dev->element, "pipewiresink")) {
+    if (!GST_IS_PIPEWIRE_SINK (element))
+      return FALSE;
+  } else {
+    g_assert_not_reached ();
+  }
+
+  serial_str = g_strdup_printf ("%"PRIu64, pipewire_dev->serial);
+  g_object_set (element, "target-object", serial_str,
+                "fd", pipewire_dev->fd, NULL);
+  g_free (serial_str);
+
+  return TRUE;
+}
+
+
+static void
+gst_pipewire_device_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec)
+{
+  GstPipeWireDevice *device;
+
+  device = GST_PIPEWIRE_DEVICE_CAST (object);
+
+  switch (prop_id) {
+    case PROP_ID:
+      g_value_set_uint (value, device->id);
+      break;
+    case PROP_SERIAL:
+      g_value_set_uint64 (value, device->serial);
+      break;
+    case PROP_FD_DEVICE:
+      g_value_set_int (value, device->fd);
+      break;
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pipewire_device_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  GstPipeWireDevice *device;
+
+  device = GST_PIPEWIRE_DEVICE_CAST (object);
+
+  switch (prop_id) {
+    case PROP_ID:
+      device->id = g_value_get_uint (value);
+      break;
+    case PROP_SERIAL:
+      device->serial = g_value_get_uint64 (value);
+      break;
+    case PROP_FD_DEVICE:
+      device->fd = g_value_get_int (value);
+      break;
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pipewire_device_finalize (GObject * object)
+{
+  G_OBJECT_CLASS (gst_pipewire_device_parent_class)->finalize (object);
+}
+
+static void
+gst_pipewire_device_class_init (GstPipeWireDeviceClass * klass)
+{
+  GstDeviceClass *dev_class = GST_DEVICE_CLASS (klass);
+  GObjectClass *object_class = G_OBJECT_CLASS (klass);
+
+  dev_class->create_element = gst_pipewire_device_create_element;
+  dev_class->reconfigure_element = gst_pipewire_device_reconfigure_element;
+
+  object_class->get_property = gst_pipewire_device_get_property;
+  object_class->set_property = gst_pipewire_device_set_property;
+  object_class->finalize = gst_pipewire_device_finalize;
+
+  g_object_class_install_property (object_class, PROP_ID,
+      g_param_spec_uint ("id", "Id",
+          "The internal id of the PipeWire device", 0, G_MAXUINT32, SPA_ID_INVALID,
+          G_PARAM_STATIC_STRINGS | G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY));
+
+  g_object_class_install_property (object_class, PROP_SERIAL,
+      g_param_spec_uint64 ("serial", "Serial",
+          "The internal serial of the PipeWire device", 0, G_MAXUINT64, SPA_ID_INVALID,
+          G_PARAM_STATIC_STRINGS | G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY));
+
+  g_object_class_install_property (object_class,
+      PROP_FD_DEVICE,
+      g_param_spec_int ("fd", "Fd", "The fd to connect with", -1, G_MAXINT, -1,
+          G_PARAM_STATIC_STRINGS | G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY));
+}
+
+static void
+gst_pipewire_device_init (GstPipeWireDevice * device)
+{
+}
+
+G_DEFINE_TYPE (GstPipeWireDeviceProvider, gst_pipewire_device_provider,
+    GST_TYPE_DEVICE_PROVIDER);
+
+enum
+{
+  PROP_0,
+  PROP_CLIENT_NAME,
+  PROP_FD,
+  PROP_LAST
+};
+
+struct node_data {
+  struct spa_list link;
+  GstPipeWireDeviceProvider *self;
+  struct pw_node *proxy;
+  struct spa_hook proxy_listener;
+  uint32_t id;
+  uint64_t serial;
+  struct spa_hook node_listener;
+  struct pw_node_info *info;
+  GstCaps *caps;
+  GstDevice *dev;
+};
+
+struct port_data {
+  struct node_data *node_data;
+  struct pw_port *proxy;
+  struct spa_hook proxy_listener;
+  uint32_t id;
+  uint64_t serial;
+  struct spa_hook port_listener;
+};
+
+static struct node_data *find_node_data(struct spa_list *nodes, uint32_t id)
+{
+  struct node_data *n;
+  spa_list_for_each(n, nodes, link) {
+    if (n->id == id)
+      return n;
+  }
+  return NULL;
+}
+
+static GstDevice *
+new_node (GstPipeWireDeviceProvider *self, struct node_data *data)
+{
+  GstStructure *props;
+  const gchar *klass = NULL, *name = NULL;
+  GstPipeWireDeviceType type;
+  const struct pw_node_info *info = data->info;
+  const gchar *element = NULL;
+  GstPipeWireDevice *gstdev;
+
+  if (info->max_input_ports > 0 && info->max_output_ports == 0) {
+    type = GST_PIPEWIRE_DEVICE_TYPE_SINK;
+    element = "pipewiresink";
+  } else if (info->max_output_ports > 0 && info->max_input_ports == 0) {
+    type = GST_PIPEWIRE_DEVICE_TYPE_SOURCE;
+    element = "pipewiresrc";
+  } else {
+    return NULL;
+  }
+
+  props = gst_structure_new_empty ("pipewire-proplist");
+  if (info->props) {
+    const struct spa_dict_item *item;
+    spa_dict_for_each (item, info->props)
+      gst_structure_set (props, item->key, G_TYPE_STRING, item->value, NULL);
+
+    klass = spa_dict_lookup (info->props, PW_KEY_MEDIA_CLASS);
+    name = spa_dict_lookup (info->props, PW_KEY_NODE_DESCRIPTION);
+  }
+  if (klass == NULL)
+    klass = "unknown/unknown";
+  if (name == NULL)
+    name = "unknown";
+
+  gstdev = g_object_new (GST_TYPE_PIPEWIRE_DEVICE,
+      "display-name", name, "caps", data->caps, "device-class", klass,
+      "id", data->id, "serial", data->serial, "fd", self->fd,
+      "properties", props, NULL);
+
+  gstdev->id = data->id;
+  gstdev->serial = data->serial;
+  gstdev->type = type;
+  gstdev->element = element;
+  if (props)
+    gst_structure_free (props);
+
+  return GST_DEVICE (gstdev);
+}
+
+static void do_add_nodes(GstPipeWireDeviceProvider *self)
+{
+  struct node_data *nd;
+
+  spa_list_for_each(nd, &self->nodes, link) {
+    if (nd->dev != NULL)
+	    continue;
+    pw_log_info("add node %d", nd->id);
+    nd->dev = new_node (self, nd);
+    if (nd->dev) {
+      if(self->list_only)
+        self->devices = g_list_prepend (self->devices, gst_object_ref_sink (nd->dev));
+      else
+        gst_device_provider_device_add (GST_DEVICE_PROVIDER (self), nd->dev);
+    }
+  }
+}
+
+static void resync(GstPipeWireDeviceProvider *self)
+{
+  self->seq = pw_core_sync(self->core->core, PW_ID_CORE, self->seq);
+  pw_log_debug("resync %d", self->seq);
+}
+
+static void
+on_core_done (void *data, uint32_t id, int seq)
+{
+  GstPipeWireDeviceProvider *self = data;
+
+  pw_log_debug("check %d %d", seq, self->seq);
+  if (id == PW_ID_CORE && seq == self->seq) {
+    do_add_nodes(self);
+    self->end = true;
+    if (self->core)
+      pw_thread_loop_signal (self->core->loop, FALSE);
+  }
+}
+
+
+static void
+on_core_error(void *data, uint32_t id, int seq, int res, const char *message)
+{
+  GstPipeWireDeviceProvider *self = data;
+
+  pw_log_warn("error id:%u seq:%d res:%d (%s): %s",
+          id, seq, res, spa_strerror(res), message);
+
+  if (id == PW_ID_CORE) {
+    self->error = res;
+  }
+  pw_thread_loop_signal(self->core->loop, FALSE);
+}
+
+static const struct pw_core_events core_events = {
+  PW_VERSION_CORE_EVENTS,
+  .done = on_core_done,
+  .error = on_core_error,
+};
+
+static void port_event_info(void *data, const struct pw_port_info *info)
+{
+  struct port_data *port_data = data;
+  struct node_data *node_data = port_data->node_data;
+  uint32_t i;
+
+  pw_log_debug("%p", port_data);
+
+  if (info->change_mask & PW_PORT_CHANGE_MASK_PARAMS) {
+    for (i = 0; i < info->n_params; i++) {
+      uint32_t id = info->params[i].id;
+
+      if (id == SPA_PARAM_EnumFormat &&
+          info->params[i].flags & SPA_PARAM_INFO_READ &&
+	  node_data->caps == NULL) {
+        node_data->caps = gst_caps_new_empty ();
+        pw_port_enum_params(port_data->proxy, 0, id, 0, UINT32_MAX, NULL);
+        resync(node_data->self);
+      }
+    }
+  }
+}
+
+static void port_event_param(void *data, int seq, uint32_t id,
+                uint32_t index, uint32_t next, const struct spa_pod *param)
+{
+  struct port_data *port_data = data;
+  struct node_data *node_data = port_data->node_data;
+  GstCaps *c1;
+
+  c1 = gst_caps_from_format (param);
+  if (c1 && node_data->caps)
+      gst_caps_append (node_data->caps, c1);
+}
+
+static const struct pw_port_events port_events = {
+  PW_VERSION_PORT_EVENTS,
+  .info = port_event_info,
+  .param = port_event_param
+};
+
+static void node_event_info(void *data, const struct pw_node_info *info)
+{
+  struct node_data *node_data = data;
+  uint32_t i;
+
+  pw_log_debug("%p", node_data->proxy);
+
+  info = node_data->info = pw_node_info_update(node_data->info, info);
+
+  if (info->change_mask & PW_NODE_CHANGE_MASK_PARAMS) {
+    for (i = 0; i < info->n_params; i++) {
+      uint32_t id = info->params[i].id;
+
+      if (id == SPA_PARAM_EnumFormat &&
+          info->params[i].flags & SPA_PARAM_INFO_READ &&
+	  node_data->caps == NULL) {
+        node_data->caps = gst_caps_new_empty ();
+        pw_node_enum_params(node_data->proxy, 0, id, 0, UINT32_MAX, NULL);
+        resync(node_data->self);
+      }
+    }
+  }
+}
+
+static void node_event_param(void *data, int seq, uint32_t id,
+                uint32_t index, uint32_t next, const struct spa_pod *param)
+{
+  struct node_data *node_data = data;
+  GstCaps *c1;
+
+  c1 = gst_caps_from_format (param);
+  if (c1 && node_data->caps)
+      gst_caps_append (node_data->caps, c1);
+}
+
+static const struct pw_node_events node_events = {
+  PW_VERSION_NODE_EVENTS,
+  .info = node_event_info,
+  .param = node_event_param
+};
+
+static void
+removed_node (void *data)
+{
+  struct node_data *nd = data;
+  pw_proxy_destroy((struct pw_proxy*)nd->proxy);
+}
+
+static void
+destroy_node (void *data)
+{
+  struct node_data *nd = data;
+  GstPipeWireDeviceProvider *self = nd->self;
+  GstDeviceProvider *provider = GST_DEVICE_PROVIDER (self);
+
+  pw_log_debug("destroy %p", nd);
+
+  if (nd->dev != NULL) {
+    gst_device_provider_device_remove (provider, GST_DEVICE (nd->dev));
+  }
+  if (nd->caps)
+    gst_caps_unref(nd->caps);
+  if (nd->info)
+    pw_node_info_free(nd->info);
+
+  spa_list_remove(&nd->link);
+}
+
+static const struct pw_proxy_events proxy_node_events = {
+  PW_VERSION_PROXY_EVENTS,
+  .removed = removed_node,
+  .destroy = destroy_node,
+};
+
+static void
+removed_port (void *data)
+{
+  struct port_data *pd = data;
+  pw_proxy_destroy((struct pw_proxy*)pd->proxy);
+}
+
+static void
+destroy_port (void *data)
+{
+  struct port_data *pd = data;
+  pw_log_debug("destroy %p", pd);
+}
+
+static const struct pw_proxy_events proxy_port_events = {
+  PW_VERSION_PROXY_EVENTS,
+  .removed = removed_port,
+  .destroy = destroy_port,
+};
+
+static void registry_event_global(void *data, uint32_t id, uint32_t permissions,
+                                const char *type, uint32_t version,
+                                const struct spa_dict *props)
+{
+  GstPipeWireDeviceProvider *self = data;
+  GstDeviceProvider *provider = (GstDeviceProvider*)self;
+  struct node_data *nd;
+  const char *str;
+
+  if (spa_streq(type, PW_TYPE_INTERFACE_Node)) {
+    struct pw_node *node;
+
+    node = pw_registry_bind(self->registry,
+                    id, type, PW_VERSION_NODE, sizeof(*nd));
+    if (node == NULL)
+      goto no_mem;
+
+    if (props != NULL) {
+      str = spa_dict_lookup(props, PW_KEY_OBJECT_PATH);
+      if (str != NULL) {
+	if (g_str_has_prefix(str, "alsa:"))
+          gst_device_provider_hide_provider (provider, "pulsedeviceprovider");
+	else if (g_str_has_prefix(str, "v4l2:"))
+          gst_device_provider_hide_provider (provider, "v4l2deviceprovider");
+	else if (g_str_has_prefix(str, "libcamera:"))
+          gst_device_provider_hide_provider (provider, "libcameraprovider");
+      }
+    }
+
+    nd = pw_proxy_get_user_data((struct pw_proxy*)node);
+    nd->self = self;
+    nd->proxy = node;
+    nd->id = id;
+    if (!props || !spa_atou64(spa_dict_lookup(props, PW_KEY_OBJECT_SERIAL), &nd->serial, 0))
+      nd->serial = SPA_ID_INVALID;
+    spa_list_append(&self->nodes, &nd->link);
+    pw_node_add_listener(node, &nd->node_listener, &node_events, nd);
+    pw_proxy_add_listener((struct pw_proxy*)node, &nd->proxy_listener, &proxy_node_events, nd);
+    resync(self);
+  }
+  else if (spa_streq(type, PW_TYPE_INTERFACE_Port)) {
+    struct pw_port *port;
+    struct port_data *pd;
+
+    if ((str = spa_dict_lookup(props, PW_KEY_NODE_ID)) == NULL)
+      return;
+
+    if ((nd = find_node_data(&self->nodes, atoi(str))) == NULL)
+      return;
+
+    port = pw_registry_bind(self->registry,
+                    id, type, PW_VERSION_PORT, sizeof(*pd));
+    if (port == NULL)
+      goto no_mem;
+
+    pd = pw_proxy_get_user_data((struct pw_proxy*)port);
+    pd->node_data = nd;
+    pd->proxy = port;
+    pd->id = id;
+    if (!props || !spa_atou64(spa_dict_lookup(props, PW_KEY_OBJECT_SERIAL), &pd->serial, 0))
+      pd->serial = SPA_ID_INVALID;
+    pw_port_add_listener(port, &pd->port_listener, &port_events, pd);
+    pw_proxy_add_listener((struct pw_proxy*)port, &pd->proxy_listener, &proxy_port_events, pd);
+    resync(self);
+  }
+
+  return;
+
+no_mem:
+  GST_ERROR_OBJECT(self, "failed to create proxy");
+  return;
+}
+
+static void registry_event_global_remove(void *data, uint32_t id)
+{
+}
+
+static const struct pw_registry_events registry_events = {
+  PW_VERSION_REGISTRY_EVENTS,
+  .global = registry_event_global,
+  .global_remove = registry_event_global_remove,
+};
+
+static GList *
+gst_pipewire_device_provider_probe (GstDeviceProvider * provider)
+{
+  GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (provider);
+
+  GST_DEBUG_OBJECT (self, "starting probe");
+
+  self->core = gst_pipewire_core_get(self->fd);
+  if (self->core == NULL) {
+    GST_ERROR_OBJECT (self, "Failed to connect");
+    goto failed;
+  }
+
+  GST_DEBUG_OBJECT (self, "connected");
+
+  pw_thread_loop_lock (self->core->loop);
+
+  spa_list_init(&self->nodes);
+  spa_list_init(&self->pending);
+  self->end = FALSE;
+  self->error = 0;
+  self->list_only = TRUE;
+  self->devices = NULL;
+  self->registry = pw_core_get_registry(self->core->core, PW_VERSION_REGISTRY, 0);
+
+  pw_core_add_listener(self->core->core, &self->core_listener, &core_events, self);
+  pw_registry_add_listener(self->registry, &self->registry_listener, &registry_events, self);
+
+  resync(self);
+
+  for (;;) {
+    if (self->error < 0)
+      break;
+    if (self->end)
+      break;
+    pw_thread_loop_wait (self->core->loop);
+  }
+
+  GST_DEBUG_OBJECT (self, "disconnect");
+
+  g_clear_pointer ((struct pw_proxy**)&self->registry, pw_proxy_destroy);
+  pw_thread_loop_unlock (self->core->loop);
+  g_clear_pointer (&self->core, gst_pipewire_core_release);
+
+  return self->devices;
+
+failed:
+  return NULL;
+}
+
+static gboolean
+gst_pipewire_device_provider_start (GstDeviceProvider * provider)
+{
+  GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (provider);
+
+  GST_DEBUG_OBJECT (self, "starting provider");
+
+  self->core = gst_pipewire_core_get(self->fd);
+  if (self->core == NULL) {
+    GST_ERROR_OBJECT (self, "Failed to connect");
+    goto failed;
+  }
+
+  GST_DEBUG_OBJECT (self, "connected");
+
+  pw_thread_loop_lock (self->core->loop);
+
+  spa_list_init(&self->nodes);
+  spa_list_init(&self->pending);
+  self->end = FALSE;
+  self->error = 0;
+  self->list_only = FALSE;
+  self->registry = pw_core_get_registry(self->core->core, PW_VERSION_REGISTRY, 0);
+
+  pw_core_add_listener(self->core->core, &self->core_listener, &core_events, self);
+  pw_registry_add_listener(self->registry, &self->registry_listener, &registry_events, self);
+
+  resync(self);
+
+  for (;;) {
+    if (self->error < 0)
+      break;
+    if (self->end)
+      break;
+    pw_thread_loop_wait (self->core->loop);
+  }
+
+  GST_DEBUG_OBJECT (self, "started");
+
+  pw_thread_loop_unlock (self->core->loop);
+
+  return TRUE;
+
+failed:
+  return TRUE;
+}
+
+static void
+gst_pipewire_device_provider_stop (GstDeviceProvider * provider)
+{
+  GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (provider);
+
+  GST_DEBUG_OBJECT (self, "stopping provider");
+
+  g_clear_pointer ((struct pw_proxy**)&self->registry, pw_proxy_destroy);
+  g_clear_pointer (&self->core, gst_pipewire_core_release);
+}
+
+static void
+gst_pipewire_device_provider_set_property (GObject * object,
+    guint prop_id, const GValue * value, GParamSpec * pspec)
+{
+  GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (object);
+
+  switch (prop_id) {
+    case PROP_CLIENT_NAME:
+      g_free (self->client_name);
+      if (!g_value_get_string (value)) {
+        GST_WARNING_OBJECT (self,
+            "Empty PipeWire client name not allowed. "
+            "Resetting to default value");
+        self->client_name = g_strdup(pw_get_client_name ());
+      } else
+        self->client_name = g_value_dup_string (value);
+      break;
+
+    case PROP_FD:
+      self->fd = g_value_get_int (value);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pipewire_device_provider_get_property (GObject * object,
+    guint prop_id, GValue * value, GParamSpec * pspec)
+{
+  GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (object);
+
+  switch (prop_id) {
+    case PROP_CLIENT_NAME:
+      g_value_set_string (value, self->client_name);
+      break;
+
+    case PROP_FD:
+      g_value_set_int (value, self->fd);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pipewire_device_provider_finalize (GObject * object)
+{
+  GstPipeWireDeviceProvider *self = GST_PIPEWIRE_DEVICE_PROVIDER (object);
+
+  g_free (self->client_name);
+
+  G_OBJECT_CLASS (gst_pipewire_device_provider_parent_class)->finalize (object);
+}
+
+static void
+gst_pipewire_device_provider_class_init (GstPipeWireDeviceProviderClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+  GstDeviceProviderClass *dm_class = GST_DEVICE_PROVIDER_CLASS (klass);
+
+  gobject_class->set_property = gst_pipewire_device_provider_set_property;
+  gobject_class->get_property = gst_pipewire_device_provider_get_property;
+  gobject_class->finalize = gst_pipewire_device_provider_finalize;
+
+  dm_class->probe = gst_pipewire_device_provider_probe;
+  dm_class->start = gst_pipewire_device_provider_start;
+  dm_class->stop = gst_pipewire_device_provider_stop;
+
+  g_object_class_install_property (gobject_class,
+      PROP_CLIENT_NAME,
+      g_param_spec_string ("client-name", "Client Name",
+          "The PipeWire client_name_to_use", pw_get_client_name (),
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS |
+          GST_PARAM_MUTABLE_READY));
+
+  g_object_class_install_property (gobject_class,
+      PROP_FD,
+      g_param_spec_int ("fd", "Fd", "The fd to connect with", -1, G_MAXINT, -1,
+          G_PARAM_STATIC_STRINGS | G_PARAM_READWRITE));
+
+  gst_device_provider_class_set_static_metadata (dm_class,
+      "PipeWire Device Provider", "Sink/Source/Audio/Video",
+      "List and provide PipeWire source and sink devices",
+      "Wim Taymans <wim.taymans@gmail.com>");
+}
+
+static void
+gst_pipewire_device_provider_init (GstPipeWireDeviceProvider * self)
+{
+  self->client_name = g_strdup(pw_get_client_name ());
+  self->fd = -1;
+}
diff --git a/src/gst/gstpipewiredeviceprovider.h b/src/gst/gstpipewiredeviceprovider.h
new file mode 100644
index 0000000..c940ba4
--- /dev/null
+++ b/src/gst/gstpipewiredeviceprovider.h
@@ -0,0 +1,110 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+
+#ifndef __GST_PIPEWIRE_DEVICE_PROVIDER_H__
+#define __GST_PIPEWIRE_DEVICE_PROVIDER_H__
+
+#include "config.h"
+
+#include <gst/gst.h>
+
+#include <pipewire/pipewire.h>
+#include <gst/gstpipewirecore.h>
+
+G_BEGIN_DECLS
+
+typedef struct _GstPipeWireDevice GstPipeWireDevice;
+typedef struct _GstPipeWireDeviceClass GstPipeWireDeviceClass;
+
+#define GST_TYPE_PIPEWIRE_DEVICE                 (gst_pipewire_device_get_type())
+#define GST_IS_PIPEWIRE_DEVICE(obj)              (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_PIPEWIRE_DEVICE))
+#define GST_IS_PIPEWIRE_DEVICE_CLASS(klass)      (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_PIPEWIRE_DEVICE))
+#define GST_PIPEWIRE_DEVICE_GET_CLASS(obj)       (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_PIPEWIRE_DEVICE, GstPipeWireDeviceClass))
+#define GST_PIPEWIRE_DEVICE(obj)                 (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_PIPEWIRE_DEVICE, GstPipeWireDevice))
+#define GST_PIPEWIRE_DEVICE_CLASS(klass)         (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_DEVICE, GstPipeWireDeviceClass))
+#define GST_PIPEWIRE_DEVICE_CAST(obj)            ((GstPipeWireDevice *)(obj))
+
+typedef enum {
+  GST_PIPEWIRE_DEVICE_TYPE_UNKNOWN,
+  GST_PIPEWIRE_DEVICE_TYPE_SOURCE,
+  GST_PIPEWIRE_DEVICE_TYPE_SINK,
+} GstPipeWireDeviceType;
+
+struct _GstPipeWireDevice {
+  GstDevice           parent;
+
+  GstPipeWireDeviceType  type;
+  uint32_t            id;
+  uint64_t            serial;
+  int                 fd;
+  const gchar        *element;
+};
+
+struct _GstPipeWireDeviceClass {
+  GstDeviceClass    parent_class;
+};
+
+GType        gst_pipewire_device_get_type (void);
+
+typedef struct _GstPipeWireDeviceProvider GstPipeWireDeviceProvider;
+typedef struct _GstPipeWireDeviceProviderClass GstPipeWireDeviceProviderClass;
+
+#define GST_TYPE_PIPEWIRE_DEVICE_PROVIDER                 (gst_pipewire_device_provider_get_type())
+#define GST_IS_PIPEWIRE_DEVICE_PROVIDER(obj)              (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_PIPEWIRE_DEVICE_PROVIDER))
+#define GST_IS_PIPEWIRE_DEVICE_PROVIDER_CLASS(klass)      (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_PIPEWIRE_DEVICE_PROVIDER))
+#define GST_PIPEWIRE_DEVICE_PROVIDER_GET_CLASS(obj)       (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_PIPEWIRE_DEVICE_PROVIDER, GstPipeWireDeviceProviderClass))
+#define GST_PIPEWIRE_DEVICE_PROVIDER(obj)                 (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_PIPEWIRE_DEVICE_PROVIDER, GstPipeWireDeviceProvider))
+#define GST_PIPEWIRE_DEVICE_PROVIDER_CLASS(klass)         (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_DEVICE_PROVIDER, GstPipeWireDeviceProviderClass))
+#define GST_PIPEWIRE_DEVICE_PROVIDER_CAST(obj)            ((GstPipeWireDeviceProvider *)(obj))
+
+struct _GstPipeWireDeviceProvider {
+  GstDeviceProvider         parent;
+
+  gchar *client_name;
+  int fd;
+
+  GstPipeWireCore *core;
+  struct spa_hook core_listener;
+  struct pw_registry *registry;
+  struct spa_hook registry_listener;
+  struct spa_list nodes;
+  struct spa_list pending;
+  int seq;
+
+  int error;
+  gboolean end;
+  gboolean list_only;
+  GList *devices;
+};
+
+struct _GstPipeWireDeviceProviderClass {
+  GstDeviceProviderClass    parent_class;
+};
+
+GType        gst_pipewire_device_provider_get_type (void);
+
+G_END_DECLS
+
+#endif /* __GST_PIPEWIRE_DEVICE_PROVIDER_H__ */
diff --git a/src/gst/gstpipewireformat.c b/src/gst/gstpipewireformat.c
new file mode 100644
index 0000000..006c499
--- /dev/null
+++ b/src/gst/gstpipewireformat.c
@@ -0,0 +1,981 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include <stdio.h>
+
+#include <gst/gst.h>
+#include <gst/allocators/gstdmabuf.h>
+#include <gst/video/video.h>
+#include <gst/audio/audio.h>
+
+#include <spa/utils/string.h>
+#include <spa/utils/type.h>
+#include <spa/param/video/format-utils.h>
+#include <spa/param/audio/format-utils.h>
+#include <spa/pod/builder.h>
+
+#include "gstpipewireformat.h"
+
+#ifndef DRM_FORMAT_MOD_INVALID
+#define DRM_FORMAT_MOD_INVALID ((1ULL << 56) - 1)
+#endif
+
+#ifndef DRM_FORMAT_MOD_LINEAR
+#define DRM_FORMAT_MOD_LINEAR 0
+#endif
+
+struct media_type {
+  const char *name;
+  uint32_t media_type;
+  uint32_t media_subtype;
+};
+
+static const struct media_type media_type_map[] = {
+  { "video/x-raw", SPA_MEDIA_TYPE_video, SPA_MEDIA_SUBTYPE_raw },
+  { "audio/x-raw", SPA_MEDIA_TYPE_audio, SPA_MEDIA_SUBTYPE_raw },
+  { "image/jpeg", SPA_MEDIA_TYPE_video, SPA_MEDIA_SUBTYPE_mjpg },
+  { "video/x-jpeg", SPA_MEDIA_TYPE_video, SPA_MEDIA_SUBTYPE_mjpg },
+  { "video/x-h264", SPA_MEDIA_TYPE_video, SPA_MEDIA_SUBTYPE_h264 },
+  { "audio/x-mulaw", SPA_MEDIA_TYPE_audio, SPA_MEDIA_SUBTYPE_raw },
+  { "audio/x-alaw", SPA_MEDIA_TYPE_audio, SPA_MEDIA_SUBTYPE_raw },
+  { "audio/mpeg", SPA_MEDIA_TYPE_audio, SPA_MEDIA_SUBTYPE_mp3 },
+  { "audio/x-flac", SPA_MEDIA_TYPE_audio, SPA_MEDIA_SUBTYPE_flac },
+  { NULL, }
+};
+
+static const uint32_t video_format_map[] = {
+  SPA_VIDEO_FORMAT_UNKNOWN,
+  SPA_VIDEO_FORMAT_ENCODED,
+  SPA_VIDEO_FORMAT_I420,
+  SPA_VIDEO_FORMAT_YV12,
+  SPA_VIDEO_FORMAT_YUY2,
+  SPA_VIDEO_FORMAT_UYVY,
+  SPA_VIDEO_FORMAT_AYUV,
+  SPA_VIDEO_FORMAT_RGBx,
+  SPA_VIDEO_FORMAT_BGRx,
+  SPA_VIDEO_FORMAT_xRGB,
+  SPA_VIDEO_FORMAT_xBGR,
+  SPA_VIDEO_FORMAT_RGBA,
+  SPA_VIDEO_FORMAT_BGRA,
+  SPA_VIDEO_FORMAT_ARGB,
+  SPA_VIDEO_FORMAT_ABGR,
+  SPA_VIDEO_FORMAT_RGB,
+  SPA_VIDEO_FORMAT_BGR,
+  SPA_VIDEO_FORMAT_Y41B,
+  SPA_VIDEO_FORMAT_Y42B,
+  SPA_VIDEO_FORMAT_YVYU,
+  SPA_VIDEO_FORMAT_Y444,
+  SPA_VIDEO_FORMAT_v210,
+  SPA_VIDEO_FORMAT_v216,
+  SPA_VIDEO_FORMAT_NV12,
+  SPA_VIDEO_FORMAT_NV21,
+  SPA_VIDEO_FORMAT_GRAY8,
+  SPA_VIDEO_FORMAT_GRAY16_BE,
+  SPA_VIDEO_FORMAT_GRAY16_LE,
+  SPA_VIDEO_FORMAT_v308,
+  SPA_VIDEO_FORMAT_RGB16,
+  SPA_VIDEO_FORMAT_BGR16,
+  SPA_VIDEO_FORMAT_RGB15,
+  SPA_VIDEO_FORMAT_BGR15,
+  SPA_VIDEO_FORMAT_UYVP,
+  SPA_VIDEO_FORMAT_A420,
+  SPA_VIDEO_FORMAT_RGB8P,
+  SPA_VIDEO_FORMAT_YUV9,
+  SPA_VIDEO_FORMAT_YVU9,
+  SPA_VIDEO_FORMAT_IYU1,
+  SPA_VIDEO_FORMAT_ARGB64,
+  SPA_VIDEO_FORMAT_AYUV64,
+  SPA_VIDEO_FORMAT_r210,
+  SPA_VIDEO_FORMAT_I420_10BE,
+  SPA_VIDEO_FORMAT_I420_10LE,
+  SPA_VIDEO_FORMAT_I422_10BE,
+  SPA_VIDEO_FORMAT_I422_10LE,
+  SPA_VIDEO_FORMAT_Y444_10BE,
+  SPA_VIDEO_FORMAT_Y444_10LE,
+  SPA_VIDEO_FORMAT_GBR,
+  SPA_VIDEO_FORMAT_GBR_10BE,
+  SPA_VIDEO_FORMAT_GBR_10LE,
+  SPA_VIDEO_FORMAT_NV16,
+  SPA_VIDEO_FORMAT_NV24,
+  SPA_VIDEO_FORMAT_NV12_64Z32,
+  SPA_VIDEO_FORMAT_A420_10BE,
+  SPA_VIDEO_FORMAT_A420_10LE,
+  SPA_VIDEO_FORMAT_A422_10BE,
+  SPA_VIDEO_FORMAT_A422_10LE,
+  SPA_VIDEO_FORMAT_A444_10BE,
+  SPA_VIDEO_FORMAT_A444_10LE,
+  SPA_VIDEO_FORMAT_NV61,
+  SPA_VIDEO_FORMAT_P010_10BE,
+  SPA_VIDEO_FORMAT_P010_10LE,
+  SPA_VIDEO_FORMAT_IYU2,
+  SPA_VIDEO_FORMAT_VYUY,
+  SPA_VIDEO_FORMAT_GBRA,
+  SPA_VIDEO_FORMAT_GBRA_10BE,
+  SPA_VIDEO_FORMAT_GBRA_10LE,
+  SPA_VIDEO_FORMAT_GBR_12BE,
+  SPA_VIDEO_FORMAT_GBR_12LE,
+  SPA_VIDEO_FORMAT_GBRA_12BE,
+  SPA_VIDEO_FORMAT_GBRA_12LE,
+  SPA_VIDEO_FORMAT_I420_12BE,
+  SPA_VIDEO_FORMAT_I420_12LE,
+  SPA_VIDEO_FORMAT_I422_12BE,
+  SPA_VIDEO_FORMAT_I422_12LE,
+  SPA_VIDEO_FORMAT_Y444_12BE,
+  SPA_VIDEO_FORMAT_Y444_12LE,
+};
+
+#if __BYTE_ORDER == __BIG_ENDIAN
+#define _FORMAT_LE(fmt)  SPA_AUDIO_FORMAT_ ## fmt ## _OE
+#define _FORMAT_BE(fmt)  SPA_AUDIO_FORMAT_ ## fmt
+#elif __BYTE_ORDER == __LITTLE_ENDIAN
+#define _FORMAT_LE(fmt)  SPA_AUDIO_FORMAT_ ## fmt
+#define _FORMAT_BE(fmt)  SPA_AUDIO_FORMAT_ ## fmt ## _OE
+#endif
+
+static const uint32_t audio_format_map[] = {
+  SPA_AUDIO_FORMAT_UNKNOWN,
+  SPA_AUDIO_FORMAT_ENCODED,
+  SPA_AUDIO_FORMAT_S8,
+  SPA_AUDIO_FORMAT_U8,
+  _FORMAT_LE (S16),
+  _FORMAT_BE (S16),
+  _FORMAT_LE (U16),
+  _FORMAT_BE (U16),
+  _FORMAT_LE (S24_32),
+  _FORMAT_BE (S24_32),
+  _FORMAT_LE (U24_32),
+  _FORMAT_BE (U24_32),
+  _FORMAT_LE (S32),
+  _FORMAT_BE (S32),
+  _FORMAT_LE (U32),
+  _FORMAT_BE (U32),
+  _FORMAT_LE (S24),
+  _FORMAT_BE (S24),
+  _FORMAT_LE (U24),
+  _FORMAT_BE (U24),
+  _FORMAT_LE (S20),
+  _FORMAT_BE (S20),
+  _FORMAT_LE (U20),
+  _FORMAT_BE (U20),
+  _FORMAT_LE (S18),
+  _FORMAT_BE (S18),
+  _FORMAT_LE (U18),
+  _FORMAT_BE (U18),
+  _FORMAT_LE (F32),
+  _FORMAT_BE (F32),
+  _FORMAT_LE (F64),
+  _FORMAT_BE (F64),
+};
+
+typedef struct {
+  struct spa_pod_builder b;
+  const struct media_type *type;
+  uint32_t id;
+  const GstCapsFeatures *cf;
+  const GstStructure *cs;
+  GPtrArray *array;
+} ConvertData;
+
+static const struct media_type *
+find_media_types (const char *name)
+{
+  int i;
+  for (i = 0; media_type_map[i].name; i++) {
+    if (spa_streq(media_type_map[i].name, name))
+      return &media_type_map[i];
+  }
+  return NULL;
+}
+
+static int find_index(const uint32_t *items, int n_items, uint32_t id)
+{
+  int i;
+  for (i = 0; i < n_items; i++)
+    if (items[i] == id)
+      return i;
+  return -1;
+}
+
+static const char *
+get_nth_string (const GValue *val, int idx)
+{
+  const GValue *v = NULL;
+  GType type = G_VALUE_TYPE (val);
+
+  if (type == G_TYPE_STRING && idx == 0)
+    v = val;
+  else if (type == GST_TYPE_LIST) {
+    GArray *array = g_value_peek_pointer (val);
+    if (idx < (int)(array->len + 1)) {
+      v = &g_array_index (array, GValue, SPA_MAX (idx - 1, 0));
+    }
+  }
+  if (v)
+    return g_value_get_string (v);
+
+  return NULL;
+}
+
+static bool
+get_nth_int (const GValue *val, int idx, int *res)
+{
+  const GValue *v = NULL;
+  GType type = G_VALUE_TYPE (val);
+
+  if (type == G_TYPE_INT && idx == 0) {
+    v = val;
+  } else if (type == GST_TYPE_INT_RANGE) {
+    if (idx == 0 || idx == 1) {
+      *res = gst_value_get_int_range_min (val);
+      return true;
+    } else if (idx == 2) {
+      *res = gst_value_get_int_range_max (val);
+      return true;
+    }
+  } else if (type == GST_TYPE_LIST) {
+    GArray *array = g_value_peek_pointer (val);
+    if (idx < (int)(array->len + 1)) {
+      v = &g_array_index (array, GValue, SPA_MAX (idx - 1, 0));
+    }
+  }
+  if (v) {
+    *res = g_value_get_int (v);
+    return true;
+  }
+  return false;
+}
+
+static gboolean
+get_nth_fraction (const GValue *val, int idx, struct spa_fraction *f)
+{
+  const GValue *v = NULL;
+  GType type = G_VALUE_TYPE (val);
+
+  if (type == GST_TYPE_FRACTION && idx == 0) {
+    v = val;
+  } else if (type == GST_TYPE_FRACTION_RANGE) {
+    if (idx == 0 || idx == 1) {
+      v = gst_value_get_fraction_range_min (val);
+    } else if (idx == 2) {
+      v = gst_value_get_fraction_range_max (val);
+    }
+  } else if (type == GST_TYPE_LIST) {
+    GArray *array = g_value_peek_pointer (val);
+    if (idx < (int)(array->len + 1)) {
+      v = &g_array_index (array, GValue, SPA_MAX (idx-1, 0));
+    }
+  }
+  if (v) {
+    f->num = gst_value_get_fraction_numerator (v);
+    f->denom = gst_value_get_fraction_denominator (v);
+    return true;
+  }
+  return false;
+}
+
+static gboolean
+get_nth_rectangle (const GValue *width, const GValue *height, int idx, struct spa_rectangle *r)
+{
+  const GValue *w = NULL, *h = NULL;
+  GType wt = G_VALUE_TYPE (width);
+  GType ht = G_VALUE_TYPE (height);
+
+  if (wt == G_TYPE_INT && ht == G_TYPE_INT && idx == 0) {
+    w = width;
+    h = height;
+  } else if (wt == GST_TYPE_INT_RANGE && ht == GST_TYPE_INT_RANGE) {
+    if (idx == 0 || idx == 1) {
+      r->width = gst_value_get_int_range_min (width);
+      r->height = gst_value_get_int_range_min (height);
+      return true;
+    } else if (idx == 2) {
+      r->width = gst_value_get_int_range_max (width);
+      r->height = gst_value_get_int_range_max (height);
+      return true;
+    } else if (idx == 3) {
+      r->width = gst_value_get_int_range_step (width);
+      r->height = gst_value_get_int_range_step (height);
+      if (r->width > 1 || r->height > 1)
+        return true;
+      else
+        return false;
+    }
+  } else if (wt == GST_TYPE_LIST && ht == GST_TYPE_LIST) {
+    GArray *wa = g_value_peek_pointer (width);
+    GArray *ha = g_value_peek_pointer (height);
+    if (idx < (int)(wa->len + 1))
+      w = &g_array_index (wa, GValue, SPA_MAX (idx-1, 0));
+    if (idx < (int)(ha->len + 1))
+      h = &g_array_index (ha, GValue, SPA_MAX (idx-1, 0));
+  }
+  if (w && h) {
+    r->width = g_value_get_int (w);
+    r->height = g_value_get_int (h);
+    return true;
+  }
+  return false;
+}
+
+static uint32_t
+get_range_type (const GValue *val)
+{
+  GType type = G_VALUE_TYPE (val);
+
+  if (type == GST_TYPE_LIST)
+    return SPA_CHOICE_Enum;
+  if (type == GST_TYPE_DOUBLE_RANGE || type == GST_TYPE_FRACTION_RANGE)
+    return SPA_CHOICE_Range;
+  if (type == GST_TYPE_INT_RANGE) {
+    if (gst_value_get_int_range_step (val) == 1)
+      return SPA_CHOICE_Range;
+    else
+      return SPA_CHOICE_Step;
+  }
+  if (type == GST_TYPE_INT64_RANGE) {
+    if (gst_value_get_int64_range_step (val) == 1)
+      return SPA_CHOICE_Range;
+    else
+      return SPA_CHOICE_Step;
+  }
+  return SPA_CHOICE_None;
+}
+
+static uint32_t
+get_range_type2 (const GValue *v1, const GValue *v2)
+{
+  uint32_t r1, r2;
+
+  r1 = get_range_type (v1);
+  r2 = get_range_type (v2);
+
+  if (r1 == r2)
+    return r1;
+  if (r1 == SPA_CHOICE_Step || r2 == SPA_CHOICE_Step)
+    return SPA_CHOICE_Step;
+  if (r1 == SPA_CHOICE_Range || r2 == SPA_CHOICE_Range)
+    return SPA_CHOICE_Range;
+  return SPA_CHOICE_Range;
+}
+
+static gboolean
+handle_video_fields (ConvertData *d)
+{
+  const GValue *value, *value2;
+  int i;
+  struct spa_pod_choice *choice;
+  struct spa_pod_frame f;
+
+  value = gst_structure_get_value (d->cs, "format");
+  if (value) {
+    const char *v;
+    int idx;
+    for (i = 0; (v = get_nth_string (value, i)); i++) {
+      if (i == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_VIDEO_format, 0);
+        spa_pod_builder_push_choice(&d->b, &f, get_range_type (value), 0);
+      }
+
+      idx = gst_video_format_from_string (v);
+      if (idx != GST_VIDEO_FORMAT_UNKNOWN && idx < (int)SPA_N_ELEMENTS (video_format_map))
+        spa_pod_builder_id (&d->b, video_format_map[idx]);
+    }
+    if (i > 0) {
+      choice = spa_pod_builder_pop(&d->b, &f);
+      if (i == 1)
+        choice->body.type = SPA_CHOICE_None;
+    }
+  }
+  value = gst_structure_get_value (d->cs, "width");
+  value2 = gst_structure_get_value (d->cs, "height");
+  if (value && value2) {
+    struct spa_rectangle v;
+    for (i = 0; get_nth_rectangle (value, value2, i, &v); i++) {
+      if (i == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_VIDEO_size, 0);
+        spa_pod_builder_push_choice(&d->b, &f, get_range_type2 (value, value2), 0);
+      }
+
+      spa_pod_builder_rectangle (&d->b, v.width, v.height);
+    }
+    if (i > 0) {
+      choice = spa_pod_builder_pop(&d->b, &f);
+      if (i == 1)
+        choice->body.type = SPA_CHOICE_None;
+    }
+  }
+
+  value = gst_structure_get_value (d->cs, "framerate");
+  if (value) {
+    struct spa_fraction v;
+    for (i = 0; get_nth_fraction (value, i, &v); i++) {
+      if (i == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_VIDEO_framerate, 0);
+        spa_pod_builder_push_choice(&d->b, &f, get_range_type (value), 0);
+      }
+
+      spa_pod_builder_fraction (&d->b, v.num, v.denom);
+    }
+    if (i > 0) {
+      choice = spa_pod_builder_pop(&d->b, &f);
+      if (i == 1)
+        choice->body.type = SPA_CHOICE_None;
+    }
+  }
+
+  value = gst_structure_get_value (d->cs, "max-framerate");
+  if (value) {
+    struct spa_fraction v;
+    for (i = 0; get_nth_fraction (value, i, &v); i++) {
+      if (i == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_VIDEO_maxFramerate, 0);
+        spa_pod_builder_push_choice(&d->b, &f, get_range_type (value), 0);
+      }
+
+      spa_pod_builder_fraction (&d->b, v.num, v.denom);
+    }
+    if (i > 0) {
+      choice = spa_pod_builder_pop(&d->b, &f);
+      if (i == 1)
+        choice->body.type = SPA_CHOICE_None;
+    }
+  }
+  return TRUE;
+}
+
+static void
+set_default_channels (struct spa_pod_builder *b, uint32_t channels)
+{
+  uint32_t position[SPA_AUDIO_MAX_CHANNELS] = {0};
+  gboolean ok = TRUE;
+
+  switch (channels) {
+  case 8:
+    position[6] = SPA_AUDIO_CHANNEL_SL;
+    position[7] = SPA_AUDIO_CHANNEL_SR;
+    SPA_FALLTHROUGH
+  case 6:
+    position[5] = SPA_AUDIO_CHANNEL_LFE;
+    SPA_FALLTHROUGH
+  case 5:
+    position[4] = SPA_AUDIO_CHANNEL_FC;
+    SPA_FALLTHROUGH
+  case 4:
+    position[2] = SPA_AUDIO_CHANNEL_RL;
+    position[3] = SPA_AUDIO_CHANNEL_RR;
+    SPA_FALLTHROUGH
+  case 2:
+    position[0] = SPA_AUDIO_CHANNEL_FL;
+    position[1] = SPA_AUDIO_CHANNEL_FR;
+    break;
+  case 1:
+    position[0] = SPA_AUDIO_CHANNEL_MONO;
+    break;
+  default:
+    ok = FALSE;
+    break;
+  }
+
+  if (ok)
+    spa_pod_builder_add (b, SPA_FORMAT_AUDIO_position,
+        SPA_POD_Array(sizeof(uint32_t), SPA_TYPE_Id, channels, position), 0);
+}
+
+static gboolean
+handle_audio_fields (ConvertData *d)
+{
+  const GValue *value;
+  struct spa_pod_choice *choice;
+  struct spa_pod_frame f;
+  int i = 0;
+
+  value = gst_structure_get_value (d->cs, "format");
+  if (value) {
+    const char *v;
+    int idx;
+    for (i = 0; (v = get_nth_string (value, i)); i++) {
+      if (i == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_AUDIO_format, 0);
+        spa_pod_builder_push_choice(&d->b, &f, get_range_type (value), 0);
+      }
+
+      idx = gst_audio_format_from_string (v);
+      if (idx < (int)SPA_N_ELEMENTS (audio_format_map))
+        spa_pod_builder_id (&d->b, audio_format_map[idx]);
+    }
+    if (i > 0) {
+      choice = spa_pod_builder_pop(&d->b, &f);
+      if (i == 1)
+        choice->body.type = SPA_CHOICE_None;
+    }
+  } else if (strcmp(d->type->name, "audio/x-mulaw") == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_AUDIO_format, 0);
+        spa_pod_builder_id (&d->b, SPA_AUDIO_FORMAT_ULAW);
+  } else if (strcmp(d->type->name, "audio/x-alaw") == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_AUDIO_format, 0);
+        spa_pod_builder_id (&d->b, SPA_AUDIO_FORMAT_ALAW);
+  } else if (strcmp(d->type->name, "audio/mpeg") == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_AUDIO_format, 0);
+        spa_pod_builder_id (&d->b, SPA_AUDIO_FORMAT_ENCODED);
+  } else if (strcmp(d->type->name, "audio/x-flac") == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_AUDIO_format, 0);
+        spa_pod_builder_id (&d->b, SPA_AUDIO_FORMAT_ENCODED);
+  }
+
+#if 0
+  value = gst_structure_get_value (d->cs, "layout");
+  if (value) {
+    const char *v;
+    for (i = 0; (v = get_nth_string (value, i)); i++) {
+      enum spa_audio_layout layout;
+
+      if (spa_streq(v, "interleaved"))
+        layout = SPA_AUDIO_LAYOUT_INTERLEAVED;
+      else if (spa_streq(v, "non-interleaved"))
+        layout = SPA_AUDIO_LAYOUT_NON_INTERLEAVED;
+      else
+        break;
+
+      if (i == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_AUDIO_layout, 0);
+        spa_pod_builder_push_choice(&d->b, &f, get_range_type (value), 0);
+      }
+
+      spa_pod_builder_id (&d->b, layout);
+    }
+    if (i > 0) {
+      choice = spa_pod_builder_pop(&d->b, &f);
+      if (i == 1)
+        choice->body.type = SPA_CHOICE_None;
+    }
+  }
+#endif
+  value = gst_structure_get_value (d->cs, "rate");
+  if (value) {
+    int v;
+    for (i = 0; get_nth_int (value, i, &v); i++) {
+      if (i == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_AUDIO_rate, 0);
+        spa_pod_builder_push_choice(&d->b, &f, get_range_type (value), 0);
+      }
+
+      spa_pod_builder_int (&d->b, v);
+    }
+    if (i > 0) {
+      choice = spa_pod_builder_pop(&d->b, &f);
+      if (i == 1)
+        choice->body.type = SPA_CHOICE_None;
+    }
+  }
+  value = gst_structure_get_value (d->cs, "channels");
+  if (value) {
+    int v;
+    for (i = 0; get_nth_int (value, i, &v); i++) {
+      if (i == 0) {
+        spa_pod_builder_prop (&d->b, SPA_FORMAT_AUDIO_channels, 0);
+        spa_pod_builder_push_choice(&d->b, &f, get_range_type (value), 0);
+      }
+
+      spa_pod_builder_int (&d->b, v);
+    }
+    if (i > 0) {
+      choice = spa_pod_builder_pop(&d->b, &f);
+      if (i == 1) {
+        choice->body.type = SPA_CHOICE_None;
+        set_default_channels (&d->b, v);
+      }
+    }
+  }
+  return TRUE;
+}
+
+static int
+builder_overflow (void *event_data, uint32_t size)
+{
+  struct spa_pod_builder *b = event_data;
+  b->size = SPA_ROUND_UP_N (size, 512);
+  b->data = realloc (b->data, b->size);
+  if (b->data == NULL)
+    return -errno;
+  return 0;
+}
+
+static const struct spa_pod_builder_callbacks builder_callbacks = {
+        SPA_VERSION_POD_BUILDER_CALLBACKS,
+        .overflow = builder_overflow
+};
+
+static struct spa_pod *
+convert_1 (ConvertData *d)
+{
+  struct spa_pod_frame f;
+
+  if (!(d->type = find_media_types (gst_structure_get_name (d->cs))))
+    return NULL;
+
+  spa_pod_builder_set_callbacks(&d->b, &builder_callbacks, &d->b);
+
+  spa_pod_builder_push_object (&d->b, &f, SPA_TYPE_OBJECT_Format, d->id);
+
+  spa_pod_builder_prop (&d->b, SPA_FORMAT_mediaType, 0);
+  spa_pod_builder_id(&d->b, d->type->media_type);
+
+  spa_pod_builder_prop (&d->b, SPA_FORMAT_mediaSubtype, 0);
+  spa_pod_builder_id(&d->b, d->type->media_subtype);
+
+  if (d->cf && gst_caps_features_contains (d->cf, GST_CAPS_FEATURE_MEMORY_DMABUF)) {
+    struct spa_pod_frame f2;
+
+    spa_pod_builder_prop (&d->b, SPA_FORMAT_VIDEO_modifier,
+                          (SPA_POD_PROP_FLAG_MANDATORY | SPA_POD_PROP_FLAG_DONT_FIXATE));
+    spa_pod_builder_push_choice (&d->b, &f2, SPA_CHOICE_Enum, 0);
+    spa_pod_builder_long (&d->b, DRM_FORMAT_MOD_INVALID);
+    spa_pod_builder_long (&d->b, DRM_FORMAT_MOD_INVALID);
+    spa_pod_builder_long (&d->b, DRM_FORMAT_MOD_LINEAR);
+    spa_pod_builder_pop (&d->b, &f2);
+  }
+
+  if (d->type->media_type == SPA_MEDIA_TYPE_video)
+    handle_video_fields (d);
+  else if (d->type->media_type == SPA_MEDIA_TYPE_audio)
+    handle_audio_fields (d);
+
+  spa_pod_builder_pop (&d->b, &f);
+
+  return SPA_PTROFF (d->b.data, 0, struct spa_pod);
+}
+
+struct spa_pod *
+gst_caps_to_format (GstCaps *caps, guint index, uint32_t id)
+{
+  ConvertData d;
+  struct spa_pod *res;
+
+  g_return_val_if_fail (GST_IS_CAPS (caps), NULL);
+  g_return_val_if_fail (gst_caps_is_fixed (caps), NULL);
+
+  spa_zero (d);
+  d.cf = gst_caps_get_features (caps, index);
+  d.cs = gst_caps_get_structure (caps, index);
+  d.id = id;
+
+  res = convert_1 (&d);
+
+  return res;
+}
+
+static gboolean
+foreach_func (GstCapsFeatures *features,
+              GstStructure    *structure,
+              ConvertData     *d)
+{
+  struct spa_pod *fmt;
+  int idx;
+
+  spa_zero(d->b);
+  d->cf = features;
+  d->cs = structure;
+
+  if (d->cf && gst_caps_features_contains (d->cf, GST_CAPS_FEATURE_MEMORY_DMABUF))
+    idx = 0;
+  else
+    idx = -1;
+
+  if ((fmt = convert_1 (d)))
+    g_ptr_array_insert (d->array, idx, fmt);
+
+  return TRUE;
+}
+
+
+GPtrArray *
+gst_caps_to_format_all (GstCaps *caps, uint32_t id)
+{
+  ConvertData d;
+
+  spa_zero (d);
+  d.id = id;
+  d.array = g_ptr_array_new_full (gst_caps_get_size (caps), (GDestroyNotify)g_free);
+
+  gst_caps_foreach (caps, (GstCapsForeachFunc) foreach_func, &d);
+
+  return d.array;
+}
+
+typedef const char *(*id_to_string_func)(uint32_t id);
+
+static const char *video_id_to_string(uint32_t id)
+{
+  int idx;
+  if ((idx = find_index(video_format_map, SPA_N_ELEMENTS(video_format_map), id)) == -1)
+    return NULL;
+  return gst_video_format_to_string(idx);
+}
+
+static const char *audio_id_to_string(uint32_t id)
+{
+  int idx;
+  if ((idx = find_index(audio_format_map, SPA_N_ELEMENTS(audio_format_map), id)) == -1)
+    return NULL;
+  return gst_audio_format_to_string(idx);
+}
+
+static void
+handle_id_prop (const struct spa_pod_prop *prop, const char *key, id_to_string_func func, GstCaps *res)
+{
+  const char * str;
+  struct spa_pod *val;
+  uint32_t *id;
+  uint32_t i, n_items, choice;
+
+  val = spa_pod_get_values(&prop->value, &n_items, &choice);
+  if (val->type != SPA_TYPE_Id)
+          return;
+
+  id = SPA_POD_BODY(val);
+
+  switch (choice) {
+    case SPA_CHOICE_None:
+      if (!(str = func(id[0])))
+        return;
+      gst_caps_set_simple (res, key, G_TYPE_STRING, str, NULL);
+      break;
+    case SPA_CHOICE_Enum:
+    {
+      GValue list = { 0 }, v = { 0 };
+
+      g_value_init (&list, GST_TYPE_LIST);
+      for (i = 1; i < n_items; i++) {
+        if (!(str = func(id[i])))
+          continue;
+
+        g_value_init (&v, G_TYPE_STRING);
+        g_value_set_string (&v, str);
+        gst_value_list_append_and_take_value (&list, &v);
+      }
+      gst_caps_set_value (res, key, &list);
+      g_value_unset (&list);
+      break;
+    }
+    default:
+      break;
+  }
+}
+
+static void
+handle_int_prop (const struct spa_pod_prop *prop, const char *key, GstCaps *res)
+{
+  struct spa_pod *val;
+  uint32_t *ints;
+  uint32_t i, n_items, choice;
+
+  val = spa_pod_get_values(&prop->value, &n_items, &choice);
+  if (val->type != SPA_TYPE_Int)
+          return;
+
+  ints = SPA_POD_BODY(val);
+
+  switch (choice) {
+    case SPA_CHOICE_None:
+      gst_caps_set_simple (res, key, G_TYPE_INT, ints[0], NULL);
+      break;
+    case SPA_CHOICE_Range:
+    case SPA_CHOICE_Step:
+    {
+      if (n_items < 3)
+        return;
+      gst_caps_set_simple (res, key, GST_TYPE_INT_RANGE, ints[1], ints[2], NULL);
+      break;
+    }
+    case SPA_CHOICE_Enum:
+    {
+      GValue list = { 0 }, v = { 0 };
+
+      g_value_init (&list, GST_TYPE_LIST);
+      for (i = 1; i < n_items; i++) {
+        g_value_init (&v, G_TYPE_INT);
+        g_value_set_int (&v, ints[i]);
+        gst_value_list_append_and_take_value (&list, &v);
+      }
+      gst_caps_set_value (res, key, &list);
+      g_value_unset (&list);
+      break;
+    }
+    default:
+      break;
+  }
+}
+
+static void
+handle_rect_prop (const struct spa_pod_prop *prop, const char *width, const char *height, GstCaps *res)
+{
+  struct spa_pod *val;
+  struct spa_rectangle *rect;
+  uint32_t i, n_items, choice;
+
+  val = spa_pod_get_values(&prop->value, &n_items, &choice);
+  if (val->type != SPA_TYPE_Rectangle)
+          return;
+
+  rect = SPA_POD_BODY(val);
+
+  switch (choice) {
+    case SPA_CHOICE_None:
+      gst_caps_set_simple (res, width, G_TYPE_INT, rect[0].width,
+                                height, G_TYPE_INT, rect[0].height, NULL);
+      break;
+    case SPA_CHOICE_Range:
+    case SPA_CHOICE_Step:
+    {
+      if (n_items < 3)
+        return;
+      gst_caps_set_simple (res, width, GST_TYPE_INT_RANGE, rect[1].width, rect[2].width,
+                                height, GST_TYPE_INT_RANGE, rect[1].height, rect[2].height, NULL);
+      break;
+    }
+    case SPA_CHOICE_Enum:
+    {
+      GValue l1 = { 0 }, l2 = { 0 }, v1 = { 0 }, v2 = { 0 };
+
+      g_value_init (&l1, GST_TYPE_LIST);
+      g_value_init (&l2, GST_TYPE_LIST);
+      for (i = 1; i < n_items; i++) {
+        g_value_init (&v1, G_TYPE_INT);
+        g_value_set_int (&v1, rect[i].width);
+        gst_value_list_append_and_take_value (&l1, &v1);
+
+        g_value_init (&v2, G_TYPE_INT);
+        g_value_set_int (&v2, rect[i].height);
+        gst_value_list_append_and_take_value (&l2, &v2);
+      }
+      gst_caps_set_value (res, width, &l1);
+      gst_caps_set_value (res, height, &l2);
+      g_value_unset (&l1);
+      g_value_unset (&l2);
+      break;
+    }
+    default:
+      break;
+  }
+}
+
+static void
+handle_fraction_prop (const struct spa_pod_prop *prop, const char *key, GstCaps *res)
+{
+  struct spa_pod *val;
+  struct spa_fraction *fract;
+  uint32_t i, n_items, choice;
+
+  val = spa_pod_get_values(&prop->value, &n_items, &choice);
+  if (val->type != SPA_TYPE_Fraction)
+          return;
+
+  fract = SPA_POD_BODY(val);
+
+  switch (choice) {
+    case SPA_CHOICE_None:
+      gst_caps_set_simple (res, key, GST_TYPE_FRACTION, fract[0].num, fract[0].denom, NULL);
+      break;
+    case SPA_CHOICE_Range:
+    case SPA_CHOICE_Step:
+    {
+      if (n_items < 3)
+        return;
+      gst_caps_set_simple (res, key, GST_TYPE_FRACTION_RANGE, fract[1].num, fract[1].denom,
+                                                              fract[2].num, fract[2].denom, NULL);
+      break;
+    }
+    case SPA_CHOICE_Enum:
+    {
+      GValue l1 = { 0 }, v1 = { 0 };
+
+      g_value_init (&l1, GST_TYPE_LIST);
+      for (i = 1; i < n_items; i++) {
+        g_value_init (&v1, GST_TYPE_FRACTION);
+        gst_value_set_fraction (&v1, fract[i].num, fract[i].denom);
+        gst_value_list_append_and_take_value (&l1, &v1);
+      }
+      gst_caps_set_value (res, key, &l1);
+      g_value_unset (&l1);
+      break;
+    }
+    default:
+      break;
+  }
+}
+GstCaps *
+gst_caps_from_format (const struct spa_pod *format)
+{
+  GstCaps *res = NULL;
+  uint32_t media_type, media_subtype;
+  const struct spa_pod_prop *prop = NULL;
+  const struct spa_pod_object *obj = (const struct spa_pod_object *) format;
+
+  if (spa_format_parse(format, &media_type, &media_subtype) < 0)
+    return res;
+
+  if (media_type == SPA_MEDIA_TYPE_video) {
+    if (media_subtype == SPA_MEDIA_SUBTYPE_raw) {
+      res = gst_caps_new_empty_simple ("video/x-raw");
+      if ((prop = spa_pod_object_find_prop (obj, prop, SPA_FORMAT_VIDEO_format))) {
+        handle_id_prop (prop, "format", video_id_to_string, res);
+      }
+    }
+    else if (media_subtype == SPA_MEDIA_SUBTYPE_mjpg) {
+      res = gst_caps_new_empty_simple ("image/jpeg");
+    }
+    else if (media_subtype == SPA_MEDIA_SUBTYPE_h264) {
+      res = gst_caps_new_simple ("video/x-h264",
+          "stream-format", G_TYPE_STRING, "byte-stream",
+          "alignment", G_TYPE_STRING, "au",
+          NULL);
+    } else {
+	    return NULL;
+    }
+    if ((prop = spa_pod_object_find_prop (obj, prop, SPA_FORMAT_VIDEO_size))) {
+      handle_rect_prop (prop, "width", "height", res);
+    }
+    if ((prop = spa_pod_object_find_prop (obj, prop, SPA_FORMAT_VIDEO_framerate))) {
+      handle_fraction_prop (prop, "framerate", res);
+    }
+    if ((prop = spa_pod_object_find_prop (obj, prop, SPA_FORMAT_VIDEO_maxFramerate))) {
+      handle_fraction_prop (prop, "max-framerate", res);
+    }
+  } else if (media_type == SPA_MEDIA_TYPE_audio) {
+    if (media_subtype == SPA_MEDIA_SUBTYPE_raw) {
+      res = gst_caps_new_simple ("audio/x-raw",
+          "layout", G_TYPE_STRING, "interleaved",
+          NULL);
+      if ((prop = spa_pod_object_find_prop (obj, prop, SPA_FORMAT_AUDIO_format))) {
+        handle_id_prop (prop, "format", audio_id_to_string, res);
+      }
+      if ((prop = spa_pod_object_find_prop (obj, prop, SPA_FORMAT_AUDIO_rate))) {
+        handle_int_prop (prop, "rate", res);
+      }
+      if ((prop = spa_pod_object_find_prop (obj, prop, SPA_FORMAT_AUDIO_channels))) {
+        handle_int_prop (prop, "channels", res);
+      }
+    }
+  }
+  return res;
+}
diff --git a/src/gst/gstpipewireformat.h b/src/gst/gstpipewireformat.h
new file mode 100644
index 0000000..e077986
--- /dev/null
+++ b/src/gst/gstpipewireformat.h
@@ -0,0 +1,42 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef _GST_PIPEWIRE_FORMAT_H_
+#define _GST_PIPEWIRE_FORMAT_H_
+
+#include <gst/gst.h>
+
+#include <spa/pod/pod.h>
+
+G_BEGIN_DECLS
+
+struct spa_pod * gst_caps_to_format      (GstCaps *caps,
+                                          guint index, uint32_t id);
+GPtrArray *      gst_caps_to_format_all  (GstCaps *caps, uint32_t id);
+
+GstCaps *        gst_caps_from_format    (const struct spa_pod *format);
+
+G_END_DECLS
+
+#endif
diff --git a/src/gst/gstpipewirepool.c b/src/gst/gstpipewirepool.c
new file mode 100644
index 0000000..7ecb802
--- /dev/null
+++ b/src/gst/gstpipewirepool.c
@@ -0,0 +1,276 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#include "config.h"
+
+#include <unistd.h>
+
+#include <gst/gst.h>
+
+#include <gst/allocators/gstfdmemory.h>
+#include <gst/allocators/gstdmabuf.h>
+
+#include <gst/video/gstvideometa.h>
+
+#include "gstpipewirepool.h"
+
+GST_DEBUG_CATEGORY_STATIC (gst_pipewire_pool_debug_category);
+#define GST_CAT_DEFAULT gst_pipewire_pool_debug_category
+
+G_DEFINE_TYPE (GstPipeWirePool, gst_pipewire_pool, GST_TYPE_BUFFER_POOL);
+
+enum
+{
+  ACTIVATED,
+  /* FILL ME */
+  LAST_SIGNAL
+};
+
+
+static guint pool_signals[LAST_SIGNAL] = { 0 };
+
+static GQuark pool_data_quark;
+
+GstPipeWirePool *
+gst_pipewire_pool_new (void)
+{
+  GstPipeWirePool *pool;
+
+  pool = g_object_new (GST_TYPE_PIPEWIRE_POOL, NULL);
+
+  return pool;
+}
+
+static void
+pool_data_destroy (gpointer user_data)
+{
+  GstPipeWirePoolData *data = user_data;
+
+  gst_object_unref (data->pool);
+  g_slice_free (GstPipeWirePoolData, data);
+}
+
+void gst_pipewire_pool_wrap_buffer (GstPipeWirePool *pool, struct pw_buffer *b)
+{
+  GstBuffer *buf;
+  uint32_t i;
+  GstPipeWirePoolData *data;
+
+  GST_LOG_OBJECT (pool, "wrap buffer");
+
+  data = g_slice_new (GstPipeWirePoolData);
+
+  buf = gst_buffer_new ();
+
+  for (i = 0; i < b->buffer->n_datas; i++) {
+    struct spa_data *d = &b->buffer->datas[i];
+    GstMemory *gmem = NULL;
+
+    GST_LOG_OBJECT (pool, "wrap buffer %d %d", d->mapoffset, d->maxsize);
+    if (d->type == SPA_DATA_MemFd) {
+      GST_LOG_OBJECT (pool, "memory type MemFd");
+      gmem = gst_fd_allocator_alloc (pool->fd_allocator, dup(d->fd),
+                d->mapoffset + d->maxsize, GST_FD_MEMORY_FLAG_NONE);
+      gst_memory_resize (gmem, d->mapoffset, d->maxsize);
+    }
+    else if(d->type == SPA_DATA_DmaBuf) {
+      GST_LOG_OBJECT (pool, "memory type DmaBuf");
+      gmem = gst_fd_allocator_alloc (pool->dmabuf_allocator, dup(d->fd),
+                d->mapoffset + d->maxsize, GST_FD_MEMORY_FLAG_NONE);
+      gst_memory_resize (gmem, d->mapoffset, d->maxsize);
+    }
+    else if (d->type == SPA_DATA_MemPtr) {
+      GST_LOG_OBJECT (pool, "memory type MemPtr");
+      gmem = gst_memory_new_wrapped (0, d->data, d->maxsize, 0,
+                                     d->maxsize, NULL, NULL);
+    }
+    if (gmem)
+      gst_buffer_insert_memory (buf, i, gmem);
+  }
+
+  data->pool = gst_object_ref (pool);
+  data->owner = NULL;
+  data->header = spa_buffer_find_meta_data (b->buffer, SPA_META_Header, sizeof(*data->header));
+  data->flags = GST_BUFFER_FLAGS (buf);
+  data->b = b;
+  data->buf = buf;
+  data->crop = spa_buffer_find_meta_data (b->buffer, SPA_META_VideoCrop, sizeof(*data->crop));
+  if (data->crop)
+	  gst_buffer_add_video_crop_meta(buf);
+  data->videotransform =
+    spa_buffer_find_meta_data (b->buffer, SPA_META_VideoTransform, sizeof(*data->videotransform));
+
+  gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (buf),
+                             pool_data_quark,
+                             data,
+                             pool_data_destroy);
+  b->user_data = data;
+}
+
+GstPipeWirePoolData *gst_pipewire_pool_get_data (GstBuffer *buffer)
+{
+  return gst_mini_object_get_qdata (GST_MINI_OBJECT_CAST (buffer), pool_data_quark);
+}
+
+#if 0
+gboolean
+gst_pipewire_pool_add_buffer (GstPipeWirePool *pool, GstBuffer *buffer)
+{
+  g_return_val_if_fail (GST_IS_PIPEWIRE_POOL (pool), FALSE);
+  g_return_val_if_fail (GST_IS_BUFFER (buffer), FALSE);
+
+  GST_OBJECT_LOCK (pool);
+  g_queue_push_tail (&pool->available, buffer);
+  g_cond_signal (&pool->cond);
+  GST_OBJECT_UNLOCK (pool);
+
+  return TRUE;
+}
+
+gboolean
+gst_pipewire_pool_remove_buffer (GstPipeWirePool *pool, GstBuffer *buffer)
+{
+  gboolean res;
+
+  g_return_val_if_fail (GST_IS_PIPEWIRE_POOL (pool), FALSE);
+  g_return_val_if_fail (GST_IS_BUFFER (buffer), FALSE);
+
+  GST_OBJECT_LOCK (pool);
+  res = g_queue_remove (&pool->available, buffer);
+  GST_OBJECT_UNLOCK (pool);
+
+  return res;
+}
+#endif
+
+static GstFlowReturn
+acquire_buffer (GstBufferPool * pool, GstBuffer ** buffer,
+        GstBufferPoolAcquireParams * params)
+{
+  GstPipeWirePool *p = GST_PIPEWIRE_POOL (pool);
+  GstPipeWirePoolData *data;
+  struct pw_buffer *b;
+
+  GST_OBJECT_LOCK (pool);
+  while (TRUE) {
+    if (G_UNLIKELY (GST_BUFFER_POOL_IS_FLUSHING (pool)))
+      goto flushing;
+
+    if ((b = pw_stream_dequeue_buffer(p->stream)))
+      break;
+
+    if (params && (params->flags & GST_BUFFER_POOL_ACQUIRE_FLAG_DONTWAIT))
+      goto no_more_buffers;
+
+    GST_WARNING ("queue empty");
+    g_cond_wait (&p->cond, GST_OBJECT_GET_LOCK (pool));
+  }
+
+  data = b->user_data;
+  *buffer = data->buf;
+
+  GST_OBJECT_UNLOCK (pool);
+  GST_DEBUG ("acquire buffer %p", *buffer);
+
+  return GST_FLOW_OK;
+
+flushing:
+  {
+    GST_OBJECT_UNLOCK (pool);
+    return GST_FLOW_FLUSHING;
+  }
+no_more_buffers:
+  {
+    GST_LOG_OBJECT (pool, "no more buffers");
+    GST_OBJECT_UNLOCK (pool);
+    return GST_FLOW_EOS;
+  }
+}
+
+static void
+flush_start (GstBufferPool * pool)
+{
+  GstPipeWirePool *p = GST_PIPEWIRE_POOL (pool);
+
+  GST_DEBUG ("flush start");
+  GST_OBJECT_LOCK (pool);
+  g_cond_signal (&p->cond);
+  GST_OBJECT_UNLOCK (pool);
+}
+
+static void
+release_buffer (GstBufferPool * pool, GstBuffer *buffer)
+{
+  GST_DEBUG ("release buffer %p", buffer);
+}
+
+static gboolean
+do_start (GstBufferPool * pool)
+{
+  g_signal_emit (pool, pool_signals[ACTIVATED], 0, NULL);
+  return TRUE;
+}
+
+static void
+gst_pipewire_pool_finalize (GObject * object)
+{
+  GstPipeWirePool *pool = GST_PIPEWIRE_POOL (object);
+
+  GST_DEBUG_OBJECT (pool, "finalize");
+  g_object_unref (pool->fd_allocator);
+  g_object_unref (pool->dmabuf_allocator);
+
+  G_OBJECT_CLASS (gst_pipewire_pool_parent_class)->finalize (object);
+}
+
+static void
+gst_pipewire_pool_class_init (GstPipeWirePoolClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+  GstBufferPoolClass *bufferpool_class = GST_BUFFER_POOL_CLASS (klass);
+
+  gobject_class->finalize = gst_pipewire_pool_finalize;
+
+  bufferpool_class->start = do_start;
+  bufferpool_class->flush_start = flush_start;
+  bufferpool_class->acquire_buffer = acquire_buffer;
+  bufferpool_class->release_buffer = release_buffer;
+
+  pool_signals[ACTIVATED] =
+      g_signal_new ("activated", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
+      0, NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 0, G_TYPE_NONE);
+
+  GST_DEBUG_CATEGORY_INIT (gst_pipewire_pool_debug_category, "pipewirepool", 0,
+      "debug category for pipewirepool object");
+
+  pool_data_quark = g_quark_from_static_string ("GstPipeWirePoolDataQuark");
+}
+
+static void
+gst_pipewire_pool_init (GstPipeWirePool * pool)
+{
+  pool->fd_allocator = gst_fd_allocator_new ();
+  pool->dmabuf_allocator = gst_dmabuf_allocator_new ();
+  g_cond_init (&pool->cond);
+}
diff --git a/src/gst/gstpipewirepool.h b/src/gst/gstpipewirepool.h
new file mode 100644
index 0000000..acf8106
--- /dev/null
+++ b/src/gst/gstpipewirepool.h
@@ -0,0 +1,92 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef __GST_PIPEWIRE_POOL_H__
+#define __GST_PIPEWIRE_POOL_H__
+
+#include <gst/gst.h>
+
+#include <pipewire/pipewire.h>
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_PIPEWIRE_POOL \
+  (gst_pipewire_pool_get_type())
+#define GST_PIPEWIRE_POOL(obj) \
+  (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_PIPEWIRE_POOL,GstPipeWirePool))
+#define GST_PIPEWIRE_POOL_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_PIPEWIRE_POOL,GstPipeWirePoolClass))
+#define GST_IS_PIPEWIRE_POOL(obj) \
+  (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_PIPEWIRE_POOL))
+#define GST_IS_PIPEWIRE_POOL_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PIPEWIRE_POOL))
+#define GST_PIPEWIRE_POOL_GET_CLASS(klass) \
+  (G_TYPE_INSTANCE_GET_CLASS ((klass), GST_TYPE_PIPEWIRE_POOL, GstPipeWirePoolClass))
+
+typedef struct _GstPipeWirePoolData GstPipeWirePoolData;
+typedef struct _GstPipeWirePool GstPipeWirePool;
+typedef struct _GstPipeWirePoolClass GstPipeWirePoolClass;
+
+struct _GstPipeWirePoolData {
+  GstPipeWirePool *pool;
+  void *owner;
+  struct spa_meta_header *header;
+  guint flags;
+  struct pw_buffer *b;
+  GstBuffer *buf;
+  gboolean queued;
+  struct spa_meta_region *crop;
+  struct spa_meta_videotransform *videotransform;
+};
+
+struct _GstPipeWirePool {
+  GstBufferPool parent;
+
+  struct pw_stream *stream;
+  struct pw_type *t;
+
+  GstAllocator *fd_allocator;
+  GstAllocator *dmabuf_allocator;
+
+  GCond cond;
+};
+
+struct _GstPipeWirePoolClass {
+  GstBufferPoolClass parent_class;
+};
+
+GType gst_pipewire_pool_get_type (void);
+
+GstPipeWirePool *  gst_pipewire_pool_new           (void);
+
+void gst_pipewire_pool_wrap_buffer (GstPipeWirePool *pool, struct pw_buffer *buffer);
+
+GstPipeWirePoolData *gst_pipewire_pool_get_data (GstBuffer *buffer);
+
+//gboolean        gst_pipewire_pool_add_buffer    (GstPipeWirePool *pool, GstBuffer *buffer);
+//gboolean        gst_pipewire_pool_remove_buffer (GstPipeWirePool *pool, GstBuffer *buffer);
+
+G_END_DECLS
+
+#endif /* __GST_PIPEWIRE_POOL_H__ */
diff --git a/src/gst/gstpipewiresink.c b/src/gst/gstpipewiresink.c
new file mode 100644
index 0000000..91bdfcc
--- /dev/null
+++ b/src/gst/gstpipewiresink.c
@@ -0,0 +1,924 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+/**
+ * SECTION:element-pipewiresink
+ *
+ * <refsect2>
+ * <title>Example launch line</title>
+ * |[
+ * gst-launch -v videotestsrc ! pipewiresink
+ * ]| Sends a test video source to PipeWire
+ * </refsect2>
+ */
+
+#define PW_ENABLE_DEPRECATED
+
+#include "config.h"
+#include "gstpipewiresink.h"
+
+#include <string.h>
+#include <stdlib.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/socket.h>
+
+#include <spa/pod/builder.h>
+#include <spa/utils/result.h>
+
+#include <gst/video/video.h>
+
+#include "gstpipewireformat.h"
+
+GST_DEBUG_CATEGORY_STATIC (pipewire_sink_debug);
+#define GST_CAT_DEFAULT pipewire_sink_debug
+
+#define DEFAULT_PROP_MODE GST_PIPEWIRE_SINK_MODE_DEFAULT
+
+#define MIN_BUFFERS     8u
+
+enum
+{
+  PROP_0,
+  PROP_PATH,
+  PROP_TARGET_OBJECT,
+  PROP_CLIENT_NAME,
+  PROP_CLIENT_PROPERTIES,
+  PROP_STREAM_PROPERTIES,
+  PROP_MODE,
+  PROP_FD
+};
+
+GType
+gst_pipewire_sink_mode_get_type (void)
+{
+  static gsize mode_type = 0;
+  static const GEnumValue mode[] = {
+    {GST_PIPEWIRE_SINK_MODE_DEFAULT, "GST_PIPEWIRE_SINK_MODE_DEFAULT", "default"},
+    {GST_PIPEWIRE_SINK_MODE_RENDER, "GST_PIPEWIRE_SINK_MODE_RENDER", "render"},
+    {GST_PIPEWIRE_SINK_MODE_PROVIDE, "GST_PIPEWIRE_SINK_MODE_PROVIDE", "provide"},
+    {0, NULL, NULL},
+  };
+
+  if (g_once_init_enter (&mode_type)) {
+    GType tmp =
+        g_enum_register_static ("GstPipeWireSinkMode", mode);
+    g_once_init_leave (&mode_type, tmp);
+  }
+
+  return (GType) mode_type;
+}
+
+
+static GstStaticPadTemplate gst_pipewire_sink_template =
+GST_STATIC_PAD_TEMPLATE ("sink",
+    GST_PAD_SINK,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY
+    );
+
+#define gst_pipewire_sink_parent_class parent_class
+G_DEFINE_TYPE (GstPipeWireSink, gst_pipewire_sink, GST_TYPE_BASE_SINK);
+
+static void gst_pipewire_sink_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec);
+static void gst_pipewire_sink_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec);
+
+static GstStateChangeReturn
+gst_pipewire_sink_change_state (GstElement * element, GstStateChange transition);
+
+static gboolean gst_pipewire_sink_setcaps (GstBaseSink * bsink, GstCaps * caps);
+static GstCaps *gst_pipewire_sink_sink_fixate (GstBaseSink * bsink,
+    GstCaps * caps);
+
+static GstFlowReturn gst_pipewire_sink_render (GstBaseSink * psink,
+    GstBuffer * buffer);
+static gboolean gst_pipewire_sink_start (GstBaseSink * basesink);
+static gboolean gst_pipewire_sink_stop (GstBaseSink * basesink);
+
+static void
+gst_pipewire_sink_finalize (GObject * object)
+{
+  GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (object);
+
+  g_object_unref (pwsink->pool);
+
+  if (pwsink->stream_properties)
+    gst_structure_free (pwsink->stream_properties);
+  if (pwsink->client_properties)
+    gst_structure_free (pwsink->client_properties);
+  g_free (pwsink->path);
+  g_free (pwsink->target_object);
+  g_free (pwsink->client_name);
+
+  G_OBJECT_CLASS (parent_class)->finalize (object);
+}
+
+static gboolean
+gst_pipewire_sink_propose_allocation (GstBaseSink * bsink, GstQuery * query)
+{
+  GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (bsink);
+
+  gst_query_add_allocation_pool (query, GST_BUFFER_POOL_CAST (pwsink->pool), 0, 0, 0);
+  return TRUE;
+}
+
+static void
+gst_pipewire_sink_class_init (GstPipeWireSinkClass * 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_pipewire_sink_finalize;
+  gobject_class->set_property = gst_pipewire_sink_set_property;
+  gobject_class->get_property = gst_pipewire_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_PARAM_DEPRECATED));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_TARGET_OBJECT,
+                                   g_param_spec_string ("target-object",
+                                                        "Target object",
+                                                        "The sink name/serial 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",
+                                                        "The client name to use (NULL = default)",
+                                                        NULL,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_STATIC_STRINGS));
+
+   g_object_class_install_property (gobject_class,
+                                    PROP_CLIENT_PROPERTIES,
+                                    g_param_spec_boxed ("client-properties",
+                                                        "Client properties",
+                                                        "List of PipeWire client properties",
+                                                        GST_TYPE_STRUCTURE,
+                                                        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 PipeWire stream properties",
+                                                        GST_TYPE_STRUCTURE,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_STATIC_STRINGS));
+
+   g_object_class_install_property (gobject_class,
+                                    PROP_MODE,
+                                    g_param_spec_enum ("mode",
+                                                       "Mode",
+                                                       "The mode to operate in",
+                                                        GST_TYPE_PIPEWIRE_SINK_MODE,
+                                                        DEFAULT_PROP_MODE,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_STATIC_STRINGS));
+
+   g_object_class_install_property (gobject_class,
+                                    PROP_FD,
+                                    g_param_spec_int ("fd",
+                                                      "Fd",
+                                                      "The fd to connect with",
+                                                      -1, G_MAXINT, -1,
+                                                      G_PARAM_READWRITE |
+                                                      G_PARAM_STATIC_STRINGS));
+
+  gstelement_class->change_state = gst_pipewire_sink_change_state;
+
+  gst_element_class_set_static_metadata (gstelement_class,
+      "PipeWire sink", "Sink/Video",
+      "Send video to PipeWire", "Wim Taymans <wim.taymans@gmail.com>");
+
+  gst_element_class_add_pad_template (gstelement_class,
+      gst_static_pad_template_get (&gst_pipewire_sink_template));
+
+  gstbasesink_class->set_caps = gst_pipewire_sink_setcaps;
+  gstbasesink_class->fixate = gst_pipewire_sink_sink_fixate;
+  gstbasesink_class->propose_allocation = gst_pipewire_sink_propose_allocation;
+  gstbasesink_class->start = gst_pipewire_sink_start;
+  gstbasesink_class->stop = gst_pipewire_sink_stop;
+  gstbasesink_class->render = gst_pipewire_sink_render;
+
+  GST_DEBUG_CATEGORY_INIT (pipewire_sink_debug, "pipewiresink", 0,
+      "PipeWire Sink");
+}
+
+static void
+pool_activated (GstPipeWirePool *pool, GstPipeWireSink *sink)
+{
+  GstStructure *config;
+  GstCaps *caps;
+  guint size;
+  guint min_buffers;
+  guint max_buffers;
+  const struct spa_pod *port_params[3];
+  struct spa_pod_builder b = { NULL };
+  uint8_t buffer[1024];
+  struct spa_pod_frame f;
+
+  config = gst_buffer_pool_get_config (GST_BUFFER_POOL (pool));
+  gst_buffer_pool_config_get_params (config, &caps, &size, &min_buffers, &max_buffers);
+
+  spa_pod_builder_init (&b, buffer, sizeof (buffer));
+  spa_pod_builder_push_object (&b, &f, SPA_TYPE_OBJECT_ParamBuffers, SPA_PARAM_Buffers);
+  if (size == 0)
+    spa_pod_builder_add (&b,
+        SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int(0, 0, INT32_MAX),
+        0);
+  else
+    spa_pod_builder_add (&b,
+        SPA_PARAM_BUFFERS_size, SPA_POD_CHOICE_RANGE_Int(size, size, INT32_MAX),
+        0);
+
+  spa_pod_builder_add (&b,
+      SPA_PARAM_BUFFERS_stride,  SPA_POD_CHOICE_RANGE_Int(0, 0, INT32_MAX),
+      SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(
+	      SPA_MAX(MIN_BUFFERS, min_buffers),
+	      SPA_MAX(MIN_BUFFERS, min_buffers),
+	      max_buffers ? max_buffers : INT32_MAX),
+      SPA_PARAM_BUFFERS_dataType, SPA_POD_CHOICE_FLAGS_Int(
+						(1<<SPA_DATA_MemFd) |
+						(1<<SPA_DATA_MemPtr)),
+      0);
+  port_params[0] = spa_pod_builder_pop (&b, &f);
+
+  port_params[1] = spa_pod_builder_add_object (&b,
+      SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta,
+      SPA_PARAM_META_type, SPA_POD_Int(SPA_META_Header),
+      SPA_PARAM_META_size, SPA_POD_Int(sizeof (struct spa_meta_header)));
+
+  port_params[2] = spa_pod_builder_add_object (&b,
+      SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta,
+      SPA_PARAM_META_type, SPA_POD_Int(SPA_META_VideoCrop),
+      SPA_PARAM_META_size, SPA_POD_Int(sizeof (struct spa_meta_region)));
+
+  pw_thread_loop_lock (sink->core->loop);
+  pw_stream_update_params (sink->stream, port_params, 3);
+  pw_thread_loop_unlock (sink->core->loop);
+}
+
+static void
+gst_pipewire_sink_init (GstPipeWireSink * sink)
+{
+  sink->pool =  gst_pipewire_pool_new ();
+  sink->client_name = g_strdup(pw_get_client_name());
+  sink->mode = DEFAULT_PROP_MODE;
+  sink->fd = -1;
+
+  g_signal_connect (sink->pool, "activated", G_CALLBACK (pool_activated), sink);
+}
+
+static GstCaps *
+gst_pipewire_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);
+  } else if (gst_structure_has_name (structure, "audio/mpeg")) {
+    gst_structure_fixate_field_string (structure, "format", "Encoded");
+    gst_structure_fixate_field_nearest_int (structure, "channels", 2);
+    gst_structure_fixate_field_nearest_int (structure, "rate", 44100);
+  } else if (gst_structure_has_name (structure, "audio/x-flac")) {
+    gst_structure_fixate_field_string (structure, "format", "Encoded");
+    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_pipewire_sink_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (object);
+
+  switch (prop_id) {
+    case PROP_PATH:
+      g_free (pwsink->path);
+      pwsink->path = g_value_dup_string (value);
+      break;
+
+    case PROP_TARGET_OBJECT:
+      g_free (pwsink->target_object);
+      pwsink->target_object = g_value_dup_string (value);
+      break;
+
+    case PROP_CLIENT_NAME:
+      g_free (pwsink->client_name);
+      pwsink->client_name = g_value_dup_string (value);
+      break;
+
+    case PROP_CLIENT_PROPERTIES:
+      if (pwsink->client_properties)
+        gst_structure_free (pwsink->client_properties);
+      pwsink->client_properties =
+          gst_structure_copy (gst_value_get_structure (value));
+      break;
+
+    case PROP_STREAM_PROPERTIES:
+      if (pwsink->stream_properties)
+        gst_structure_free (pwsink->stream_properties);
+      pwsink->stream_properties =
+          gst_structure_copy (gst_value_get_structure (value));
+      break;
+
+    case PROP_MODE:
+      pwsink->mode = g_value_get_enum (value);
+      break;
+
+    case PROP_FD:
+      pwsink->fd = g_value_get_int (value);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pipewire_sink_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec)
+{
+  GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (object);
+
+  switch (prop_id) {
+    case PROP_PATH:
+      g_value_set_string (value, pwsink->path);
+      break;
+
+    case PROP_TARGET_OBJECT:
+      g_value_set_string (value, pwsink->target_object);
+      break;
+
+    case PROP_CLIENT_NAME:
+      g_value_set_string (value, pwsink->client_name);
+      break;
+
+    case PROP_CLIENT_PROPERTIES:
+      gst_value_set_structure (value, pwsink->client_properties);
+      break;
+
+    case PROP_STREAM_PROPERTIES:
+      gst_value_set_structure (value, pwsink->stream_properties);
+      break;
+
+    case PROP_MODE:
+      g_value_set_enum (value, pwsink->mode);
+      break;
+
+    case PROP_FD:
+      g_value_set_int (value, pwsink->fd);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+on_add_buffer (void *_data, struct pw_buffer *b)
+{
+  GstPipeWireSink *pwsink = _data;
+  gst_pipewire_pool_wrap_buffer (pwsink->pool, b);
+}
+
+static void
+on_remove_buffer (void *_data, struct pw_buffer *b)
+{
+  GstPipeWireSink *pwsink = _data;
+  GstPipeWirePoolData *data = b->user_data;
+
+  GST_LOG_OBJECT (pwsink, "remove buffer");
+
+  gst_buffer_unref (data->buf);
+}
+
+static void
+do_send_buffer (GstPipeWireSink *pwsink, GstBuffer *buffer)
+{
+  GstPipeWirePoolData *data;
+  gboolean res;
+  guint i;
+  struct spa_buffer *b;
+
+  data = gst_pipewire_pool_get_data(buffer);
+
+  b = data->b->buffer;
+
+  if (data->header) {
+    data->header->seq = GST_BUFFER_OFFSET (buffer);
+    data->header->pts = GST_BUFFER_PTS (buffer);
+    data->header->dts_offset = GST_BUFFER_DTS (buffer);
+  }
+  if (data->crop) {
+    GstVideoCropMeta *meta = gst_buffer_get_video_crop_meta (buffer);
+    if (meta) {
+      data->crop->region.position.x = meta->x;
+      data->crop->region.position.y = meta->y;
+      data->crop->region.size.width = meta->width;
+      data->crop->region.size.height = meta->width;
+    }
+  }
+  for (i = 0; i < b->n_datas; i++) {
+    struct spa_data *d = &b->datas[i];
+    GstMemory *mem = gst_buffer_peek_memory (buffer, i);
+    d->chunk->offset = mem->offset;
+    d->chunk->size = mem->size;
+    d->chunk->stride = 0;
+  }
+
+  if ((res = pw_stream_queue_buffer (pwsink->stream, data->b)) < 0) {
+    g_warning ("can't send buffer %s", spa_strerror(res));
+  }
+}
+
+
+static void
+on_process (void *data)
+{
+  GstPipeWireSink *pwsink = data;
+  GST_DEBUG ("signal");
+  g_cond_signal (&pwsink->pool->cond);
+}
+
+static void
+on_state_changed (void *data, enum pw_stream_state old, enum pw_stream_state state, const char *error)
+{
+  GstPipeWireSink *pwsink = data;
+
+  GST_DEBUG ("got stream state %d", state);
+
+  switch (state) {
+    case PW_STREAM_STATE_UNCONNECTED:
+    case PW_STREAM_STATE_CONNECTING:
+    case PW_STREAM_STATE_PAUSED:
+      break;
+    case PW_STREAM_STATE_STREAMING:
+      if (pw_stream_is_driving (pwsink->stream))
+        pw_stream_trigger_process (pwsink->stream);
+      break;
+    case PW_STREAM_STATE_ERROR:
+      GST_ELEMENT_ERROR (pwsink, RESOURCE, FAILED,
+          ("stream error: %s", error), (NULL));
+      break;
+  }
+  pw_thread_loop_signal (pwsink->core->loop, FALSE);
+}
+
+static void
+on_param_changed (void *data, uint32_t id, const struct spa_pod *param)
+{
+  GstPipeWireSink *pwsink = data;
+
+  if (param == NULL || id != SPA_PARAM_Format)
+          return;
+
+  if (gst_buffer_pool_is_active (GST_BUFFER_POOL_CAST (pwsink->pool)))
+    pool_activated (pwsink->pool, pwsink);
+}
+
+static gboolean
+gst_pipewire_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
+{
+  GstPipeWireSink *pwsink;
+  GPtrArray *possible;
+  enum pw_stream_state state;
+  const char *error = NULL;
+  gboolean res = FALSE;
+  GstStructure *config;
+  guint size;
+  guint min_buffers;
+  guint max_buffers;
+  struct timespec abstime;
+
+  pwsink = GST_PIPEWIRE_SINK (bsink);
+
+  possible = gst_caps_to_format_all (caps, SPA_PARAM_EnumFormat);
+
+  pw_thread_loop_lock (pwsink->core->loop);
+  state = pw_stream_get_state (pwsink->stream, &error);
+
+  if (state == PW_STREAM_STATE_ERROR)
+    goto start_error;
+
+  if (state == PW_STREAM_STATE_UNCONNECTED) {
+    enum pw_stream_flags flags = 0;
+    uint32_t target_id;
+
+    if (pwsink->mode != GST_PIPEWIRE_SINK_MODE_PROVIDE)
+      flags |= PW_STREAM_FLAG_AUTOCONNECT;
+    else
+      flags |= PW_STREAM_FLAG_DRIVER;
+
+    target_id = pwsink->path ? (uint32_t)atoi(pwsink->path) : PW_ID_ANY;
+
+    if (pwsink->target_object) {
+      struct spa_dict_item items[2] = {
+        SPA_DICT_ITEM_INIT(PW_KEY_TARGET_OBJECT, pwsink->target_object),
+	/* XXX deprecated but the portal and some example apps only
+	 * provide the object id */
+        SPA_DICT_ITEM_INIT(PW_KEY_NODE_TARGET, NULL),
+      };
+      struct spa_dict dict = SPA_DICT_INIT_ARRAY(items);
+      uint64_t serial;
+
+      /* If target.object is a name, set it also to node.target */
+      if (spa_atou64(pwsink->target_object, &serial, 0)) {
+        dict.n_items = 1;
+      } else {
+        target_id = PW_ID_ANY;
+        items[1].value = pwsink->target_object;
+      }
+
+      pw_stream_update_properties (pwsink->stream, &dict);
+    }
+
+    pw_stream_connect (pwsink->stream,
+                          PW_DIRECTION_OUTPUT,
+                          target_id,
+                          flags,
+                          (const struct spa_pod **) possible->pdata,
+                          possible->len);
+
+    pw_thread_loop_get_time (pwsink->core->loop, &abstime,
+              GST_PIPEWIRE_DEFAULT_TIMEOUT * SPA_NSEC_PER_SEC);
+
+    while (TRUE) {
+      state = pw_stream_get_state (pwsink->stream, &error);
+
+      if (state >= PW_STREAM_STATE_PAUSED)
+        break;
+
+      if (state == PW_STREAM_STATE_ERROR)
+        goto start_error;
+
+      if (pw_thread_loop_timed_wait_full (pwsink->core->loop, &abstime) < 0) {
+        error = "timeout";
+        goto start_error;
+      }
+    }
+  }
+  res = TRUE;
+
+  config = gst_buffer_pool_get_config (GST_BUFFER_POOL_CAST (pwsink->pool));
+  gst_buffer_pool_config_get_params (config, NULL, &size, &min_buffers, &max_buffers);
+  gst_buffer_pool_config_set_params (config, caps, size, min_buffers, max_buffers);
+  gst_buffer_pool_set_config (GST_BUFFER_POOL_CAST (pwsink->pool), config);
+
+  pw_thread_loop_unlock (pwsink->core->loop);
+
+  pwsink->negotiated = res;
+
+  return res;
+
+start_error:
+  {
+    GST_ERROR ("could not start stream: %s", error);
+    pw_thread_loop_unlock (pwsink->core->loop);
+    g_ptr_array_unref (possible);
+    return FALSE;
+  }
+}
+
+static GstFlowReturn
+gst_pipewire_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
+{
+  GstPipeWireSink *pwsink;
+  GstFlowReturn res = GST_FLOW_OK;
+  const char *error = NULL;
+  gboolean unref_buffer = FALSE;
+
+  pwsink = GST_PIPEWIRE_SINK (bsink);
+
+  if (!pwsink->negotiated)
+    goto not_negotiated;
+
+  if (buffer->pool != GST_BUFFER_POOL_CAST (pwsink->pool) &&
+      !gst_buffer_pool_is_active (GST_BUFFER_POOL_CAST (pwsink->pool))) {
+    GstStructure *config;
+    GstCaps *caps;
+    guint size, min_buffers, max_buffers;
+
+    config = gst_buffer_pool_get_config (GST_BUFFER_POOL_CAST (pwsink->pool));
+    gst_buffer_pool_config_get_params (config, &caps, &size, &min_buffers, &max_buffers);
+
+    size = (size == 0) ? gst_buffer_get_size (buffer) : size;
+
+    gst_buffer_pool_config_set_params (config, caps, size, min_buffers, max_buffers);
+    gst_buffer_pool_set_config (GST_BUFFER_POOL_CAST (pwsink->pool), config);
+
+    gst_buffer_pool_set_active (GST_BUFFER_POOL_CAST (pwsink->pool), TRUE);
+  }
+
+  pw_thread_loop_lock (pwsink->core->loop);
+  if (pw_stream_get_state (pwsink->stream, &error) != PW_STREAM_STATE_STREAMING)
+    goto done_unlock;
+
+  if (buffer->pool != GST_BUFFER_POOL_CAST (pwsink->pool)) {
+    GstBuffer *b = NULL;
+    GstMapInfo info = { 0, };
+    GstBufferPoolAcquireParams params = { 0, };
+
+    pw_thread_loop_unlock (pwsink->core->loop);
+
+    if ((res = gst_buffer_pool_acquire_buffer (GST_BUFFER_POOL_CAST (pwsink->pool), &b, &params)) != GST_FLOW_OK)
+      goto done;
+
+    gst_buffer_map (b, &info, GST_MAP_WRITE);
+    gst_buffer_extract (buffer, 0, info.data, info.maxsize);
+    gst_buffer_unmap (b, &info);
+    gst_buffer_resize (b, 0, gst_buffer_get_size (buffer));
+    buffer = b;
+    unref_buffer = TRUE;
+
+    pw_thread_loop_lock (pwsink->core->loop);
+    if (pw_stream_get_state (pwsink->stream, &error) != PW_STREAM_STATE_STREAMING)
+      goto done_unlock;
+  }
+
+  GST_DEBUG ("push buffer");
+  do_send_buffer (pwsink, buffer);
+  if (unref_buffer)
+    gst_buffer_unref (buffer);
+
+  if (pw_stream_is_driving (pwsink->stream))
+    pw_stream_trigger_process (pwsink->stream);
+
+done_unlock:
+  pw_thread_loop_unlock (pwsink->core->loop);
+done:
+  return res;
+
+not_negotiated:
+  {
+    return GST_FLOW_NOT_NEGOTIATED;
+  }
+}
+
+static gboolean
+copy_properties (GQuark field_id,
+                 const GValue *value,
+                 gpointer user_data)
+{
+  struct pw_properties *properties = user_data;
+  GValue dst = { 0 };
+
+  if (g_value_type_transformable (G_VALUE_TYPE(value), G_TYPE_STRING)) {
+    g_value_init(&dst, G_TYPE_STRING);
+    if (g_value_transform(value, &dst)) {
+      pw_properties_set (properties,
+                         g_quark_to_string (field_id),
+                         g_value_get_string (&dst));
+    }
+    g_value_unset(&dst);
+  }
+  return TRUE;
+}
+
+static const struct pw_stream_events stream_events = {
+        PW_VERSION_STREAM_EVENTS,
+        .state_changed = on_state_changed,
+        .param_changed = on_param_changed,
+        .add_buffer = on_add_buffer,
+        .remove_buffer = on_remove_buffer,
+        .process = on_process,
+};
+
+static gboolean
+gst_pipewire_sink_start (GstBaseSink * basesink)
+{
+  GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (basesink);
+  struct pw_properties *props;
+
+  pwsink->negotiated = FALSE;
+
+  pw_thread_loop_lock (pwsink->core->loop);
+
+  props = pw_properties_new (NULL, NULL);
+  if (pwsink->client_name) {
+    pw_properties_set (props, PW_KEY_NODE_NAME, pwsink->client_name);
+    pw_properties_set (props, PW_KEY_NODE_DESCRIPTION, pwsink->client_name);
+  }
+  if (pwsink->stream_properties) {
+    gst_structure_foreach (pwsink->stream_properties, copy_properties, props);
+  }
+
+  if ((pwsink->stream = pw_stream_new (pwsink->core->core, pwsink->client_name, props)) == NULL)
+    goto no_stream;
+
+  pwsink->pool->stream = pwsink->stream;
+
+  pw_stream_add_listener(pwsink->stream,
+                         &pwsink->stream_listener,
+                         &stream_events,
+                         pwsink);
+
+  pw_thread_loop_unlock (pwsink->core->loop);
+
+  return TRUE;
+
+no_stream:
+  {
+    GST_ELEMENT_ERROR (pwsink, RESOURCE, FAILED, ("can't create stream"), (NULL));
+    pw_thread_loop_unlock (pwsink->core->loop);
+    return FALSE;
+  }
+}
+
+static gboolean
+gst_pipewire_sink_stop (GstBaseSink * basesink)
+{
+  GstPipeWireSink *pwsink = GST_PIPEWIRE_SINK (basesink);
+
+  pw_thread_loop_lock (pwsink->core->loop);
+  if (pwsink->stream) {
+    pw_stream_destroy (pwsink->stream);
+    pwsink->stream = NULL;
+    pwsink->pool->stream = NULL;
+  }
+  pw_thread_loop_unlock (pwsink->core->loop);
+
+  pwsink->negotiated = FALSE;
+
+  return TRUE;
+}
+
+static gboolean
+gst_pipewire_sink_open (GstPipeWireSink * pwsink)
+{
+  struct pw_properties *props;
+
+  GST_DEBUG_OBJECT (pwsink, "open");
+
+  pwsink->core = gst_pipewire_core_get(pwsink->fd);
+  if (pwsink->core == NULL)
+      goto connect_error;
+
+  pw_thread_loop_lock (pwsink->core->loop);
+
+  props = pw_properties_new (NULL, NULL);
+  if (pwsink->client_properties) {
+    gst_structure_foreach (pwsink->client_properties, copy_properties, props);
+    pw_core_update_properties (pwsink->core->core, &props->dict);
+  }
+  pw_properties_free(props);
+  pw_thread_loop_unlock (pwsink->core->loop);
+
+  return TRUE;
+
+  /* ERRORS */
+connect_error:
+  {
+    GST_ELEMENT_ERROR (pwsink, RESOURCE, FAILED,
+        ("Failed to connect"), (NULL));
+    return FALSE;
+  }
+}
+
+static gboolean
+gst_pipewire_sink_close (GstPipeWireSink * pwsink)
+{
+  pw_thread_loop_lock (pwsink->core->loop);
+  if (pwsink->stream) {
+    pw_stream_destroy (pwsink->stream);
+    pwsink->stream = NULL;
+  }
+  pw_thread_loop_unlock (pwsink->core->loop);
+
+  if (pwsink->core) {
+    gst_pipewire_core_release (pwsink->core);
+    pwsink->core = NULL;
+  }
+  return TRUE;
+}
+
+static GstStateChangeReturn
+gst_pipewire_sink_change_state (GstElement * element, GstStateChange transition)
+{
+  GstStateChangeReturn ret;
+  GstPipeWireSink *this = GST_PIPEWIRE_SINK_CAST (element);
+
+  switch (transition) {
+    case GST_STATE_CHANGE_NULL_TO_READY:
+      if (!gst_pipewire_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 play */
+      pw_thread_loop_lock (this->core->loop);
+      pw_stream_set_active(this->stream, true);
+      pw_thread_loop_unlock (this->core->loop);
+      gst_buffer_pool_set_flushing(GST_BUFFER_POOL_CAST(this->pool), FALSE);
+      break;
+    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
+      /* stop play ASAP by corking */
+      pw_thread_loop_lock (this->core->loop);
+      pw_stream_set_active(this->stream, false);
+      pw_thread_loop_unlock (this->core->loop);
+      gst_buffer_pool_set_flushing(GST_BUFFER_POOL_CAST(this->pool), TRUE);
+      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:
+      gst_buffer_pool_set_active(GST_BUFFER_POOL_CAST(this->pool), FALSE);
+      break;
+    case GST_STATE_CHANGE_READY_TO_NULL:
+      gst_pipewire_sink_close (this);
+      break;
+    default:
+      break;
+  }
+  return ret;
+
+  /* ERRORS */
+open_failed:
+  {
+    return GST_STATE_CHANGE_FAILURE;
+  }
+}
diff --git a/src/gst/gstpipewiresink.h b/src/gst/gstpipewiresink.h
new file mode 100644
index 0000000..703cd0f
--- /dev/null
+++ b/src/gst/gstpipewiresink.h
@@ -0,0 +1,110 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef __GST_PIPEWIRE_SINK_H__
+#define __GST_PIPEWIRE_SINK_H__
+
+#include <gst/gst.h>
+#include <gst/base/gstbasesink.h>
+
+#include <pipewire/pipewire.h>
+#include <gst/gstpipewirepool.h>
+#include <gst/gstpipewirecore.h>
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_PIPEWIRE_SINK \
+  (gst_pipewire_sink_get_type())
+#define GST_PIPEWIRE_SINK(obj) \
+  (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_PIPEWIRE_SINK,GstPipeWireSink))
+#define GST_PIPEWIRE_SINK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_PIPEWIRE_SINK,GstPipeWireSinkClass))
+#define GST_IS_PIPEWIRE_SINK(obj) \
+  (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_PIPEWIRE_SINK))
+#define GST_IS_PIPEWIRE_SINK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PIPEWIRE_SINK))
+#define GST_PIPEWIRE_SINK_CAST(obj) \
+  ((GstPipeWireSink *) (obj))
+
+typedef struct _GstPipeWireSink GstPipeWireSink;
+typedef struct _GstPipeWireSinkClass GstPipeWireSinkClass;
+
+
+/**
+ * GstPipeWireSinkMode:
+ * @GST_PIPEWIRE_SINK_MODE_DEFAULT: the default mode as configured in the server
+ * @GST_PIPEWIRE_SINK_MODE_RENDER: try to render the media
+ * @GST_PIPEWIRE_SINK_MODE_PROVIDE: provide the media
+ *
+ * Different modes of operation.
+ */
+typedef enum
+{
+  GST_PIPEWIRE_SINK_MODE_DEFAULT,
+  GST_PIPEWIRE_SINK_MODE_RENDER,
+  GST_PIPEWIRE_SINK_MODE_PROVIDE,
+} GstPipeWireSinkMode;
+
+#define GST_TYPE_PIPEWIRE_SINK_MODE (gst_pipewire_sink_mode_get_type ())
+
+/**
+ * GstPipeWireSink:
+ *
+ * Opaque data structure.
+ */
+struct _GstPipeWireSink {
+  GstBaseSink element;
+
+  /*< private >*/
+  gchar *path;
+  gchar *target_object;
+  gchar *client_name;
+  int fd;
+
+  /* video state */
+  gboolean negotiated;
+
+  GstPipeWireCore *core;
+  struct spa_hook core_listener;
+  GstStructure *client_properties;
+
+  struct pw_stream *stream;
+  struct spa_hook stream_listener;
+
+  GstStructure *stream_properties;
+  GstPipeWireSinkMode mode;
+
+  GstPipeWirePool *pool;
+};
+
+struct _GstPipeWireSinkClass {
+  GstBaseSinkClass parent_class;
+};
+
+GType gst_pipewire_sink_get_type (void);
+GType gst_pipewire_sink_mode_get_type (void);
+
+G_END_DECLS
+
+#endif /* __GST_PIPEWIRE_SINK_H__ */
diff --git a/src/gst/gstpipewiresrc.c b/src/gst/gstpipewiresrc.c
new file mode 100644
index 0000000..7243cc1
--- /dev/null
+++ b/src/gst/gstpipewiresrc.c
@@ -0,0 +1,1439 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+/**
+ * SECTION:element-pipewiresrc
+ *
+ * <refsect2>
+ * <title>Example launch line</title>
+ * |[
+ * gst-launch -v pipewiresrc ! videoconvert ! ximagesink
+ * ]| Shows pipewire output in an X window.
+ * </refsect2>
+ */
+
+#define PW_ENABLE_DEPRECATED
+
+#include "config.h"
+#include "gstpipewiresrc.h"
+#include "gstpipewireformat.h"
+
+#include <string.h>
+#include <stdlib.h>
+#include <fcntl.h>
+#include <sys/socket.h>
+#include <unistd.h>
+
+#include <spa/param/video/format.h>
+#include <spa/pod/builder.h>
+#include <spa/utils/result.h>
+
+#include <gst/net/gstnetclientclock.h>
+#include <gst/allocators/gstfdmemory.h>
+#include <gst/allocators/gstdmabuf.h>
+#include <gst/video/video.h>
+
+#include "gstpipewireclock.h"
+
+static GQuark process_mem_data_quark;
+
+GST_DEBUG_CATEGORY_STATIC (pipewire_src_debug);
+#define GST_CAT_DEFAULT pipewire_src_debug
+
+#define DEFAULT_ALWAYS_COPY     false
+#define DEFAULT_MIN_BUFFERS     8
+#define DEFAULT_MAX_BUFFERS     INT32_MAX
+#define DEFAULT_RESEND_LAST     false
+#define DEFAULT_KEEPALIVE_TIME  0
+
+enum
+{
+  PROP_0,
+  PROP_PATH,
+  PROP_TARGET_OBJECT,
+  PROP_CLIENT_NAME,
+  PROP_CLIENT_PROPERTIES,
+  PROP_STREAM_PROPERTIES,
+  PROP_ALWAYS_COPY,
+  PROP_MIN_BUFFERS,
+  PROP_MAX_BUFFERS,
+  PROP_FD,
+  PROP_RESEND_LAST,
+  PROP_KEEPALIVE_TIME,
+};
+
+
+static GstStaticPadTemplate gst_pipewire_src_template =
+GST_STATIC_PAD_TEMPLATE ("src",
+    GST_PAD_SRC,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY
+    );
+
+#define gst_pipewire_src_parent_class parent_class
+G_DEFINE_TYPE (GstPipeWireSrc, gst_pipewire_src, GST_TYPE_PUSH_SRC);
+
+static GstStateChangeReturn
+gst_pipewire_src_change_state (GstElement * element, GstStateChange transition);
+
+static gboolean gst_pipewire_src_send_event (GstElement * elem, GstEvent * event);
+
+static gboolean gst_pipewire_src_negotiate (GstBaseSrc * basesrc);
+
+static GstFlowReturn gst_pipewire_src_create (GstPushSrc * psrc,
+    GstBuffer ** buffer);
+static gboolean gst_pipewire_src_unlock (GstBaseSrc * basesrc);
+static gboolean gst_pipewire_src_unlock_stop (GstBaseSrc * basesrc);
+static gboolean gst_pipewire_src_start (GstBaseSrc * basesrc);
+static gboolean gst_pipewire_src_stop (GstBaseSrc * basesrc);
+static gboolean gst_pipewire_src_event (GstBaseSrc * src, GstEvent * event);
+static gboolean gst_pipewire_src_query (GstBaseSrc * src, GstQuery * query);
+
+static void
+gst_pipewire_src_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  GstPipeWireSrc *pwsrc = GST_PIPEWIRE_SRC (object);
+
+  switch (prop_id) {
+    case PROP_PATH:
+      g_free (pwsrc->path);
+      pwsrc->path = g_value_dup_string (value);
+      break;
+
+    case PROP_TARGET_OBJECT:
+      g_free (pwsrc->target_object);
+      pwsrc->target_object = g_value_dup_string (value);
+      break;
+
+    case PROP_CLIENT_NAME:
+      g_free (pwsrc->client_name);
+      pwsrc->client_name = g_value_dup_string (value);
+      break;
+
+    case PROP_CLIENT_PROPERTIES:
+      if (pwsrc->client_properties)
+        gst_structure_free (pwsrc->client_properties);
+      pwsrc->client_properties =
+          gst_structure_copy (gst_value_get_structure (value));
+      break;
+
+    case PROP_STREAM_PROPERTIES:
+      if (pwsrc->stream_properties)
+        gst_structure_free (pwsrc->stream_properties);
+      pwsrc->stream_properties =
+          gst_structure_copy (gst_value_get_structure (value));
+      break;
+
+    case PROP_ALWAYS_COPY:
+      pwsrc->always_copy = g_value_get_boolean (value);
+      break;
+
+    case PROP_MIN_BUFFERS:
+      pwsrc->min_buffers = g_value_get_int (value);
+      break;
+
+    case PROP_MAX_BUFFERS:
+      pwsrc->max_buffers = g_value_get_int (value);
+      break;
+
+    case PROP_FD:
+      pwsrc->fd = g_value_get_int (value);
+      break;
+
+    case PROP_RESEND_LAST:
+      pwsrc->resend_last = g_value_get_boolean (value);
+      break;
+
+    case PROP_KEEPALIVE_TIME:
+      pwsrc->keepalive_time = g_value_get_int (value);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_pipewire_src_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec)
+{
+  GstPipeWireSrc *pwsrc = GST_PIPEWIRE_SRC (object);
+
+  switch (prop_id) {
+    case PROP_PATH:
+      g_value_set_string (value, pwsrc->path);
+      break;
+
+    case PROP_TARGET_OBJECT:
+      g_value_set_string (value, pwsrc->target_object);
+      break;
+
+    case PROP_CLIENT_NAME:
+      g_value_set_string (value, pwsrc->client_name);
+      break;
+
+    case PROP_CLIENT_PROPERTIES:
+      gst_value_set_structure (value, pwsrc->client_properties);
+      break;
+
+    case PROP_STREAM_PROPERTIES:
+      gst_value_set_structure (value, pwsrc->stream_properties);
+      break;
+
+    case PROP_ALWAYS_COPY:
+      g_value_set_boolean (value, pwsrc->always_copy);
+      break;
+
+    case PROP_MIN_BUFFERS:
+      g_value_set_int (value, pwsrc->min_buffers);
+      break;
+
+    case PROP_MAX_BUFFERS:
+      g_value_set_int (value, pwsrc->max_buffers);
+      break;
+
+    case PROP_FD:
+      g_value_set_int (value, pwsrc->fd);
+      break;
+
+    case PROP_RESEND_LAST:
+      g_value_set_boolean (value, pwsrc->resend_last);
+      break;
+
+    case PROP_KEEPALIVE_TIME:
+      g_value_set_int (value, pwsrc->keepalive_time);
+      break;
+
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static GstClock *
+gst_pipewire_src_provide_clock (GstElement * elem)
+{
+  GstPipeWireSrc *pwsrc = GST_PIPEWIRE_SRC (elem);
+  GstClock *clock;
+
+  GST_OBJECT_LOCK (pwsrc);
+  if (!GST_OBJECT_FLAG_IS_SET (pwsrc, GST_ELEMENT_FLAG_PROVIDE_CLOCK))
+    goto clock_disabled;
+
+  if (pwsrc->clock && pwsrc->is_live)
+    clock = GST_CLOCK_CAST (gst_object_ref (pwsrc->clock));
+  else
+    clock = NULL;
+  GST_OBJECT_UNLOCK (pwsrc);
+
+  return clock;
+
+  /* ERRORS */
+clock_disabled:
+  {
+    GST_DEBUG_OBJECT (pwsrc, "clock provide disabled");
+    GST_OBJECT_UNLOCK (pwsrc);
+    return NULL;
+  }
+}
+
+static void
+gst_pipewire_src_finalize (GObject * object)
+{
+  GstPipeWireSrc *pwsrc = GST_PIPEWIRE_SRC (object);
+
+  if (pwsrc->stream_properties)
+    gst_structure_free (pwsrc->stream_properties);
+  if (pwsrc->client_properties)
+    gst_structure_free (pwsrc->client_properties);
+  if (pwsrc->clock)
+    gst_object_unref (pwsrc->clock);
+  g_free (pwsrc->path);
+  g_free (pwsrc->target_object);
+  g_free (pwsrc->client_name);
+  g_object_unref(pwsrc->pool);
+
+  G_OBJECT_CLASS (parent_class)->finalize (object);
+}
+
+static void
+gst_pipewire_src_class_init (GstPipeWireSrcClass * klass)
+{
+  GObjectClass *gobject_class;
+  GstElementClass *gstelement_class;
+  GstBaseSrcClass *gstbasesrc_class;
+  GstPushSrcClass *gstpushsrc_class;
+
+  gobject_class = (GObjectClass *) klass;
+  gstelement_class = (GstElementClass *) klass;
+  gstbasesrc_class = (GstBaseSrcClass *) klass;
+  gstpushsrc_class = (GstPushSrcClass *) klass;
+
+  gobject_class->finalize = gst_pipewire_src_finalize;
+  gobject_class->set_property = gst_pipewire_src_set_property;
+  gobject_class->get_property = gst_pipewire_src_get_property;
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_PATH,
+                                   g_param_spec_string ("path",
+                                                        "Path",
+                                                        "The source path to connect to (NULL = default)",
+                                                        NULL,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_STATIC_STRINGS |
+                                                        G_PARAM_DEPRECATED));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_TARGET_OBJECT,
+                                   g_param_spec_string ("target-object",
+                                                        "Target object",
+                                                        "The source name/serial 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",
+                                                        "The client name to use (NULL = default)",
+                                                        NULL,
+                                                        G_PARAM_READWRITE |
+                                                        G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_CLIENT_PROPERTIES,
+                                   g_param_spec_boxed ("client-properties",
+                                                       "client properties",
+                                                       "list of PipeWire client properties",
+                                                       GST_TYPE_STRUCTURE,
+                                                       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 PipeWire stream properties",
+                                                       GST_TYPE_STRUCTURE,
+                                                       G_PARAM_READWRITE |
+                                                       G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_ALWAYS_COPY,
+                                   g_param_spec_boolean ("always-copy",
+                                                         "Always copy",
+                                                         "Always copy the buffer and data",
+                                                         DEFAULT_ALWAYS_COPY,
+                                                         G_PARAM_READWRITE |
+                                                         G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_MIN_BUFFERS,
+                                   g_param_spec_int ("min-buffers",
+                                                     "Min Buffers",
+                                                     "Minimum number of buffers to negotiate with PipeWire",
+                                                     1, G_MAXINT, DEFAULT_MIN_BUFFERS,
+                                                     G_PARAM_READWRITE |
+                                                     G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_MAX_BUFFERS,
+                                   g_param_spec_int ("max-buffers",
+                                                     "Max Buffers",
+                                                     "Maximum number of buffers to negotiate with PipeWire",
+                                                     1, G_MAXINT, DEFAULT_MAX_BUFFERS,
+                                                     G_PARAM_READWRITE |
+                                                     G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_FD,
+                                   g_param_spec_int ("fd",
+                                                     "Fd",
+                                                     "The fd to connect with",
+                                                     -1, G_MAXINT, -1,
+                                                     G_PARAM_READWRITE |
+                                                     G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_RESEND_LAST,
+                                   g_param_spec_boolean ("resend-last",
+                                                         "Resend last",
+                                                         "Resend last buffer on EOS",
+                                                         DEFAULT_RESEND_LAST,
+                                                         G_PARAM_READWRITE |
+                                                         G_PARAM_STATIC_STRINGS));
+
+  g_object_class_install_property (gobject_class,
+                                   PROP_KEEPALIVE_TIME,
+                                   g_param_spec_int ("keepalive-time",
+                                                     "Keepalive Time",
+                                                     "Periodically send last buffer (in milliseconds, 0 = disabled)",
+                                                     0, G_MAXINT, DEFAULT_KEEPALIVE_TIME,
+                                                     G_PARAM_READWRITE |
+                                                     G_PARAM_STATIC_STRINGS));
+
+  gstelement_class->provide_clock = gst_pipewire_src_provide_clock;
+  gstelement_class->change_state = gst_pipewire_src_change_state;
+  gstelement_class->send_event = gst_pipewire_src_send_event;
+
+  gst_element_class_set_static_metadata (gstelement_class,
+      "PipeWire source", "Source/Video",
+      "Uses PipeWire to create video", "Wim Taymans <wim.taymans@gmail.com>");
+
+  gst_element_class_add_pad_template (gstelement_class,
+      gst_static_pad_template_get (&gst_pipewire_src_template));
+
+  gstbasesrc_class->negotiate = gst_pipewire_src_negotiate;
+  gstbasesrc_class->unlock = gst_pipewire_src_unlock;
+  gstbasesrc_class->unlock_stop = gst_pipewire_src_unlock_stop;
+  gstbasesrc_class->start = gst_pipewire_src_start;
+  gstbasesrc_class->stop = gst_pipewire_src_stop;
+  gstbasesrc_class->event = gst_pipewire_src_event;
+  gstbasesrc_class->query = gst_pipewire_src_query;
+  gstpushsrc_class->create = gst_pipewire_src_create;
+
+  GST_DEBUG_CATEGORY_INIT (pipewire_src_debug, "pipewiresrc", 0,
+      "PipeWire Source");
+
+  process_mem_data_quark = g_quark_from_static_string ("GstPipeWireSrcProcessMemQuark");
+}
+
+static void
+gst_pipewire_src_init (GstPipeWireSrc * src)
+{
+  /* we operate in time */
+  gst_base_src_set_format (GST_BASE_SRC (src), GST_FORMAT_TIME);
+
+  /* we're a live source, unless explicitly requested not to be */
+  gst_base_src_set_live (GST_BASE_SRC (src), TRUE);
+
+  GST_OBJECT_FLAG_SET (src, GST_ELEMENT_FLAG_PROVIDE_CLOCK);
+
+  src->always_copy = DEFAULT_ALWAYS_COPY;
+  src->min_buffers = DEFAULT_MIN_BUFFERS;
+  src->max_buffers = DEFAULT_MAX_BUFFERS;
+  src->fd = -1;
+  src->resend_last = DEFAULT_RESEND_LAST;
+  src->keepalive_time = DEFAULT_KEEPALIVE_TIME;
+
+  src->client_name = g_strdup(pw_get_client_name ());
+
+  src->pool =  gst_pipewire_pool_new ();
+}
+
+static gboolean
+buffer_recycle (GstMiniObject *obj)
+{
+  GstPipeWireSrc *src;
+  GstPipeWirePoolData *data;
+  int res;
+
+  data = gst_pipewire_pool_get_data (GST_BUFFER_CAST(obj));
+
+  GST_OBJECT_LOCK (data->pool);
+  if (!obj->dispose) {
+    GST_OBJECT_UNLOCK (data->pool);
+    return TRUE;
+  }
+
+  GST_BUFFER_FLAGS (obj) = data->flags;
+  src = data->owner;
+
+  pw_thread_loop_lock (src->core->loop);
+  if (!obj->dispose) {
+    pw_thread_loop_unlock (src->core->loop);
+    GST_OBJECT_UNLOCK (data->pool);
+    return TRUE;
+  }
+
+  gst_mini_object_ref (obj);
+
+  data->queued = TRUE;
+
+  if ((res = pw_stream_queue_buffer (src->stream, data->b)) < 0)
+    GST_WARNING_OBJECT (src, "can't queue recycled buffer %p, %s", obj, spa_strerror(res));
+  else
+    GST_LOG_OBJECT (src, "recycle buffer %p", obj);
+
+  pw_thread_loop_unlock (src->core->loop);
+
+  GST_OBJECT_UNLOCK (data->pool);
+
+  return FALSE;
+}
+
+static void
+on_add_buffer (void *_data, struct pw_buffer *b)
+{
+  GstPipeWireSrc *pwsrc = _data;
+  GstPipeWirePoolData *data;
+
+  gst_pipewire_pool_wrap_buffer (pwsrc->pool, b);
+  data = b->user_data;
+  GST_DEBUG_OBJECT (pwsrc, "add buffer %p", data->buf);
+  data->owner = pwsrc;
+  data->queued = TRUE;
+  GST_MINI_OBJECT_CAST (data->buf)->dispose = buffer_recycle;
+}
+
+static void
+on_remove_buffer (void *_data, struct pw_buffer *b)
+{
+  GstPipeWireSrc *pwsrc = _data;
+  GstPipeWirePoolData *data = b->user_data;
+  GstBuffer *buf = data->buf;
+  int res;
+
+  GST_DEBUG_OBJECT (pwsrc, "remove buffer %p", buf);
+
+  GST_MINI_OBJECT_CAST (buf)->dispose = NULL;
+
+  if (data->queued) {
+    gst_buffer_unref (buf);
+  } else {
+    if ((res = pw_stream_queue_buffer (pwsrc->stream, b)) < 0)
+      GST_WARNING_OBJECT (pwsrc, "can't queue removed buffer %p, %s", buf, spa_strerror(res));
+  }
+}
+
+static const char * const transform_map[] = {
+  [SPA_META_TRANSFORMATION_None] = "rotate-0",
+  [SPA_META_TRANSFORMATION_90] = "rotate-90",
+  [SPA_META_TRANSFORMATION_180] = "rotate-180",
+  [SPA_META_TRANSFORMATION_270] = "rotate-270",
+  [SPA_META_TRANSFORMATION_Flipped] = "flip-rotate-0",
+  [SPA_META_TRANSFORMATION_Flipped90] = "flip-rotate-270",
+  [SPA_META_TRANSFORMATION_Flipped180] = "flip-rotate-180",
+  [SPA_META_TRANSFORMATION_Flipped270] = "flip-rotate-90",
+};
+
+static const char *spa_transform_value_to_gst_image_orientation(uint32_t transform_value)
+{
+  if (transform_value >= SPA_N_ELEMENTS(transform_map))
+    transform_value = SPA_META_TRANSFORMATION_None;
+
+  return transform_map[transform_value];
+}
+
+static GstBuffer *dequeue_buffer(GstPipeWireSrc *pwsrc)
+{
+  struct pw_buffer *b;
+  GstBuffer *buf;
+  GstPipeWirePoolData *data;
+  struct spa_meta_header *h;
+  struct spa_meta_region *crop;
+  struct spa_meta_videotransform *videotransform;
+  guint i;
+
+  b = pw_stream_dequeue_buffer (pwsrc->stream);
+  if (b == NULL)
+          return NULL;
+
+  data = b->user_data;
+
+  if (!GST_IS_BUFFER (data->buf)) {
+    GST_ERROR_OBJECT (pwsrc, "stream buffer %p is missing", data->buf);
+    return NULL;
+  }
+
+  if (!data->queued) {
+    GST_ERROR_OBJECT (pwsrc, "buffer %p was not recycled", data->buf);
+    return NULL;
+  }
+
+  GST_LOG_OBJECT (pwsrc, "got new buffer %p", data->buf);
+
+  buf = gst_buffer_new ();
+
+  data->queued = FALSE;
+  GST_BUFFER_PTS (buf) = GST_CLOCK_TIME_NONE;
+  GST_BUFFER_DTS (buf) = GST_CLOCK_TIME_NONE;
+
+  h = data->header;
+  if (h) {
+    GST_LOG_OBJECT (pwsrc, "pts %" G_GUINT64_FORMAT ", dts_offset %" G_GUINT64_FORMAT, h->pts, h->dts_offset);
+
+    if (GST_CLOCK_TIME_IS_VALID (h->pts)) {
+      GST_BUFFER_PTS (buf) = h->pts + GST_PIPEWIRE_CLOCK (pwsrc->clock)->time_offset;
+      if (GST_BUFFER_PTS (buf) + h->dts_offset > 0)
+        GST_BUFFER_DTS (buf) = GST_BUFFER_PTS (buf) + h->dts_offset;
+    }
+    GST_BUFFER_OFFSET (buf) = h->seq;
+  }
+  crop = data->crop;
+  if (crop) {
+    GstVideoCropMeta *meta = gst_buffer_get_video_crop_meta(buf);
+    if (meta) {
+      meta->x = crop->region.position.x;
+      meta->y = crop->region.position.y;
+      meta->width = crop->region.size.width;
+      meta->height = crop->region.size.height;
+    }
+  }
+
+  videotransform = data->videotransform;
+  if (videotransform) {
+    if (pwsrc->transform_value != videotransform->transform) {
+      GstEvent *tag_event;
+      const char* tag_string;
+
+      tag_string =
+          spa_transform_value_to_gst_image_orientation(videotransform->transform);
+
+      GST_LOG_OBJECT (pwsrc, "got new videotransform: %u / %s",
+          videotransform->transform, tag_string);
+
+      tag_event = gst_event_new_tag(gst_tag_list_new(GST_TAG_IMAGE_ORIENTATION,
+          tag_string, NULL));
+      gst_pad_push_event (GST_BASE_SRC_PAD (pwsrc), tag_event);
+
+      pwsrc->transform_value = videotransform->transform;
+    }
+  }
+
+  for (i = 0; i < b->buffer->n_datas; i++) {
+    struct spa_data *d = &b->buffer->datas[i];
+    GstMemory *pmem = gst_buffer_peek_memory (data->buf, i);
+    if (pmem) {
+      GstMemory *mem;
+      if (!pwsrc->always_copy)
+        mem = gst_memory_share (pmem, d->chunk->offset, d->chunk->size);
+      else
+        mem = gst_memory_copy (pmem, d->chunk->offset, d->chunk->size);
+      gst_buffer_insert_memory (buf, i, mem);
+    }
+    if (d->chunk->flags & SPA_CHUNK_FLAG_CORRUPTED)
+      GST_BUFFER_FLAG_SET (buf, GST_BUFFER_FLAG_CORRUPTED);
+  }
+  if (!pwsrc->always_copy)
+    gst_buffer_add_parent_buffer_meta (buf, data->buf);
+  gst_buffer_unref (data->buf);
+  return buf;
+}
+
+static void
+on_process (void *_data)
+{
+  GstPipeWireSrc *pwsrc = _data;
+  pw_thread_loop_signal (pwsrc->core->loop, FALSE);
+}
+
+static void
+on_state_changed (void *data,
+                  enum pw_stream_state old,
+                  enum pw_stream_state state, const char *error)
+{
+  GstPipeWireSrc *pwsrc = data;
+
+  GST_DEBUG ("got stream state %s", pw_stream_state_as_string (state));
+
+  switch (state) {
+    case PW_STREAM_STATE_UNCONNECTED:
+    case PW_STREAM_STATE_CONNECTING:
+    case PW_STREAM_STATE_PAUSED:
+    case PW_STREAM_STATE_STREAMING:
+      break;
+    case PW_STREAM_STATE_ERROR:
+      GST_ELEMENT_ERROR (pwsrc, RESOURCE, FAILED,
+          ("stream error: %s", error), (NULL));
+      break;
+  }
+  pw_thread_loop_signal (pwsrc->core->loop, FALSE);
+}
+
+static void
+parse_stream_properties (GstPipeWireSrc *pwsrc, const struct pw_properties *props)
+{
+  const gchar *var;
+  gboolean is_live;
+
+  GST_OBJECT_LOCK (pwsrc);
+  var = pw_properties_get (props, PW_KEY_STREAM_IS_LIVE);
+  is_live = pwsrc->is_live = var ? pw_properties_parse_bool(var) : TRUE;
+
+  var = pw_properties_get (props, PW_KEY_STREAM_LATENCY_MIN);
+  pwsrc->min_latency = var ? (GstClockTime) atoi (var) : 0;
+
+  var = pw_properties_get (props, PW_KEY_STREAM_LATENCY_MAX);
+  pwsrc->max_latency = var ? (GstClockTime) atoi (var) : GST_CLOCK_TIME_NONE;
+  GST_OBJECT_UNLOCK (pwsrc);
+
+  GST_DEBUG_OBJECT (pwsrc, "live %d", is_live);
+
+  gst_base_src_set_live (GST_BASE_SRC (pwsrc), is_live);
+}
+
+static gboolean
+gst_pipewire_src_stream_start (GstPipeWireSrc *pwsrc)
+{
+  const char *error = NULL;
+  struct timespec abstime;
+
+  pw_thread_loop_lock (pwsrc->core->loop);
+  GST_DEBUG_OBJECT (pwsrc, "doing stream start");
+
+  pw_thread_loop_get_time (pwsrc->core->loop, &abstime,
+                  GST_PIPEWIRE_DEFAULT_TIMEOUT * SPA_NSEC_PER_SEC);
+
+  while (TRUE) {
+    enum pw_stream_state state = pw_stream_get_state (pwsrc->stream, &error);
+
+    GST_DEBUG_OBJECT (pwsrc, "waiting for STREAMING, now %s", pw_stream_state_as_string (state));
+    if (state == PW_STREAM_STATE_STREAMING)
+      break;
+
+    if (state == PW_STREAM_STATE_ERROR)
+      goto start_error;
+
+    if (pwsrc->flushing) {
+      error = "flushing";
+      goto start_error;
+    }
+
+    if (pw_thread_loop_timed_wait_full (pwsrc->core->loop, &abstime) < 0) {
+      error = "timeout";
+      goto start_error;
+    }
+  }
+
+  parse_stream_properties (pwsrc, pw_stream_get_properties (pwsrc->stream));
+  GST_DEBUG_OBJECT (pwsrc, "signal started");
+  pwsrc->started = TRUE;
+  pw_thread_loop_signal (pwsrc->core->loop, FALSE);
+  pw_thread_loop_unlock (pwsrc->core->loop);
+
+  return TRUE;
+
+start_error:
+  {
+    GST_DEBUG_OBJECT (pwsrc, "error starting stream: %s", error);
+    pw_thread_loop_signal (pwsrc->core->loop, FALSE);
+    pw_thread_loop_unlock (pwsrc->core->loop);
+    return FALSE;
+  }
+}
+
+static enum pw_stream_state
+wait_started (GstPipeWireSrc *this)
+{
+  enum pw_stream_state state;
+  const char *error = NULL;
+  struct timespec abstime;
+
+  pw_thread_loop_lock (this->core->loop);
+
+  pw_thread_loop_get_time (this->core->loop, &abstime,
+                  GST_PIPEWIRE_DEFAULT_TIMEOUT * SPA_NSEC_PER_SEC);
+
+  while (TRUE) {
+    state = pw_stream_get_state (this->stream, &error);
+
+    GST_DEBUG_OBJECT (this, "waiting for started signal, state now %s",
+        pw_stream_state_as_string (state));
+
+    if (state == PW_STREAM_STATE_ERROR)
+      break;
+
+    if (this->flushing) {
+      state = PW_STREAM_STATE_ERROR;
+      break;
+    }
+
+    if (this->started)
+      break;
+
+    if (pw_thread_loop_timed_wait_full (this->core->loop, &abstime) < 0) {
+      state = PW_STREAM_STATE_ERROR;
+      break;
+    }
+  }
+  GST_DEBUG_OBJECT (this, "got started signal: %s",
+                  pw_stream_state_as_string (state));
+  pw_thread_loop_unlock (this->core->loop);
+
+  return state;
+}
+
+static gboolean
+gst_pipewire_src_negotiate (GstBaseSrc * basesrc)
+{
+  GstPipeWireSrc *pwsrc = GST_PIPEWIRE_SRC (basesrc);
+  GstCaps *thiscaps;
+  GstCaps *caps = NULL;
+  GstCaps *peercaps = NULL;
+  gboolean result = FALSE;
+  GPtrArray *possible;
+  const char *error = NULL;
+  struct timespec abstime;
+  uint32_t target_id;
+
+  /* first see what is possible on our source pad */
+  thiscaps = gst_pad_query_caps (GST_BASE_SRC_PAD (basesrc), NULL);
+  GST_DEBUG_OBJECT (basesrc, "caps of src: %" GST_PTR_FORMAT, thiscaps);
+  /* nothing or anything is allowed, we're done */
+  if (thiscaps == NULL)
+    goto no_nego_needed;
+
+  if (G_UNLIKELY (gst_caps_is_empty (thiscaps)))
+    goto no_caps;
+
+  /* get the peer caps */
+  peercaps = gst_pad_peer_query_caps (GST_BASE_SRC_PAD (basesrc), thiscaps);
+  GST_DEBUG_OBJECT (basesrc, "caps of peer: %" GST_PTR_FORMAT, peercaps);
+  if (peercaps) {
+    /* The result is already a subset of our caps */
+    caps = peercaps;
+    gst_caps_unref (thiscaps);
+  } else {
+    /* no peer, work with our own caps then */
+    caps = thiscaps;
+  }
+  if (caps == NULL || gst_caps_is_empty (caps))
+    goto no_common_caps;
+
+  GST_DEBUG_OBJECT (basesrc, "have common caps: %" GST_PTR_FORMAT, caps);
+
+  /* open a connection with these caps */
+  possible = gst_caps_to_format_all (caps, SPA_PARAM_EnumFormat);
+  gst_caps_unref (caps);
+
+  /* first disconnect */
+  pw_thread_loop_lock (pwsrc->core->loop);
+  if (pw_stream_get_state(pwsrc->stream, &error) != PW_STREAM_STATE_UNCONNECTED) {
+    GST_DEBUG_OBJECT (basesrc, "disconnect capture");
+    pw_stream_disconnect (pwsrc->stream);
+    while (TRUE) {
+      enum pw_stream_state state = pw_stream_get_state (pwsrc->stream, &error);
+
+      GST_DEBUG_OBJECT (basesrc, "waiting for UNCONNECTED, now %s", pw_stream_state_as_string (state));
+      if (state == PW_STREAM_STATE_UNCONNECTED)
+        break;
+
+      if (state == PW_STREAM_STATE_ERROR || pwsrc->flushing) {
+        g_ptr_array_unref (possible);
+        goto connect_error;
+      }
+
+      pw_thread_loop_wait (pwsrc->core->loop);
+    }
+  }
+
+  target_id = pwsrc->path ? (uint32_t)atoi(pwsrc->path) : PW_ID_ANY;
+
+  if (pwsrc->target_object) {
+      struct spa_dict_item items[2] = {
+        SPA_DICT_ITEM_INIT(PW_KEY_TARGET_OBJECT, pwsrc->target_object),
+	/* XXX deprecated but the portal and some example apps only
+	 * provide the object id */
+        SPA_DICT_ITEM_INIT(PW_KEY_NODE_TARGET, NULL),
+      };
+      struct spa_dict dict = SPA_DICT_INIT_ARRAY(items);
+      uint64_t serial;
+
+      /* If target.object is a name, set it also to node.target */
+      if (spa_atou64(pwsrc->target_object, &serial, 0)) {
+        dict.n_items = 1;
+      } else {
+        target_id = PW_ID_ANY;
+        items[1].value = pwsrc->target_object;
+      }
+
+      pw_stream_update_properties (pwsrc->stream, &dict);
+  }
+
+  GST_DEBUG_OBJECT (basesrc, "connect capture with path %s, target-object %s",
+                    pwsrc->path, pwsrc->target_object);
+  pwsrc->negotiated = FALSE;
+  pw_stream_connect (pwsrc->stream,
+                     PW_DIRECTION_INPUT,
+                     target_id,
+                     PW_STREAM_FLAG_AUTOCONNECT | PW_STREAM_FLAG_DONT_RECONNECT,
+                     (const struct spa_pod **)possible->pdata,
+                     possible->len);
+  g_ptr_array_free (possible, TRUE);
+
+  pw_thread_loop_get_time (pwsrc->core->loop, &abstime,
+                  GST_PIPEWIRE_DEFAULT_TIMEOUT * SPA_NSEC_PER_SEC);
+
+  while (TRUE) {
+    enum pw_stream_state state = pw_stream_get_state (pwsrc->stream, &error);
+
+    GST_DEBUG_OBJECT (basesrc, "waiting for NEGOTIATED, now %s", pw_stream_state_as_string (state));
+    if (state == PW_STREAM_STATE_ERROR || pwsrc->flushing)
+      goto connect_error;
+
+    if (pwsrc->negotiated)
+      break;
+
+    if (pw_thread_loop_timed_wait_full (pwsrc->core->loop, &abstime) < 0)
+        goto connect_error;
+  }
+  caps = pwsrc->caps;
+  pwsrc->caps = NULL;
+  pw_thread_loop_unlock (pwsrc->core->loop);
+
+  if (caps == NULL)
+    goto no_caps;
+
+  gst_pipewire_clock_reset (GST_PIPEWIRE_CLOCK (pwsrc->clock), 0);
+
+  GST_DEBUG_OBJECT (pwsrc, "set format %" GST_PTR_FORMAT, caps);
+  result = gst_base_src_set_caps (GST_BASE_SRC (pwsrc), caps);
+  gst_caps_unref (caps);
+
+  result = gst_pipewire_src_stream_start (pwsrc);
+
+  pwsrc->started = result;
+
+  return result;
+
+no_nego_needed:
+  {
+    GST_DEBUG_OBJECT (basesrc, "no negotiation needed");
+    if (thiscaps)
+      gst_caps_unref (thiscaps);
+    return TRUE;
+  }
+no_caps:
+  {
+    GST_ELEMENT_ERROR (basesrc, STREAM, FORMAT,
+        ("No supported formats found"),
+        ("This element did not produce valid caps"));
+    if (thiscaps)
+      gst_caps_unref (thiscaps);
+    return FALSE;
+  }
+no_common_caps:
+  {
+    GST_ELEMENT_ERROR (basesrc, STREAM, FORMAT,
+        ("No supported formats found"),
+        ("This element does not have formats in common with the peer"));
+    if (caps)
+      gst_caps_unref (caps);
+    return FALSE;
+  }
+connect_error:
+  {
+    GST_DEBUG_OBJECT (basesrc, "connect error");
+    pw_thread_loop_unlock (pwsrc->core->loop);
+    return FALSE;
+  }
+}
+
+static void
+on_param_changed (void *data, uint32_t id,
+                   const struct spa_pod *param)
+{
+  GstPipeWireSrc *pwsrc = data;
+
+  if (param == NULL || id != SPA_PARAM_Format) {
+    GST_DEBUG_OBJECT (pwsrc, "clear format");
+    return;
+  }
+  if (pwsrc->caps)
+          gst_caps_unref(pwsrc->caps);
+  pwsrc->caps = gst_caps_from_format (param);
+
+  pwsrc->negotiated = pwsrc->caps != NULL;
+
+  if (pwsrc->negotiated) {
+    const struct spa_pod *params[4];
+    struct spa_pod_builder b = { NULL };
+    uint8_t buffer[512];
+    uint32_t buffers = CLAMP (16, pwsrc->min_buffers, pwsrc->max_buffers);
+    int buffertypes;
+
+    buffertypes = (1<<SPA_DATA_DmaBuf);
+    if (spa_pod_find_prop (param, NULL, SPA_FORMAT_VIDEO_modifier)) {
+      gst_caps_features_remove (gst_caps_get_features (pwsrc->caps, 0),
+          GST_CAPS_FEATURE_MEMORY_SYSTEM_MEMORY);
+      gst_caps_features_add (gst_caps_get_features (pwsrc->caps, 0),
+          GST_CAPS_FEATURE_MEMORY_DMABUF);
+    } else {
+      buffertypes |= ((1<<SPA_DATA_MemFd) | (1<<SPA_DATA_MemPtr));
+    }
+
+    GST_DEBUG_OBJECT (pwsrc, "we got format %" GST_PTR_FORMAT, pwsrc->caps);
+
+    spa_pod_builder_init (&b, buffer, sizeof (buffer));
+    params[0] = spa_pod_builder_add_object (&b,
+        SPA_TYPE_OBJECT_ParamBuffers, SPA_PARAM_Buffers,
+        SPA_PARAM_BUFFERS_buffers, SPA_POD_CHOICE_RANGE_Int(buffers,
+                                                            pwsrc->min_buffers,
+                                                            pwsrc->max_buffers),
+        SPA_PARAM_BUFFERS_blocks,  SPA_POD_CHOICE_RANGE_Int(0, 1, INT32_MAX),
+        SPA_PARAM_BUFFERS_size,    SPA_POD_CHOICE_RANGE_Int(0, 0, INT32_MAX),
+        SPA_PARAM_BUFFERS_stride,  SPA_POD_CHOICE_RANGE_Int(0, 0, INT32_MAX),
+        SPA_PARAM_BUFFERS_dataType, SPA_POD_CHOICE_FLAGS_Int(buffertypes));
+
+    params[1] = spa_pod_builder_add_object (&b,
+        SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta,
+        SPA_PARAM_META_type, SPA_POD_Id(SPA_META_Header),
+        SPA_PARAM_META_size, SPA_POD_Int(sizeof (struct spa_meta_header)));
+    params[2] = spa_pod_builder_add_object (&b,
+        SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta,
+        SPA_PARAM_META_type, SPA_POD_Id(SPA_META_VideoCrop),
+        SPA_PARAM_META_size, SPA_POD_Int(sizeof (struct spa_meta_region)));
+    params[3] = spa_pod_builder_add_object (&b,
+        SPA_TYPE_OBJECT_ParamMeta, SPA_PARAM_Meta,
+        SPA_PARAM_META_type, SPA_POD_Id(SPA_META_VideoTransform),
+        SPA_PARAM_META_size, SPA_POD_Int(sizeof (struct spa_meta_videotransform)));
+
+    GST_DEBUG_OBJECT (pwsrc, "doing finish format");
+    pw_stream_update_params (pwsrc->stream, params, SPA_N_ELEMENTS(params));
+  } else {
+    GST_WARNING_OBJECT (pwsrc, "finish format with error");
+    pw_stream_set_error (pwsrc->stream, -EINVAL, "unhandled format");
+  }
+  pw_thread_loop_signal (pwsrc->core->loop, FALSE);
+}
+
+static gboolean
+gst_pipewire_src_unlock (GstBaseSrc * basesrc)
+{
+  GstPipeWireSrc *pwsrc = GST_PIPEWIRE_SRC (basesrc);
+
+  pw_thread_loop_lock (pwsrc->core->loop);
+  GST_DEBUG_OBJECT (pwsrc, "setting flushing");
+  pwsrc->flushing = TRUE;
+  pw_thread_loop_signal (pwsrc->core->loop, FALSE);
+  pw_thread_loop_unlock (pwsrc->core->loop);
+
+  return TRUE;
+}
+
+static gboolean
+gst_pipewire_src_unlock_stop (GstBaseSrc * basesrc)
+{
+  GstPipeWireSrc *pwsrc = GST_PIPEWIRE_SRC (basesrc);
+
+  pw_thread_loop_lock (pwsrc->core->loop);
+  GST_DEBUG_OBJECT (pwsrc, "unsetting flushing");
+  pwsrc->flushing = FALSE;
+  pw_thread_loop_unlock (pwsrc->core->loop);
+
+  return TRUE;
+}
+
+static gboolean
+gst_pipewire_src_event (GstBaseSrc * src, GstEvent * event)
+{
+  gboolean res = FALSE;
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_CUSTOM_UPSTREAM:
+      if (gst_video_event_is_force_key_unit (event)) {
+        GstClockTime running_time;
+        gboolean all_headers;
+        guint count;
+
+        gst_video_event_parse_upstream_force_key_unit (event,
+                &running_time, &all_headers, &count);
+
+        res = TRUE;
+      } else {
+        res = GST_BASE_SRC_CLASS (parent_class)->event (src, event);
+      }
+      break;
+    default:
+      res = GST_BASE_SRC_CLASS (parent_class)->event (src, event);
+      break;
+  }
+  return res;
+}
+
+static gboolean
+gst_pipewire_src_query (GstBaseSrc * src, GstQuery * query)
+{
+  gboolean res = FALSE;
+  GstPipeWireSrc *pwsrc;
+
+  pwsrc = GST_PIPEWIRE_SRC (src);
+
+  switch (GST_QUERY_TYPE (query)) {
+    case GST_QUERY_LATENCY:
+      GST_OBJECT_LOCK (pwsrc);
+      pwsrc->min_latency = 10000000;
+      pwsrc->max_latency = GST_CLOCK_TIME_NONE;
+      gst_query_set_latency (query, pwsrc->is_live, pwsrc->min_latency, pwsrc->max_latency);
+      GST_OBJECT_UNLOCK (pwsrc);
+      res = TRUE;
+      break;
+    default:
+      res = GST_BASE_SRC_CLASS (parent_class)->query (src, query);
+      break;
+  }
+  return res;
+}
+
+static GstFlowReturn
+gst_pipewire_src_create (GstPushSrc * psrc, GstBuffer ** buffer)
+{
+  GstPipeWireSrc *pwsrc;
+  GstClockTime pts, dts, base_time;
+  const char *error = NULL;
+  GstBuffer *buf;
+  gboolean update_time = FALSE, timeout = FALSE;
+
+  pwsrc = GST_PIPEWIRE_SRC (psrc);
+
+  pw_thread_loop_lock (pwsrc->core->loop);
+  if (!pwsrc->negotiated)
+    goto not_negotiated;
+
+  while (TRUE) {
+    enum pw_stream_state state;
+
+    if (pwsrc->flushing)
+      goto streaming_stopped;
+
+    if (pwsrc->stream == NULL)
+      goto streaming_error;
+
+    state = pw_stream_get_state (pwsrc->stream, &error);
+    if (state == PW_STREAM_STATE_ERROR)
+      goto streaming_error;
+
+    if (state != PW_STREAM_STATE_STREAMING)
+      goto streaming_stopped;
+
+    if (pwsrc->eos) {
+      if (pwsrc->last_buffer == NULL)
+        goto streaming_eos;
+      buf = pwsrc->last_buffer;
+      pwsrc->last_buffer = NULL;
+      update_time = TRUE;
+      GST_LOG_OBJECT (pwsrc, "EOS, send last buffer");
+      break;
+    } else if (timeout) {
+      if (pwsrc->last_buffer != NULL) {
+        update_time = TRUE;
+        buf = gst_buffer_ref(pwsrc->last_buffer);
+        GST_LOG_OBJECT (pwsrc, "timeout, send keepalive buffer");
+        break;
+      }
+    } else {
+      buf = dequeue_buffer (pwsrc);
+      GST_LOG_OBJECT (pwsrc, "popped buffer %p", buf);
+      if (buf != NULL) {
+        if (pwsrc->resend_last || pwsrc->keepalive_time > 0)
+          gst_buffer_replace (&pwsrc->last_buffer, buf);
+        break;
+      }
+    }
+    timeout = FALSE;
+    if (pwsrc->keepalive_time > 0) {
+      struct timespec abstime;
+      pw_thread_loop_get_time(pwsrc->core->loop, &abstime,
+                      pwsrc->keepalive_time * SPA_NSEC_PER_MSEC);
+      if (pw_thread_loop_timed_wait_full (pwsrc->core->loop, &abstime) == -ETIMEDOUT)
+        timeout = TRUE;
+    } else {
+      pw_thread_loop_wait (pwsrc->core->loop);
+    }
+  }
+  pw_thread_loop_unlock (pwsrc->core->loop);
+
+  *buffer = buf;
+
+  if (pwsrc->is_live)
+    base_time = GST_ELEMENT_CAST (psrc)->base_time;
+  else
+    base_time = 0;
+
+  if (update_time) {
+    GstClock *clock = gst_element_get_clock (GST_ELEMENT_CAST (pwsrc));
+    if (clock != NULL) {
+      pts = dts = gst_clock_get_time (clock);
+      gst_object_unref (clock);
+    } else {
+      pts = dts = GST_CLOCK_TIME_NONE;
+    }
+  } else {
+    pts = GST_BUFFER_PTS (*buffer);
+    dts = GST_BUFFER_DTS (*buffer);
+  }
+
+  if (GST_CLOCK_TIME_IS_VALID (pts))
+    pts = (pts >= base_time ? pts - base_time : 0);
+  if (GST_CLOCK_TIME_IS_VALID (dts))
+    dts = (dts >= base_time ? dts - base_time : 0);
+
+  GST_LOG_OBJECT (pwsrc,
+      "pts %" G_GUINT64_FORMAT ", dts %" G_GUINT64_FORMAT
+      ", base-time %" GST_TIME_FORMAT " -> %" GST_TIME_FORMAT ", %" GST_TIME_FORMAT,
+      GST_BUFFER_PTS (*buffer), GST_BUFFER_DTS (*buffer), GST_TIME_ARGS (base_time),
+      GST_TIME_ARGS (pts), GST_TIME_ARGS (dts));
+
+  GST_BUFFER_PTS (*buffer) = pts;
+  GST_BUFFER_DTS (*buffer) = dts;
+
+  return GST_FLOW_OK;
+
+not_negotiated:
+  {
+    pw_thread_loop_unlock (pwsrc->core->loop);
+    return GST_FLOW_NOT_NEGOTIATED;
+  }
+streaming_eos:
+  {
+    pw_thread_loop_unlock (pwsrc->core->loop);
+    return GST_FLOW_EOS;
+  }
+streaming_error:
+  {
+    pw_thread_loop_unlock (pwsrc->core->loop);
+    return GST_FLOW_ERROR;
+  }
+streaming_stopped:
+  {
+    pw_thread_loop_unlock (pwsrc->core->loop);
+    return GST_FLOW_FLUSHING;
+  }
+}
+
+static gboolean
+gst_pipewire_src_start (GstBaseSrc * basesrc)
+{
+  return TRUE;
+}
+
+static gboolean
+gst_pipewire_src_stop (GstBaseSrc * basesrc)
+{
+  GstPipeWireSrc *pwsrc;
+
+  pwsrc = GST_PIPEWIRE_SRC (basesrc);
+
+  pw_thread_loop_lock (pwsrc->core->loop);
+  pwsrc->eos = false;
+  gst_buffer_replace (&pwsrc->last_buffer, NULL);
+  gst_caps_replace(&pwsrc->caps, NULL);
+  pw_thread_loop_unlock (pwsrc->core->loop);
+
+  return TRUE;
+}
+
+static gboolean
+copy_properties (GQuark field_id,
+                 const GValue *value,
+                 gpointer user_data)
+{
+  struct pw_properties *properties = user_data;
+  GValue dst = { 0 };
+
+  if (g_value_type_transformable (G_VALUE_TYPE(value), G_TYPE_STRING)) {
+    g_value_init(&dst, G_TYPE_STRING);
+    if (g_value_transform(value, &dst)) {
+      pw_properties_set (properties,
+                         g_quark_to_string (field_id),
+                         g_value_get_string (&dst));
+    }
+    g_value_unset(&dst);
+  }
+  return TRUE;
+}
+
+static const struct pw_stream_events stream_events = {
+        PW_VERSION_STREAM_EVENTS,
+        .state_changed = on_state_changed,
+        .param_changed = on_param_changed,
+        .add_buffer = on_add_buffer,
+        .remove_buffer = on_remove_buffer,
+        .process = on_process,
+};
+
+static gboolean
+gst_pipewire_src_open (GstPipeWireSrc * pwsrc)
+{
+  struct pw_properties *props;
+
+  GST_DEBUG_OBJECT (pwsrc, "open");
+
+  pwsrc->core = gst_pipewire_core_get(pwsrc->fd);
+  if (pwsrc->core == NULL)
+      goto connect_error;
+
+  pw_thread_loop_lock (pwsrc->core->loop);
+
+  props = pw_properties_new (NULL, NULL);
+  if (pwsrc->client_properties) {
+    gst_structure_foreach (pwsrc->client_properties, copy_properties, props);
+    pw_core_update_properties (pwsrc->core->core, &props->dict);
+    pw_properties_clear(props);
+  }
+  if (pwsrc->client_name) {
+    pw_properties_set (props, PW_KEY_NODE_NAME, pwsrc->client_name);
+    pw_properties_set (props, PW_KEY_NODE_DESCRIPTION, pwsrc->client_name);
+  }
+  if (pwsrc->stream_properties) {
+    gst_structure_foreach (pwsrc->stream_properties, copy_properties, props);
+  }
+
+  if ((pwsrc->stream = pw_stream_new (pwsrc->core->core,
+                                  pwsrc->client_name, props)) == NULL)
+    goto no_stream;
+
+
+  pw_stream_add_listener(pwsrc->stream,
+                         &pwsrc->stream_listener,
+                         &stream_events,
+                         pwsrc);
+
+  pwsrc->clock = gst_pipewire_clock_new (pwsrc->stream, pwsrc->last_time);
+  pw_thread_loop_unlock (pwsrc->core->loop);
+
+  return TRUE;
+
+  /* ERRORS */
+connect_error:
+  {
+    GST_ELEMENT_ERROR (pwsrc, RESOURCE, FAILED, ("can't connect"), (NULL));
+    return FALSE;
+  }
+no_stream:
+  {
+    GST_ELEMENT_ERROR (pwsrc, RESOURCE, FAILED, ("can't create stream"), (NULL));
+    pw_thread_loop_unlock (pwsrc->core->loop);
+    gst_pipewire_core_release (pwsrc->core);
+    pwsrc->core = NULL;
+    return FALSE;
+  }
+}
+
+static void
+gst_pipewire_src_close (GstPipeWireSrc * pwsrc)
+{
+  pwsrc->last_time = gst_clock_get_time (pwsrc->clock);
+
+  GST_DEBUG_OBJECT (pwsrc, "close");
+
+  gst_element_post_message (GST_ELEMENT (pwsrc),
+    gst_message_new_clock_lost (GST_OBJECT_CAST (pwsrc), pwsrc->clock));
+
+  GST_OBJECT_LOCK (pwsrc);
+  GST_PIPEWIRE_CLOCK (pwsrc->clock)->stream = NULL;
+  g_clear_object (&pwsrc->clock);
+  GST_OBJECT_UNLOCK (pwsrc);
+
+  GST_OBJECT_LOCK (pwsrc->pool);
+  pw_thread_loop_lock (pwsrc->core->loop);
+  if (pwsrc->stream) {
+    pw_stream_destroy (pwsrc->stream);
+    pwsrc->stream = NULL;
+  }
+  pw_thread_loop_unlock (pwsrc->core->loop);
+  GST_OBJECT_UNLOCK (pwsrc->pool);
+
+  if (pwsrc->core) {
+    gst_pipewire_core_release (pwsrc->core);
+    pwsrc->core = NULL;
+  }
+}
+
+static gboolean
+gst_pipewire_src_send_event (GstElement * elem, GstEvent * event)
+{
+  GstPipeWireSrc *this = GST_PIPEWIRE_SRC_CAST (elem);
+  gboolean ret;
+
+  switch (GST_EVENT_TYPE (event)) {
+    case GST_EVENT_EOS:
+      GST_DEBUG_OBJECT (this, "got EOS");
+      pw_thread_loop_lock (this->core->loop);
+      this->eos = true;
+      pw_thread_loop_signal (this->core->loop, FALSE);
+      pw_thread_loop_unlock (this->core->loop);
+      ret = TRUE;
+      break;
+    default:
+      ret = GST_ELEMENT_CLASS (parent_class)->send_event (elem, event);
+      break;
+  }
+  return ret;
+}
+
+static GstStateChangeReturn
+gst_pipewire_src_change_state (GstElement * element, GstStateChange transition)
+{
+  GstStateChangeReturn ret;
+  GstPipeWireSrc *this = GST_PIPEWIRE_SRC_CAST (element);
+
+  switch (transition) {
+    case GST_STATE_CHANGE_NULL_TO_READY:
+      if (!gst_pipewire_src_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 */
+      pw_thread_loop_lock (this->core->loop);
+      pw_stream_set_active(this->stream, true);
+      pw_thread_loop_unlock (this->core->loop);
+      break;
+    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
+      /* stop recording ASAP by corking */
+      pw_thread_loop_lock (this->core->loop);
+      pw_stream_set_active(this->stream, false);
+      pw_thread_loop_unlock (this->core->loop);
+      break;
+    default:
+      break;
+  }
+
+  ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
+
+  switch (transition) {
+    case GST_STATE_CHANGE_READY_TO_PAUSED:
+      if (wait_started (this) == PW_STREAM_STATE_ERROR)
+        goto open_failed;
+
+      if (gst_base_src_is_live (GST_BASE_SRC (element)))
+        ret = GST_STATE_CHANGE_NO_PREROLL;
+      break;
+    case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
+      break;
+    case GST_STATE_CHANGE_PAUSED_TO_READY:
+      pw_thread_loop_lock (this->core->loop);
+      this->negotiated = FALSE;
+      pw_thread_loop_unlock (this->core->loop);
+      break;
+    case GST_STATE_CHANGE_READY_TO_NULL:
+      gst_pipewire_src_close (this);
+      break;
+    default:
+      break;
+  }
+  return ret;
+
+  /* ERRORS */
+open_failed:
+  {
+    return GST_STATE_CHANGE_FAILURE;
+  }
+}
diff --git a/src/gst/gstpipewiresrc.h b/src/gst/gstpipewiresrc.h
new file mode 100644
index 0000000..e1def85
--- /dev/null
+++ b/src/gst/gstpipewiresrc.h
@@ -0,0 +1,110 @@
+/* GStreamer
+ *
+ * Copyright © 2018 Wim Taymans
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the "Software"),
+ * to deal in the Software without restriction, including without limitation
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,
+ * and/or sell copies of the Software, and to permit persons to whom the
+ * Software is furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice (including the next
+ * paragraph) shall be included in all copies or substantial portions of the
+ * Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
+ * DEALINGS IN THE SOFTWARE.
+ */
+
+#ifndef __GST_PIPEWIRE_SRC_H__
+#define __GST_PIPEWIRE_SRC_H__
+
+#include <gst/gst.h>
+#include <gst/base/gstpushsrc.h>
+
+#include <pipewire/pipewire.h>
+#include <gst/gstpipewirepool.h>
+#include <gst/gstpipewirecore.h>
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_PIPEWIRE_SRC \
+  (gst_pipewire_src_get_type())
+#define GST_PIPEWIRE_SRC(obj) \
+  (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_PIPEWIRE_SRC,GstPipeWireSrc))
+#define GST_PIPEWIRE_SRC_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_PIPEWIRE_SRC,GstPipeWireSrcClass))
+#define GST_IS_PIPEWIRE_SRC(obj) \
+  (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_PIPEWIRE_SRC))
+#define GST_IS_PIPEWIRE_SRC_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PIPEWIRE_SRC))
+#define GST_PIPEWIRE_SRC_CAST(obj) \
+  ((GstPipeWireSrc *) (obj))
+
+typedef struct _GstPipeWireSrc GstPipeWireSrc;
+typedef struct _GstPipeWireSrcClass GstPipeWireSrcClass;
+
+/**
+ * GstPipeWireSrc:
+ *
+ * Opaque data structure.
+ */
+struct _GstPipeWireSrc {
+  GstPushSrc element;
+
+  /*< private >*/
+  gchar *path;
+  gchar *target_object;
+  gchar *client_name;
+  gboolean always_copy;
+  gint min_buffers;
+  gint max_buffers;
+  int fd;
+  gboolean resend_last;
+  gint keepalive_time;
+
+  GstCaps *caps;
+
+  gboolean negotiated;
+  gboolean flushing;
+  gboolean started;
+  gboolean eos;
+
+  gboolean is_live;
+  GstClockTime min_latency;
+  GstClockTime max_latency;
+
+  GstStructure *client_properties;
+  GstPipeWireCore *core;
+  struct spa_hook core_listener;
+  int last_seq;
+  int pending_seq;
+
+  struct pw_stream *stream;
+  struct spa_hook stream_listener;
+
+  GstBuffer *last_buffer;
+  GstStructure *stream_properties;
+
+  GstPipeWirePool *pool;
+  GstClock *clock;
+  GstClockTime last_time;
+
+  enum spa_meta_videotransform_value transform_value;
+};
+
+struct _GstPipeWireSrcClass {
+  GstPushSrcClass parent_class;
+};
+
+GType gst_pipewire_src_get_type (void);
+
+G_END_DECLS
+
+#endif /* __GST_PIPEWIRE_SRC_H__ */
diff --git a/src/gst/meson.build b/src/gst/meson.build
new file mode 100644
index 0000000..fd552f6
--- /dev/null
+++ b/src/gst/meson.build
@@ -0,0 +1,33 @@
+pipewire_gst_sources = [
+  'gstpipewire.c',
+  'gstpipewirecore.c',
+  'gstpipewireclock.c',
+  'gstpipewireformat.c',
+  'gstpipewirepool.c',
+  'gstpipewiresink.c',
+  'gstpipewiresrc.c',
+]
+
+if get_option('gstreamer-device-provider').allowed()
+  pipewire_gst_sources += [ 'gstpipewiredeviceprovider.c' ]
+endif
+
+pipewire_gst_headers = [
+  'gstpipewireclock.h',
+  'gstpipewirecore.h',
+  'gstpipewiredeviceprovider.h',
+  'gstpipewireformat.h',
+  'gstpipewirepool.h',
+  'gstpipewiresink.h',
+  'gstpipewiresrc.h',
+]
+
+pipewire_gst = shared_library('gstpipewire',
+    pipewire_gst_sources,
+    include_directories : [ configinc ],
+    dependencies : [ spa_dep, gst_dep, pipewire_dep ],
+    install : true,
+    install_dir : '@0@/gstreamer-1.0'.format(get_option('libdir')),
+)
+
+plugins = [pipewire_gst]