srt: Integrate server and client element into one

We have srt{client,server}{src,sink} elements in accordance to the
norm of the connection oriented protocols. However, SRT connection
mode can be changed by uri parameters so it requires an integrated
element to handle the parameters.

fix: #740

23 files changed, 2549 insertions, 3070 deletions

diff --git a/ext/srt/Makefile.am b/ext/srt/Makefile.am
index bc504f891..069f0ecfe 100644
--- a/ext/srt/Makefile.am
+++ b/ext/srt/Makefile.am
@@ -1,13 +1,9 @@
 plugin_LTLIBRARIES = libgstsrt.la
 
 libgstsrt_la_SOURCES = \
-	gstsrt.c \
-	gstsrtbasesrc.c \
-	gstsrtclientsrc.c \
-	gstsrtserversrc.c \
-	gstsrtbasesink.c \
-	gstsrtclientsink.c \
-	gstsrtserversink.c \
+	gstsrtobject.c \
+	gstsrtsink.c \
+	gstsrtsrc.c \
 	$(NULL)
 
 libgstsrt_la_CFLAGS = \
@@ -30,13 +26,9 @@ libgstsrt_la_LDFLAGS = $(GST_PLUGIN_LDFLAGS)
 CLEANFILES = $(BUILT_SOURCES)
 
 noinst_HEADERS = \
-	gstsrt.h \
-	gstsrtbasesrc.h \
-	gstsrtclientsrc.h \
-	gstsrtserversrc.h \
-	gstsrtbasesink.h \
-	gstsrtclientsink.h  \
-	gstsrtserversink.h \
+	gstsrtobject.h \
+	gstsrtsink.h \
+	gstsrtsrc.h \
 	$(NULL)
 
 include $(top_srcdir)/common/gst-glib-gen.mak
diff --git a/ext/srt/gstsrt-enums.h b/ext/srt/gstsrt-enums.h
new file mode 100644
index 000000000..1da1a11c0
--- /dev/null
+++ b/ext/srt/gstsrt-enums.h
@@ -0,0 +1,67 @@
+/* GStreamer
+ * Copyright (C) 2018, Collabora Ltd.
+ * Copyright (C) 2018, SK Telecom, Co., Ltd.
+ *   Author: Jeongseok Kim <jeongseok.kim@sk.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __GST_SRT_ENUM_H__
+#define __GST_SRT_ENUM_H__
+
+#include <gst/gst.h>
+
+G_BEGIN_DECLS
+
+/**
+ * GstSRTConnectionMode:
+ * @GST_SRT_CONNECTION_MODE_NONE: not connected
+ * @GST_SRT_CONNECTION_MODE_CALLER: The mode to send the connection request like a client
+ * @GST_SRT_CONNECTION_MODE_LISTENER: The mode to wait for being connected by peer caller
+ * @GST_SRT_CONNECTION_MODE_RENDEZVOUS: The mode to support one-to-one only connection
+ *
+ * SRT connection types.
+ */
+typedef enum
+{
+  GST_SRT_CONNECTION_MODE_NONE = 0,
+  GST_SRT_CONNECTION_MODE_CALLER,
+  GST_SRT_CONNECTION_MODE_LISTENER,
+  GST_SRT_CONNECTION_MODE_RENDEZVOUS,
+} GstSRTConnectionMode;
+
+/**
+ * GstSRTKeyLengthBits:
+ * @GST_SRT_KEY_LENGTH_BITS_NO_KEY: no encryption
+ * @GST_SRT_KEY_LENGTH_BITS_0: no encryption
+ * @GST_SRT_KEY_LENGTH_BITS_128: 128-bit length
+ * @GST_SRT_KEY_LENGTH_BITS_192: 192-bit length
+ * @GST_SRT_KEY_LENGTH_BITS_256: 256-bit length
+ *
+ * Crypto key length in bits
+ */
+typedef enum
+{
+  GST_SRT_KEY_LENGTH_BITS_NO_KEY = 0,
+  GST_SRT_KEY_LENGTH_BITS_0 = GST_SRT_KEY_LENGTH_BITS_NO_KEY,
+  GST_SRT_KEY_LENGTH_BITS_128 = 128,
+  GST_SRT_KEY_LENGTH_BITS_192 = 192,
+  GST_SRT_KEY_LENGTH_BITS_256 = 256,
+} GstSRTKeyLengthBits;
+
+G_END_DECLS
+
+#endif // __GST_SRT_ENUM_H__
diff --git a/ext/srt/gstsrt.c b/ext/srt/gstsrt.c
index 79e27a0fd..9b85de199 100644
--- a/ext/srt/gstsrt.c
+++ b/ext/srt/gstsrt.c
@@ -22,317 +22,35 @@
 #include "config.h"
 #endif
 
-#include "gstsrt.h"
-
-#include "gstsrtclientsrc.h"
-#include "gstsrtserversrc.h"
-#include "gstsrtclientsink.h"
-#include "gstsrtserversink.h"
-
-#define GST_CAT_DEFAULT gst_debug_srt
-GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
-
-static GSocketAddress *
-gst_srt_socket_address_new (GstElement * elem, const gchar * host, guint16 port)
-{
-  GInetAddress *iaddr = NULL;
-  GSocketAddress *addr = NULL;
-  GError *error = NULL;
-
-  if (host == NULL) {
-    iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV4);
-  } else {
-    iaddr = g_inet_address_new_from_string (host);
-  }
-
-  if (!iaddr) {
-    GList *results;
-    GResolver *resolver = g_resolver_get_default ();
-
-    results = g_resolver_lookup_by_name (resolver, host, NULL, &error);
-
-    if (!results) {
-      GST_ERROR_OBJECT (elem, "Failed to resolve %s: %s", host, error->message);
-      g_object_unref (resolver);
-      goto failed;
-    }
-
-    iaddr = G_INET_ADDRESS (g_object_ref (results->data));
-
-    g_resolver_free_addresses (results);
-    g_object_unref (resolver);
-  }
-#ifndef GST_DISABLE_GST_DEBUG
-  {
-    gchar *ip = g_inet_address_to_string (iaddr);
-
-    GST_DEBUG_OBJECT (elem, "IP address for host %s is %s", host, ip);
-    g_free (ip);
-  }
-#endif
-
-  addr = g_inet_socket_address_new (iaddr, port);
-  g_object_unref (iaddr);
-
-  return addr;
-
-failed:
-  g_clear_error (&error);
-
-  return NULL;
-}
-
-SRTSOCKET
-gst_srt_client_connect (GstElement * elem, int sender,
-    const gchar * host, guint16 port, int rendez_vous,
-    const gchar * bind_address, guint16 bind_port, int latency,
-    GSocketAddress ** socket_address, gint * poll_id, const gchar * passphrase,
-    int key_length)
-{
-  SRTSOCKET sock = SRT_INVALID_SOCK;
-  GError *error = NULL;
-  gpointer sa;
-  size_t sa_len;
-  int poll_event = SRT_EPOLL_ERR;
-
-  poll_event |= sender ? SRT_EPOLL_OUT : SRT_EPOLL_IN;
-
-  if (host == NULL) {
-    GST_ELEMENT_ERROR (elem, RESOURCE, OPEN_READ, ("Invalid host"),
-        ("Unspecified NULL host"));
-    goto failed;
-  }
-
-  *socket_address = gst_srt_socket_address_new (elem, host, port);
-
-  if (*socket_address == NULL) {
-    GST_ELEMENT_ERROR (elem, RESOURCE, OPEN_READ, ("Invalid host"),
-        ("Failed to parse host"));
-    goto failed;
-  }
-
-  sa_len = g_socket_address_get_native_size (*socket_address);
-  sa = g_alloca (sa_len);
-  if (!g_socket_address_to_native (*socket_address, sa, sa_len, &error)) {
-    GST_ELEMENT_ERROR (elem, RESOURCE, OPEN_READ, ("Invalid address"),
-        ("cannot resolve address (reason: %s)", error->message));
-    goto failed;
-  }
-
-  sock = srt_socket (g_socket_address_get_family (*socket_address), SOCK_DGRAM,
-      0);
-  if (sock == SRT_ERROR) {
-    GST_ELEMENT_ERROR (elem, LIBRARY, INIT, (NULL),
-        ("failed to create SRT socket (reason: %s)", srt_getlasterror_str ()));
-    goto failed;
-  }
-
-  /* Make sure TSBPD mode is enable (SRT mode) */
-  srt_setsockopt (sock, 0, SRTO_TSBPDMODE, &(int) {
-      1}, sizeof (int));
-
-  srt_setsockopt (sock, 0, SRTO_SENDER, &sender, sizeof (int));
-
-  srt_setsockopt (sock, 0, SRTO_TSBPDDELAY, &latency, sizeof (int));
-
-  srt_setsockopt (sock, 0, SRTO_RENDEZVOUS, &rendez_vous, sizeof (int));
-
-  if (passphrase != NULL && passphrase[0] != '\0') {
-    srt_setsockopt (sock, 0, SRTO_PASSPHRASE, passphrase, strlen (passphrase));
-    srt_setsockopt (sock, 0, SRTO_PBKEYLEN, &key_length, sizeof (int));
-  }
-
-  if (bind_address || bind_port || rendez_vous) {
-    gpointer bsa;
-    size_t bsa_len;
-    GSocketAddress *b_socket_address = NULL;
-
-    if (bind_address == NULL)
-      bind_address = "0.0.0.0";
-
-    if (rendez_vous)
-      bind_port = port;
-
-    b_socket_address = g_inet_socket_address_new_from_string (bind_address,
-        bind_port);
-
-    if (b_socket_address == NULL) {
-      GST_ELEMENT_ERROR (elem, RESOURCE, OPEN_READ, ("Invalid bind address"),
-          ("Failed to parse bind address: %s:%d", bind_address, bind_port));
-      goto failed;
-    }
-
-    bsa_len = g_socket_address_get_native_size (b_socket_address);
-    bsa = g_alloca (bsa_len);
-    if (!g_socket_address_to_native (b_socket_address, bsa, bsa_len, &error)) {
-      GST_ELEMENT_ERROR (elem, RESOURCE, OPEN_READ, ("Invalid bind address"),
-          ("Can't parse bind address to sockaddr: %s", error->message));
-      g_clear_object (&b_socket_address);
-      goto failed;
-    }
-    g_clear_object (&b_socket_address);
-
-    if (srt_bind (sock, bsa, bsa_len) == SRT_ERROR) {
-      GST_ELEMENT_ERROR (elem, RESOURCE, OPEN_READ,
-          ("Can't bind to address"),
-          ("Can't bind to %s:%d (reason: %s)", bind_address, bind_port,
-              srt_getlasterror_str ()));
-      goto failed;
-    }
-  }
-
-  *poll_id = srt_epoll_create ();
-  if (*poll_id == -1) {
-    GST_ELEMENT_ERROR (elem, LIBRARY, INIT, (NULL),
-        ("failed to create poll id for SRT socket (reason: %s)",
-            srt_getlasterror_str ()));
-    goto failed;
-  }
-
-  srt_epoll_add_usock (*poll_id, sock, &poll_event);
-
-  if (srt_connect (sock, sa, sa_len) == SRT_ERROR) {
-    GST_ELEMENT_ERROR (elem, RESOURCE, OPEN_READ, ("Connection error"),
-        ("failed to connect to host (reason: %s)", srt_getlasterror_str ()));
-    goto failed;
-  }
-
-  return sock;
-
-failed:
-  if (*poll_id != SRT_ERROR) {
-    srt_epoll_release (*poll_id);
-    *poll_id = SRT_ERROR;
-  }
-
-  if (sock != SRT_INVALID_SOCK) {
-    srt_close (sock);
-    sock = SRT_INVALID_SOCK;
-  }
-
-  g_clear_error (&error);
-  g_clear_object (socket_address);
-
-  return SRT_INVALID_SOCK;
-}
-
-SRTSOCKET
-gst_srt_server_listen (GstElement * elem, int sender, const gchar * host,
-    guint16 port, int latency, gint * poll_id, const gchar * passphrase,
-    int key_length)
-{
-  SRTSOCKET sock = SRT_INVALID_SOCK;
-  GError *error = NULL;
-  struct sockaddr sa;
-  size_t sa_len;
-  GSocketAddress *addr = NULL;
-
-  addr = gst_srt_socket_address_new (elem, host, port);
-
-  if (addr == NULL) {
-    GST_WARNING_OBJECT (elem,
-        "failed to extract host or port from the given URI");
-    goto failed;
-  }
-
-  sa_len = g_socket_address_get_native_size (addr);
-  if (!g_socket_address_to_native (addr, &sa, sa_len, &error)) {
-    GST_ELEMENT_ERROR (elem, RESOURCE, OPEN_READ, ("Invalid address"),
-        ("cannot resolve address (reason: %s)", error->message));
-    goto failed;
-  }
-
-  sock = srt_socket (sa.sa_family, SOCK_DGRAM, 0);
-  if (sock == SRT_INVALID_SOCK) {
-    GST_WARNING_OBJECT (elem, "failed to create SRT socket (reason: %s)",
-        srt_getlasterror_str ());
-    goto failed;
-  }
-
-  /* Make SRT server socket non-blocking */
-  /* for non-blocking srt_close() */
-  srt_setsockopt (sock, 0, SRTO_SNDSYN, &(int) {
-      0}, sizeof (int));
-
-  /* for non-blocking srt_accept() */
-  srt_setsockopt (sock, 0, SRTO_RCVSYN, &(int) {
-      0}, sizeof (int));
-
-  /* Make sure TSBPD mode is enable (SRT mode) */
-  srt_setsockopt (sock, 0, SRTO_TSBPDMODE, &(int) {
-      1}, sizeof (int));
-
-  srt_setsockopt (sock, 0, SRTO_SENDER, &sender, sizeof (int));
-  srt_setsockopt (sock, 0, SRTO_TSBPDDELAY, &latency, sizeof (int));
-
-  if (passphrase != NULL && passphrase[0] != '\0') {
-    srt_setsockopt (sock, 0, SRTO_PASSPHRASE, passphrase, strlen (passphrase));
-    srt_setsockopt (sock, 0, SRTO_PBKEYLEN, &key_length, sizeof (int));
-  }
-
-  *poll_id = srt_epoll_create ();
-  if (*poll_id == -1) {
-    GST_ELEMENT_ERROR (elem, LIBRARY, INIT, (NULL),
-        ("failed to create poll id for SRT socket (reason: %s)",
-            srt_getlasterror_str ()));
-    goto failed;
-  }
-
-  srt_epoll_add_usock (*poll_id, sock, &(int) {
-      SRT_EPOLL_IN | SRT_EPOLL_ERR});
-
-  if (srt_bind (sock, &sa, sa_len) == SRT_ERROR) {
-    GST_WARNING_OBJECT (elem, "failed to bind SRT server socket (reason: %s)",
-        srt_getlasterror_str ());
-    goto failed;
-  }
-
-  if (srt_listen (sock, 1) == SRT_ERROR) {
-    GST_WARNING_OBJECT (elem, "failed to listen SRT socket (reason: %s)",
-        srt_getlasterror_str ());
-    goto failed;
-  }
-
-  g_clear_object (&addr);
-
-  return sock;
-
-failed:
-  if (*poll_id != SRT_ERROR) {
-    srt_epoll_release (*poll_id);
-    *poll_id = SRT_ERROR;
-  }
-
-  if (sock != SRT_INVALID_SOCK) {
-    srt_close (sock);
-    sock = SRT_INVALID_SOCK;
-  }
-
-  g_clear_error (&error);
-  g_clear_object (&addr);
-
-  return SRT_INVALID_SOCK;
-}
+#include "gstsrtsrc.h"
+#include "gstsrtsink.h"
 
 static gboolean
 plugin_init (GstPlugin * plugin)
 {
-  GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srt", 0, "SRT Common code");
+  if (!gst_element_register (plugin, "srtsrc", GST_RANK_PRIMARY,
+          GST_TYPE_SRT_SRC))
+    return FALSE;
+
+  if (!gst_element_register (plugin, "srtsink", GST_RANK_PRIMARY,
+          GST_TYPE_SRT_SINK))
+    return FALSE;
 
-  if (!gst_element_register (plugin, "srtclientsrc", GST_RANK_PRIMARY,
-          GST_TYPE_SRT_CLIENT_SRC))
+  /* deprecated */
+  if (!gst_element_register (plugin, "srtclientsrc", GST_RANK_NONE,
+          GST_TYPE_SRT_SRC))
     return FALSE;
 
-  if (!gst_element_register (plugin, "srtserversrc", GST_RANK_PRIMARY,
-          GST_TYPE_SRT_SERVER_SRC))
+  if (!gst_element_register (plugin, "srtserversrc", GST_RANK_NONE,
+          GST_TYPE_SRT_SRC))
     return FALSE;
 
-  if (!gst_element_register (plugin, "srtclientsink", GST_RANK_PRIMARY,
-          GST_TYPE_SRT_CLIENT_SINK))
+  if (!gst_element_register (plugin, "srtclientsink", GST_RANK_NONE,
+          GST_TYPE_SRT_SINK))
     return FALSE;
 
-  if (!gst_element_register (plugin, "srtserversink", GST_RANK_PRIMARY,
-          GST_TYPE_SRT_SERVER_SINK))
+  if (!gst_element_register (plugin, "srtserversink", GST_RANK_NONE,
+          GST_TYPE_SRT_SINK))
     return FALSE;
 
   return TRUE;
diff --git a/ext/srt/gstsrt.h b/ext/srt/gstsrt.h
deleted file mode 100644
index 5f0e973f2..000000000
--- a/ext/srt/gstsrt.h
+++ /dev/null
@@ -1,54 +0,0 @@
-/* GStreamer
- * Copyright (C) 2017, Collabora Ltd.
- *   Author: Olivier Crete <olivier.crete@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifndef __GST_SRT_H__
-#define __GST_SRT_H__
-
-#include <gst/gst.h>
-#include <gio/gio.h>
-#include <gio/gnetworking.h>
-
-#include <srt/srt.h>
-
-#define SRT_URI_SCHEME "srt"
-#define SRT_DEFAULT_PORT 7001
-#define SRT_DEFAULT_HOST "localhost"
-#define SRT_DEFAULT_URI SRT_URI_SCHEME"://"SRT_DEFAULT_HOST":"G_STRINGIFY(SRT_DEFAULT_PORT)
-#define SRT_DEFAULT_LATENCY 125
-#define SRT_DEFAULT_KEY_LENGTH 16
-
-G_BEGIN_DECLS
-
-SRTSOCKET
-gst_srt_client_connect (GstElement * elem, int sender,
-    const gchar * host, guint16 port, int rendez_vous,
-    const gchar * bind_address, guint16 bind_port, int latency,
-    GSocketAddress ** socket_address, gint * poll_id,
-    const gchar * passphrase, int key_length);
-
-SRTSOCKET
-gst_srt_server_listen (GstElement * elem, int sender,
-    const gchar * host, guint16 port, gint latency, gint * poll_id,
-    const gchar * passphrase, int key_length);
-
-G_END_DECLS
-
-
-#endif /* __GST_SRT_H__ */
diff --git a/ext/srt/gstsrtbasesink.c b/ext/srt/gstsrtbasesink.c
deleted file mode 100644
index d5521c8bb..000000000
--- a/ext/srt/gstsrtbasesink.c
+++ /dev/null
@@ -1,441 +0,0 @@
-/* GStreamer SRT plugin based on libsrt
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
-#include "gstsrtserversink.h"
-#include "gstsrt.h"
-#include <srt/srt.h>
-
-#define SRT_DEFAULT_POLL_TIMEOUT -1
-
-#define GST_CAT_DEFAULT gst_debug_srt_base_sink
-GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
-
-enum
-{
-  PROP_URI = 1,
-  PROP_LATENCY,
-  PROP_PASSPHRASE,
-  PROP_KEY_LENGTH,
-
-  /*< private > */
-  PROP_LAST
-};
-
-static GParamSpec *properties[PROP_LAST];
-
-static void gst_srt_base_sink_uri_handler_init (gpointer g_iface,
-    gpointer iface_data);
-static gchar *gst_srt_base_sink_uri_get_uri (GstURIHandler * handler);
-static gboolean gst_srt_base_sink_uri_set_uri (GstURIHandler * handler,
-    const gchar * uri, GError ** error);
-
-#define gst_srt_base_sink_parent_class parent_class
-G_DEFINE_ABSTRACT_TYPE_WITH_CODE (GstSRTBaseSink, gst_srt_base_sink,
-    GST_TYPE_BASE_SINK,
-    G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER,
-        gst_srt_base_sink_uri_handler_init)
-    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtbasesink", 0,
-        "SRT Base Sink"));
-
-static void
-gst_srt_base_sink_get_property (GObject * object,
-    guint prop_id, GValue * value, GParamSpec * pspec)
-{
-  GstSRTBaseSink *self = GST_SRT_BASE_SINK (object);
-
-  switch (prop_id) {
-    case PROP_URI:
-      if (self->uri != NULL) {
-        gchar *uri_str = gst_srt_base_sink_uri_get_uri (GST_URI_HANDLER (self));
-        g_value_take_string (value, uri_str);
-      }
-      break;
-    case PROP_LATENCY:
-      g_value_set_int (value, self->latency);
-      break;
-    case PROP_PASSPHRASE:
-      g_value_set_string (value, self->passphrase);
-      break;
-    case PROP_KEY_LENGTH:
-      g_value_set_int (value, self->key_length);
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_base_sink_set_property (GObject * object,
-    guint prop_id, const GValue * value, GParamSpec * pspec)
-{
-  GstSRTBaseSink *self = GST_SRT_BASE_SINK (object);
-
-  switch (prop_id) {
-    case PROP_URI:
-      gst_srt_base_sink_uri_set_uri (GST_URI_HANDLER (self),
-          g_value_get_string (value), NULL);
-      break;
-    case PROP_LATENCY:
-      self->latency = g_value_get_int (value);
-      break;
-    case PROP_PASSPHRASE:
-      g_free (self->passphrase);
-      self->passphrase = g_value_dup_string (value);
-      break;
-    case PROP_KEY_LENGTH:
-    {
-      gint key_length = g_value_get_int (value);
-      g_return_if_fail (key_length == 16 || key_length == 24
-          || key_length == 32);
-      self->key_length = key_length;
-      break;
-    }
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_base_sink_finalize (GObject * object)
-{
-  GstSRTBaseSink *self = GST_SRT_BASE_SINK (object);
-
-  g_clear_pointer (&self->headers, gst_buffer_list_unref);
-  g_clear_pointer (&self->uri, gst_uri_unref);
-  g_clear_pointer (&self->passphrase, g_free);
-
-  G_OBJECT_CLASS (parent_class)->finalize (object);
-}
-
-static gboolean
-gst_srt_base_sink_set_caps (GstBaseSink * sink, GstCaps * caps)
-{
-  GstSRTBaseSink *self = GST_SRT_BASE_SINK (sink);
-  GstStructure *s;
-  const GValue *streamheader;
-
-  GST_DEBUG_OBJECT (self, "setcaps %" GST_PTR_FORMAT, caps);
-
-  g_clear_pointer (&self->headers, gst_buffer_list_unref);
-
-  s = gst_caps_get_structure (caps, 0);
-  streamheader = gst_structure_get_value (s, "streamheader");
-
-  if (!streamheader) {
-    GST_DEBUG_OBJECT (self, "'streamheader' field not present");
-  } else if (GST_VALUE_HOLDS_BUFFER (streamheader)) {
-    GST_DEBUG_OBJECT (self, "'streamheader' field holds buffer");
-    self->headers = gst_buffer_list_new_sized (1);
-    gst_buffer_list_add (self->headers, g_value_dup_boxed (streamheader));
-  } else if (GST_VALUE_HOLDS_ARRAY (streamheader)) {
-    guint i, size;
-
-    GST_DEBUG_OBJECT (self, "'streamheader' field holds array");
-
-    size = gst_value_array_get_size (streamheader);
-    self->headers = gst_buffer_list_new_sized (size);
-
-    for (i = 0; i < size; i++) {
-      const GValue *v = gst_value_array_get_value (streamheader, i);
-      if (!GST_VALUE_HOLDS_BUFFER (v)) {
-        GST_ERROR_OBJECT (self, "'streamheader' item of unexpected type '%s'",
-            G_VALUE_TYPE_NAME (v));
-        return FALSE;
-      }
-
-      gst_buffer_list_add (self->headers, g_value_dup_boxed (v));
-    }
-  } else {
-    GST_ERROR_OBJECT (self, "'streamheader' field has unexpected type '%s'",
-        G_VALUE_TYPE_NAME (streamheader));
-    return FALSE;
-  }
-
-  GST_DEBUG_OBJECT (self, "Collected streamheaders: %u buffers",
-      self->headers ? gst_buffer_list_length (self->headers) : 0);
-
-  return TRUE;
-}
-
-static gboolean
-gst_srt_base_sink_stop (GstBaseSink * sink)
-{
-  GstSRTBaseSink *self = GST_SRT_BASE_SINK (sink);
-
-  g_clear_pointer (&self->headers, gst_buffer_list_unref);
-
-  return TRUE;
-}
-
-static GstFlowReturn
-gst_srt_base_sink_render (GstBaseSink * sink, GstBuffer * buffer)
-{
-  GstSRTBaseSink *self = GST_SRT_BASE_SINK (sink);
-  GstMapInfo info;
-  GstSRTBaseSinkClass *bclass = GST_SRT_BASE_SINK_GET_CLASS (sink);
-  GstFlowReturn ret = GST_FLOW_OK;
-
-  if (self->headers && GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_HEADER)) {
-    GST_DEBUG_OBJECT (self, "Have streamheaders,"
-        " ignoring header %" GST_PTR_FORMAT, buffer);
-    return GST_FLOW_OK;
-  }
-
-  GST_TRACE_OBJECT (self, "sending buffer %p, offset %"
-      G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT
-      ", timestamp %" GST_TIME_FORMAT ", duration %" GST_TIME_FORMAT
-      ", size %" G_GSIZE_FORMAT,
-      buffer, GST_BUFFER_OFFSET (buffer),
-      GST_BUFFER_OFFSET_END (buffer),
-      GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
-      GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)),
-      gst_buffer_get_size (buffer));
-
-  if (!gst_buffer_map (buffer, &info, GST_MAP_READ)) {
-    GST_ELEMENT_ERROR (self, RESOURCE, READ,
-        ("Could not map the input stream"), (NULL));
-    return GST_FLOW_ERROR;
-  }
-
-  if (!bclass->send_buffer (self, &info))
-    ret = GST_FLOW_ERROR;
-
-  gst_buffer_unmap (buffer, &info);
-
-  return ret;
-}
-
-static void
-gst_srt_base_sink_class_init (GstSRTBaseSinkClass * klass)
-{
-  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
-  GstBaseSinkClass *gstbasesink_class = GST_BASE_SINK_CLASS (klass);
-
-  gobject_class->set_property = gst_srt_base_sink_set_property;
-  gobject_class->get_property = gst_srt_base_sink_get_property;
-  gobject_class->finalize = gst_srt_base_sink_finalize;
-
-  /**
-   * GstSRTBaseSink:uri:
-   *
-   * The URI used by SRT Connection.
-   */
-  properties[PROP_URI] = g_param_spec_string ("uri", "URI",
-      "URI in the form of srt://address:port", SRT_DEFAULT_URI,
-      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_LATENCY] =
-      g_param_spec_int ("latency", "latency",
-      "Minimum latency (milliseconds)", 0,
-      G_MAXINT32, SRT_DEFAULT_LATENCY,
-      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_PASSPHRASE] = g_param_spec_string ("passphrase", "Passphrase",
-      "The password for the encrypted transmission", NULL,
-      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_KEY_LENGTH] =
-      g_param_spec_int ("key-length", "key length",
-      "Crypto key length in bytes{16,24,32}", 16,
-      32, SRT_DEFAULT_KEY_LENGTH, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  g_object_class_install_properties (gobject_class, PROP_LAST, properties);
-
-  gstbasesink_class->set_caps = GST_DEBUG_FUNCPTR (gst_srt_base_sink_set_caps);
-  gstbasesink_class->stop = GST_DEBUG_FUNCPTR (gst_srt_base_sink_stop);
-  gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_srt_base_sink_render);
-}
-
-static void
-gst_srt_base_sink_init (GstSRTBaseSink * self)
-{
-  self->uri = gst_uri_from_string (SRT_DEFAULT_URI);
-  self->latency = SRT_DEFAULT_LATENCY;
-  self->passphrase = NULL;
-  self->key_length = SRT_DEFAULT_KEY_LENGTH;
-}
-
-static GstURIType
-gst_srt_base_sink_uri_get_type (GType type)
-{
-  return GST_URI_SINK;
-}
-
-static const gchar *const *
-gst_srt_base_sink_uri_get_protocols (GType type)
-{
-  static const gchar *protocols[] = { SRT_URI_SCHEME, NULL };
-
-  return protocols;
-}
-
-static gchar *
-gst_srt_base_sink_uri_get_uri (GstURIHandler * handler)
-{
-  gchar *uri_str;
-  GstSRTBaseSink *self = GST_SRT_BASE_SINK (handler);
-
-  GST_OBJECT_LOCK (self);
-  uri_str = gst_uri_to_string (self->uri);
-  GST_OBJECT_UNLOCK (self);
-
-  return uri_str;
-}
-
-static gboolean
-gst_srt_base_sink_uri_set_uri (GstURIHandler * handler,
-    const gchar * uri, GError ** error)
-{
-  GstSRTBaseSink *self = GST_SRT_BASE_SINK (handler);
-  gboolean ret = TRUE;
-  GstUri *parsed_uri = gst_uri_from_string (uri);
-
-  GST_TRACE_OBJECT (self, "Requested URI=%s", uri);
-
-  if (g_strcmp0 (gst_uri_get_scheme (parsed_uri), SRT_URI_SCHEME) != 0) {
-    g_set_error (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
-        "Invalid SRT URI scheme");
-    ret = FALSE;
-    goto out;
-  }
-
-  GST_OBJECT_LOCK (self);
-
-  g_clear_pointer (&self->uri, gst_uri_unref);
-  self->uri = gst_uri_ref (parsed_uri);
-
-  GST_OBJECT_UNLOCK (self);
-
-out:
-  g_clear_pointer (&parsed_uri, gst_uri_unref);
-  return ret;
-}
-
-static void
-gst_srt_base_sink_uri_handler_init (gpointer g_iface, gpointer iface_data)
-{
-  GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
-
-  iface->get_type = gst_srt_base_sink_uri_get_type;
-  iface->get_protocols = gst_srt_base_sink_uri_get_protocols;
-  iface->get_uri = gst_srt_base_sink_uri_get_uri;
-  iface->set_uri = gst_srt_base_sink_uri_set_uri;
-}
-
-gboolean
-gst_srt_base_sink_send_headers (GstSRTBaseSink * self,
-    GstSRTBaseSinkSendCallback send_cb, gpointer user_data)
-{
-  guint size, i;
-
-  g_return_val_if_fail (GST_IS_SRT_BASE_SINK (self), FALSE);
-  g_return_val_if_fail (send_cb, FALSE);
-
-  if (!self->headers)
-    return TRUE;
-
-  size = gst_buffer_list_length (self->headers);
-
-  GST_DEBUG_OBJECT (self, "Sending %u stream headers", size);
-
-  for (i = 0; i < size; i++) {
-    GstBuffer *buffer = gst_buffer_list_get (self->headers, i);
-    GstMapInfo info;
-    gboolean ret;
-
-    GST_TRACE_OBJECT (self, "sending header %u %" GST_PTR_FORMAT, i, buffer);
-
-    if (!gst_buffer_map (buffer, &info, GST_MAP_READ)) {
-      GST_ELEMENT_ERROR (self, RESOURCE, READ,
-          ("Could not map the input stream"), (NULL));
-      return FALSE;
-    }
-
-    ret = send_cb (self, &info, user_data);
-
-    gst_buffer_unmap (buffer, &info);
-
-    if (!ret)
-      return FALSE;
-  }
-
-  return TRUE;
-}
-
-GstStructure *
-gst_srt_base_sink_get_stats (GSocketAddress * sockaddr, SRTSOCKET sock)
-{
-  SRT_TRACEBSTATS stats;
-  int ret;
-  GValue v = G_VALUE_INIT;
-  GstStructure *s;
-
-  if (sock == SRT_INVALID_SOCK || sockaddr == NULL)
-    return gst_structure_new_empty ("application/x-srt-statistics");
-
-  s = gst_structure_new ("application/x-srt-statistics",
-      "sockaddr", G_TYPE_SOCKET_ADDRESS, sockaddr, NULL);
-
-  ret = srt_bstats (sock, &stats, 0);
-  if (ret >= 0) {
-    gst_structure_set (s,
-        /* number of sent data packets, including retransmissions */
-        "packets-sent", G_TYPE_INT64, stats.pktSent,
-        /* number of lost packets (sender side) */
-        "packets-sent-lost", G_TYPE_INT, stats.pktSndLoss,
-        /* number of retransmitted packets */
-        "packets-retransmitted", G_TYPE_INT, stats.pktRetrans,
-        /* number of received ACK packets */
-        "packet-ack-received", G_TYPE_INT, stats.pktRecvACK,
-        /* number of received NAK packets */
-        "packet-nack-received", G_TYPE_INT, stats.pktRecvNAK,
-        /* time duration when UDT is sending data (idle time exclusive) */
-        "send-duration-us", G_TYPE_INT64, stats.usSndDuration,
-        /* number of sent data bytes, including retransmissions */
-        "bytes-sent", G_TYPE_UINT64, stats.byteSent,
-        /* number of retransmitted bytes */
-        "bytes-retransmitted", G_TYPE_UINT64, stats.byteRetrans,
-        /* number of too-late-to-send dropped bytes */
-        "bytes-sent-dropped", G_TYPE_UINT64, stats.byteSndDrop,
-        /* number of too-late-to-send dropped packets */
-        "packets-sent-dropped", G_TYPE_INT, stats.pktSndDrop,
-        /* sending rate in Mb/s */
-        "send-rate-mbps", G_TYPE_DOUBLE, stats.msRTT,
-        /* estimated bandwidth, in Mb/s */
-        "bandwidth-mbps", G_TYPE_DOUBLE, stats.mbpsBandwidth,
-        /* busy sending time (i.e., idle time exclusive) */
-        "send-duration-us", G_TYPE_UINT64, stats.usSndDuration,
-        "rtt-ms", G_TYPE_DOUBLE, stats.msRTT,
-        "negotiated-latency-ms", G_TYPE_INT, stats.msSndTsbPdDelay, NULL);
-  }
-
-  g_value_init (&v, G_TYPE_STRING);
-  g_value_take_string (&v,
-      g_socket_connectable_to_string (G_SOCKET_CONNECTABLE (sockaddr)));
-  gst_structure_take_value (s, "sockaddr-str", &v);
-
-  return s;
-}
diff --git a/ext/srt/gstsrtbasesink.h b/ext/srt/gstsrtbasesink.h
deleted file mode 100644
index 55b40a1ae..000000000
--- a/ext/srt/gstsrtbasesink.h
+++ /dev/null
@@ -1,81 +0,0 @@
-/* GStreamer
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifndef __GST_SRT_BASE_SINK_H__
-#define __GST_SRT_BASE_SINK_H__
-
-#include <gst/gst.h>
-#include <gst/base/gstbasesink.h>
-#include <gio/gio.h>
-
-#include <srt/srt.h>
-
-G_BEGIN_DECLS
-
-#define GST_TYPE_SRT_BASE_SINK              (gst_srt_base_sink_get_type ())
-#define GST_IS_SRT_BASE_SINK(obj)           (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SRT_BASE_SINK))
-#define GST_IS_SRT_BASE_SINK_CLASS(klass)   (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SRT_BASE_SINK))
-#define GST_SRT_BASE_SINK_GET_CLASS(obj)    (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_SRT_BASE_SINK, GstSRTBaseSinkClass))
-#define GST_SRT_BASE_SINK(obj)              (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SRT_BASE_SINK, GstSRTBaseSink))
-#define GST_SRT_BASE_SINK_CLASS(klass)      (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SRT_BASE_SINK, GstSRTBaseSinkClass))
-#define GST_SRT_BASE_SINK_CAST(obj)         ((GstSRTBaseSink*)(obj))
-#define GST_SRT_BASE_SINK_CLASS_CAST(klass) ((GstSRTBaseSinkClass*)(klass))
-
-typedef struct _GstSRTBaseSink GstSRTBaseSink;
-typedef struct _GstSRTBaseSinkClass GstSRTBaseSinkClass;
-
-struct _GstSRTBaseSink {
-  GstBaseSink parent;
-
-  GstUri *uri;
-  GstBufferList *headers;
-  gint latency;
-  gchar *passphrase;
-  gint key_length;
-
-  /*< private >*/
-  gpointer _gst_reserved[GST_PADDING];
-};
-
-struct _GstSRTBaseSinkClass {
-  GstBaseSinkClass parent_class;
-
-  /* ask the subclass to send a buffer */
-  gboolean (*send_buffer)       (GstSRTBaseSink *self, const GstMapInfo *mapinfo);
-
-  gpointer _gst_reserved[GST_PADDING_LARGE];
-};
-
-GST_EXPORT
-GType gst_srt_base_sink_get_type (void);
-
-typedef gboolean (*GstSRTBaseSinkSendCallback) (GstSRTBaseSink *sink,
-    const GstMapInfo *mapinfo, gpointer user_data);
-
-gboolean gst_srt_base_sink_send_headers (GstSRTBaseSink *sink,
-    GstSRTBaseSinkSendCallback send_cb, gpointer user_data);
-
-GstStructure * gst_srt_base_sink_get_stats (GSocketAddress *sockaddr,
-    SRTSOCKET sock);
-
-
-G_END_DECLS
-
-#endif /* __GST_SRT_BASE_SINK_H__ */
diff --git a/ext/srt/gstsrtbasesrc.c b/ext/srt/gstsrtbasesrc.c
deleted file mode 100644
index 44ca40f5a..000000000
--- a/ext/srt/gstsrtbasesrc.c
+++ /dev/null
@@ -1,293 +0,0 @@
-/* GStreamer SRT plugin based on libsrt
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
-#include "gstsrtbasesrc.h"
-#include "gstsrt.h"
-#include <srt/srt.h>
-#include <gio/gio.h>
-
-#define GST_CAT_DEFAULT gst_debug_srt_base_src
-GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
-
-enum
-{
-  PROP_URI = 1,
-  PROP_CAPS,
-  PROP_LATENCY,
-  PROP_PASSPHRASE,
-  PROP_KEY_LENGTH,
-
-  /*< private > */
-  PROP_LAST
-};
-
-static GParamSpec *properties[PROP_LAST];
-
-static void gst_srt_base_src_uri_handler_init (gpointer g_iface,
-    gpointer iface_data);
-static gchar *gst_srt_base_src_uri_get_uri (GstURIHandler * handler);
-static gboolean gst_srt_base_src_uri_set_uri (GstURIHandler * handler,
-    const gchar * uri, GError ** error);
-
-#define gst_srt_base_src_parent_class parent_class
-G_DEFINE_ABSTRACT_TYPE_WITH_CODE (GstSRTBaseSrc, gst_srt_base_src,
-    GST_TYPE_PUSH_SRC, G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER,
-        gst_srt_base_src_uri_handler_init)
-    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtbasesrc", 0,
-        "SRT Base Source"));
-
-static void
-gst_srt_base_src_get_property (GObject * object,
-    guint prop_id, GValue * value, GParamSpec * pspec)
-{
-  GstSRTBaseSrc *self = GST_SRT_BASE_SRC (object);
-
-  switch (prop_id) {
-    case PROP_URI:
-      if (self->uri != NULL) {
-        gchar *uri_str = gst_srt_base_src_uri_get_uri (GST_URI_HANDLER (self));
-        g_value_take_string (value, uri_str);
-      }
-      break;
-    case PROP_CAPS:
-      GST_OBJECT_LOCK (self);
-      gst_value_set_caps (value, self->caps);
-      GST_OBJECT_UNLOCK (self);
-      break;
-    case PROP_LATENCY:
-      g_value_set_int (value, self->latency);
-      break;
-    case PROP_PASSPHRASE:
-      g_value_set_string (value, self->passphrase);
-      break;
-    case PROP_KEY_LENGTH:
-      g_value_set_int (value, self->key_length);
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_base_src_set_property (GObject * object,
-    guint prop_id, const GValue * value, GParamSpec * pspec)
-{
-  GstSRTBaseSrc *self = GST_SRT_BASE_SRC (object);
-
-  switch (prop_id) {
-    case PROP_URI:
-      gst_srt_base_src_uri_set_uri (GST_URI_HANDLER (self),
-          g_value_get_string (value), NULL);
-      break;
-    case PROP_CAPS:
-      GST_OBJECT_LOCK (self);
-      g_clear_pointer (&self->caps, gst_caps_unref);
-      self->caps = gst_caps_copy (gst_value_get_caps (value));
-      GST_OBJECT_UNLOCK (self);
-      break;
-    case PROP_LATENCY:
-      self->latency = g_value_get_int (value);
-      break;
-    case PROP_PASSPHRASE:
-      g_free (self->passphrase);
-      self->passphrase = g_value_dup_string (value);
-      break;
-    case PROP_KEY_LENGTH:
-    {
-      gint key_length = g_value_get_int (value);
-      g_return_if_fail (key_length == 16 || key_length == 24
-          || key_length == 32);
-      self->key_length = key_length;
-      break;
-    }
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_base_src_finalize (GObject * object)
-{
-  GstSRTBaseSrc *self = GST_SRT_BASE_SRC (object);
-
-  g_clear_pointer (&self->uri, gst_uri_unref);
-  g_clear_pointer (&self->caps, gst_caps_unref);
-  g_clear_pointer (&self->passphrase, g_free);
-
-  G_OBJECT_CLASS (parent_class)->finalize (object);
-}
-
-static GstCaps *
-gst_srt_base_src_get_caps (GstBaseSrc * src, GstCaps * filter)
-{
-  GstSRTBaseSrc *self = GST_SRT_BASE_SRC (src);
-  GstCaps *result, *caps = NULL;
-
-  GST_OBJECT_LOCK (self);
-  if (self->caps != NULL) {
-    caps = gst_caps_ref (self->caps);
-  }
-  GST_OBJECT_UNLOCK (self);
-
-  if (caps) {
-    if (filter) {
-      result = gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
-      gst_caps_unref (caps);
-    } else {
-      result = caps;
-    }
-  } else {
-    result = (filter) ? gst_caps_ref (filter) : gst_caps_new_any ();
-  }
-
-  return result;
-}
-
-
-static void
-gst_srt_base_src_class_init (GstSRTBaseSrcClass * klass)
-{
-  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
-  GstBaseSrcClass *gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
-
-  gobject_class->set_property = gst_srt_base_src_set_property;
-  gobject_class->get_property = gst_srt_base_src_get_property;
-  gobject_class->finalize = gst_srt_base_src_finalize;
-
-  /**
-   * GstSRTBaseSrc:uri:
-   *
-   * The URI used by SRT Connection.
-   */
-  properties[PROP_URI] = g_param_spec_string ("uri", "URI",
-      "URI in the form of srt://address:port", SRT_DEFAULT_URI,
-      G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS);
-
-  /**
-   * GstSRTBaseSrc:caps:
-   *
-   * The Caps used by the source pad.
-   */
-  properties[PROP_CAPS] =
-      g_param_spec_boxed ("caps", "Caps", "The caps of the source pad",
-      GST_TYPE_CAPS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_LATENCY] =
-      g_param_spec_int ("latency", "latency",
-      "Minimum latency (milliseconds)", 0,
-      G_MAXINT32, SRT_DEFAULT_LATENCY,
-      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_PASSPHRASE] = g_param_spec_string ("passphrase", "Passphrase",
-      "The password for the encrypted transmission", NULL,
-      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_KEY_LENGTH] =
-      g_param_spec_int ("key-length", "key length",
-      "Crypto key length in bytes{16,24,32}", 16,
-      32, SRT_DEFAULT_KEY_LENGTH, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  g_object_class_install_properties (gobject_class, PROP_LAST, properties);
-
-  gstbasesrc_class->get_caps = GST_DEBUG_FUNCPTR (gst_srt_base_src_get_caps);
-}
-
-static void
-gst_srt_base_src_init (GstSRTBaseSrc * self)
-{
-  gst_srt_base_src_uri_set_uri (GST_URI_HANDLER (self), SRT_DEFAULT_URI, NULL);
-  gst_base_src_set_format (GST_BASE_SRC (self), GST_FORMAT_TIME);
-  gst_base_src_set_live (GST_BASE_SRC (self), TRUE);
-  gst_base_src_set_do_timestamp (GST_BASE_SRC (self), TRUE);
-
-  self->latency = SRT_DEFAULT_LATENCY;
-  self->key_length = SRT_DEFAULT_KEY_LENGTH;
-}
-
-static GstURIType
-gst_srt_base_src_uri_get_type (GType type)
-{
-  return GST_URI_SRC;
-}
-
-static const gchar *const *
-gst_srt_base_src_uri_get_protocols (GType type)
-{
-  static const gchar *protocols[] = { SRT_URI_SCHEME, NULL };
-
-  return protocols;
-}
-
-static gchar *
-gst_srt_base_src_uri_get_uri (GstURIHandler * handler)
-{
-  gchar *uri_str;
-  GstSRTBaseSrc *self = GST_SRT_BASE_SRC (handler);
-
-  GST_OBJECT_LOCK (self);
-  uri_str = gst_uri_to_string (self->uri);
-  GST_OBJECT_UNLOCK (self);
-
-  return uri_str;
-}
-
-static gboolean
-gst_srt_base_src_uri_set_uri (GstURIHandler * handler,
-    const gchar * uri, GError ** error)
-{
-  GstSRTBaseSrc *self = GST_SRT_BASE_SRC (handler);
-  gboolean ret = TRUE;
-  GstUri *parsed_uri = gst_uri_from_string (uri);
-
-  if (g_strcmp0 (gst_uri_get_scheme (parsed_uri), SRT_URI_SCHEME) != 0) {
-    g_set_error (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
-        "Invalid SRT URI scheme");
-    ret = FALSE;
-    goto out;
-  }
-
-  GST_OBJECT_LOCK (self);
-
-  g_clear_pointer (&self->uri, gst_uri_unref);
-  self->uri = gst_uri_ref (parsed_uri);
-
-  GST_OBJECT_UNLOCK (self);
-
-out:
-  g_clear_pointer (&parsed_uri, gst_uri_unref);
-  return ret;
-}
-
-static void
-gst_srt_base_src_uri_handler_init (gpointer g_iface, gpointer iface_data)
-{
-  GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
-
-  iface->get_type = gst_srt_base_src_uri_get_type;
-  iface->get_protocols = gst_srt_base_src_uri_get_protocols;
-  iface->get_uri = gst_srt_base_src_uri_get_uri;
-  iface->set_uri = gst_srt_base_src_uri_set_uri;
-}
diff --git a/ext/srt/gstsrtbasesrc.h b/ext/srt/gstsrtbasesrc.h
deleted file mode 100644
index 270a1c460..000000000
--- a/ext/srt/gstsrtbasesrc.h
+++ /dev/null
@@ -1,66 +0,0 @@
-/* GStreamer
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifndef __GST_SRT_BASE_SRC_H__
-#define __GST_SRT_BASE_SRC_H__
-
-#include <gst/gst.h>
-#include <gst/base/gstpushsrc.h>
-
-G_BEGIN_DECLS
-
-#define GST_TYPE_SRT_BASE_SRC              (gst_srt_base_src_get_type ())
-#define GST_IS_SRT_BASE_SRC(obj)           (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SRT_BASE_SRC))
-#define GST_IS_SRT_BASE_SRC_CLASS(klass)   (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SRT_BASE_SRC))
-#define GST_SRT_BASE_SRC_GET_CLASS(obj)    (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_SRT_BASE_SRC, GstSRTBaseSrcClass))
-#define GST_SRT_BASE_SRC(obj)              (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SRT_BASE_SRC, GstSRTBaseSrc))
-#define GST_SRT_BASE_SRC_CLASS(klass)      (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SRT_BASE_SRC, GstSRTBaseSrcClass))
-#define GST_SRT_BASE_SRC_CAST(obj)         ((GstSRTBaseSrc*)(obj))
-#define GST_SRT_BASE_SRC_CLASS_CAST(klass) ((GstSRTBaseSrcClass*)(klass))
-
-typedef struct _GstSRTBaseSrc GstSRTBaseSrc;
-typedef struct _GstSRTBaseSrcClass GstSRTBaseSrcClass;
-
-struct _GstSRTBaseSrc {
-  GstPushSrc parent;
-
-  GstUri *uri;
-  GstCaps *caps;
-  gint latency;
-  gchar *passphrase;
-  gint key_length;
-
-
-  /*< private >*/
-  gpointer _gst_reserved[GST_PADDING];
-};
-
-struct _GstSRTBaseSrcClass {
-  GstPushSrcClass parent_class;
-
-  gpointer _gst_reserved[GST_PADDING_LARGE];
-};
-
-GST_EXPORT
-GType gst_srt_base_src_get_type (void);
-
-G_END_DECLS
-
-#endif /* __GST_SRT_BASE_SRC_H__ */
diff --git a/ext/srt/gstsrtclientsink.c b/ext/srt/gstsrtclientsink.c
deleted file mode 100644
index ab3ef4a64..000000000
--- a/ext/srt/gstsrtclientsink.c
+++ /dev/null
@@ -1,301 +0,0 @@
-/* GStreamer SRT plugin based on libsrt
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-/**
- * SECTION:element-srtserversink
- * @title: srtserversink
- *
- * srtserversink is a network sink that sends <ulink url="http://www.srtalliance.org/">SRT</ulink>
- * packets to the network. Although SRT is an UDP-based protocol, srtserversink works like
- * a server socket of connection-oriented protocol.
- *
- * <refsect2>
- * <title>Examples</title>
- * |[
- * gst-launch-1.0 -v audiotestsrc ! srtserversink
- * ]| This pipeline shows how to serve SRT packets through the default port.
-
- * |[
- * gst-launch-1.0 -v audiotestsrc ! srtserversink uri=srt://192.168.1.10:8888/ rendez-vous=1
- * ]| This pipeline shows how to serve SRT packets to 192.168.1.10 port 8888 using the rendez-vous mode.
- * </refsect2>
- *
- */
-
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
-#include "gstsrtclientsink.h"
-#include "gstsrt.h"
-#include <srt/srt.h>
-#include <gio/gio.h>
-
-#define SRT_DEFAULT_POLL_TIMEOUT -1
-
-static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
-    GST_PAD_SINK,
-    GST_PAD_ALWAYS,
-    GST_STATIC_CAPS_ANY);
-
-#define GST_CAT_DEFAULT gst_debug_srt_client_sink
-GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
-
-struct _GstSRTClientSinkPrivate
-{
-  SRTSOCKET sock;
-  GSocketAddress *sockaddr;
-  gint poll_id;
-  gint poll_timeout;
-
-  gboolean rendez_vous;
-  gchar *bind_address;
-  guint16 bind_port;
-
-  gboolean sent_headers;
-};
-
-#define GST_SRT_CLIENT_SINK_GET_PRIVATE(obj)  \
-       (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_SRT_CLIENT_SINK, GstSRTClientSinkPrivate))
-
-enum
-{
-  PROP_POLL_TIMEOUT = 1,
-  PROP_BIND_ADDRESS,
-  PROP_BIND_PORT,
-  PROP_RENDEZ_VOUS,
-  PROP_STATS,
-  /*< private > */
-  PROP_LAST
-};
-
-static GParamSpec *properties[PROP_LAST];
-
-#define gst_srt_client_sink_parent_class parent_class
-G_DEFINE_TYPE_WITH_CODE (GstSRTClientSink, gst_srt_client_sink,
-    GST_TYPE_SRT_BASE_SINK, G_ADD_PRIVATE (GstSRTClientSink)
-    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtclientsink", 0,
-        "SRT Client Sink"));
-
-static void
-gst_srt_client_sink_get_property (GObject * object,
-    guint prop_id, GValue * value, GParamSpec * pspec)
-{
-  GstSRTClientSink *self = GST_SRT_CLIENT_SINK (object);
-  GstSRTClientSinkPrivate *priv = GST_SRT_CLIENT_SINK_GET_PRIVATE (self);
-
-  switch (prop_id) {
-    case PROP_POLL_TIMEOUT:
-      g_value_set_int (value, priv->poll_timeout);
-      break;
-    case PROP_BIND_PORT:
-      g_value_set_int (value, priv->bind_port);
-      break;
-    case PROP_BIND_ADDRESS:
-      g_value_set_string (value, priv->bind_address);
-      break;
-    case PROP_RENDEZ_VOUS:
-      g_value_set_boolean (value, priv->rendez_vous);
-      break;
-    case PROP_STATS:
-      g_value_take_boxed (value, gst_srt_base_sink_get_stats (priv->sockaddr,
-              priv->sock));
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_client_sink_set_property (GObject * object,
-    guint prop_id, const GValue * value, GParamSpec * pspec)
-{
-  GstSRTClientSink *self = GST_SRT_CLIENT_SINK (object);
-  GstSRTClientSinkPrivate *priv = GST_SRT_CLIENT_SINK_GET_PRIVATE (self);
-
-  switch (prop_id) {
-    case PROP_POLL_TIMEOUT:
-      priv->poll_timeout = g_value_get_int (value);
-      break;
-    case PROP_BIND_ADDRESS:
-      g_free (priv->bind_address);
-      priv->bind_address = g_value_dup_string (value);
-      break;
-    case PROP_BIND_PORT:
-      priv->bind_port = g_value_get_int (value);
-      break;
-    case PROP_RENDEZ_VOUS:
-      priv->rendez_vous = g_value_get_boolean (value);
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_client_sink_finalize (GObject * object)
-{
-  GstSRTClientSink *self = GST_SRT_CLIENT_SINK (object);
-  GstSRTClientSinkPrivate *priv = GST_SRT_CLIENT_SINK_GET_PRIVATE (self);
-
-  g_free (priv->bind_address);
-
-  G_OBJECT_CLASS (parent_class)->finalize (object);
-}
-
-static gboolean
-gst_srt_client_sink_start (GstBaseSink * sink)
-{
-  GstSRTClientSink *self = GST_SRT_CLIENT_SINK (sink);
-  GstSRTClientSinkPrivate *priv = GST_SRT_CLIENT_SINK_GET_PRIVATE (self);
-  GstSRTBaseSink *base = GST_SRT_BASE_SINK (sink);
-  GstUri *uri = gst_uri_ref (GST_SRT_BASE_SINK (self)->uri);
-
-  priv->sock = gst_srt_client_connect (GST_ELEMENT (sink), TRUE,
-      gst_uri_get_host (uri), gst_uri_get_port (uri), priv->rendez_vous,
-      priv->bind_address, priv->bind_port, base->latency,
-      &priv->sockaddr, &priv->poll_id, base->passphrase, base->key_length);
-
-  g_clear_pointer (&uri, gst_uri_unref);
-
-  return (priv->sock != SRT_INVALID_SOCK);
-}
-
-static gboolean
-send_buffer_internal (GstSRTBaseSink * sink,
-    const GstMapInfo * mapinfo, gpointer user_data)
-{
-  SRTSOCKET sock = GPOINTER_TO_INT (user_data);
-
-  if (srt_sendmsg2 (sock, (char *) mapinfo->data, mapinfo->size,
-          0) == SRT_ERROR) {
-    GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, NULL,
-        ("%s", srt_getlasterror_str ()));
-    return FALSE;
-  }
-
-  return TRUE;
-}
-
-static gboolean
-gst_srt_client_sink_send_buffer (GstSRTBaseSink * sink,
-    const GstMapInfo * mapinfo)
-{
-  GstSRTClientSink *self = GST_SRT_CLIENT_SINK (sink);
-  GstSRTClientSinkPrivate *priv = GST_SRT_CLIENT_SINK_GET_PRIVATE (self);
-
-  if (!priv->sent_headers) {
-    if (!gst_srt_base_sink_send_headers (sink, send_buffer_internal,
-            GINT_TO_POINTER (priv->sock)))
-      return FALSE;
-
-    priv->sent_headers = TRUE;
-  }
-
-  return send_buffer_internal (sink, mapinfo, GINT_TO_POINTER (priv->sock));
-}
-
-static gboolean
-gst_srt_client_sink_stop (GstBaseSink * sink)
-{
-  GstSRTClientSink *self = GST_SRT_CLIENT_SINK (sink);
-  GstSRTClientSinkPrivate *priv = GST_SRT_CLIENT_SINK_GET_PRIVATE (self);
-
-  GST_DEBUG_OBJECT (self, "closing SRT connection");
-
-  if (priv->poll_id != SRT_ERROR) {
-    srt_epoll_remove_usock (priv->poll_id, priv->sock);
-    srt_epoll_release (priv->poll_id);
-    priv->poll_id = SRT_ERROR;
-  }
-
-  if (priv->sock != SRT_INVALID_SOCK) {
-    srt_close (priv->sock);
-    priv->sock = SRT_INVALID_SOCK;
-  }
-
-  g_clear_object (&priv->sockaddr);
-
-  priv->sent_headers = FALSE;
-
-  return GST_BASE_SINK_CLASS (parent_class)->stop (sink);
-}
-
-static void
-gst_srt_client_sink_class_init (GstSRTClientSinkClass * klass)
-{
-  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
-  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
-  GstBaseSinkClass *gstbasesink_class = GST_BASE_SINK_CLASS (klass);
-  GstSRTBaseSinkClass *gstsrtbasesink_class = GST_SRT_BASE_SINK_CLASS (klass);
-
-  gobject_class->set_property = gst_srt_client_sink_set_property;
-  gobject_class->get_property = gst_srt_client_sink_get_property;
-  gobject_class->finalize = gst_srt_client_sink_finalize;
-
-  properties[PROP_POLL_TIMEOUT] =
-      g_param_spec_int ("poll-timeout", "Poll Timeout",
-      "Return poll wait after timeout miliseconds (-1 = infinite)", -1,
-      G_MAXINT32, SRT_DEFAULT_POLL_TIMEOUT,
-      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_BIND_ADDRESS] =
-      g_param_spec_string ("bind-address", "Bind Address",
-      "Address to bind socket to (required for rendez-vous mode) ", NULL,
-      G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_BIND_PORT] =
-      g_param_spec_int ("bind-port", "Bind Port",
-      "Port to bind socket to (Ignored in rendez-vous mode)", 0,
-      G_MAXUINT16, 0,
-      G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_RENDEZ_VOUS] =
-      g_param_spec_boolean ("rendez-vous", "Rendez Vous",
-      "Work in Rendez-Vous mode instead of client/caller mode", FALSE,
-      G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_STATS] = g_param_spec_boxed ("stats", "Statistics",
-      "SRT Statistics", GST_TYPE_STRUCTURE,
-      G_PARAM_READABLE | G_PARAM_STATIC_STRINGS);
-
-  g_object_class_install_properties (gobject_class, PROP_LAST, properties);
-
-  gst_element_class_add_static_pad_template (gstelement_class, &sink_template);
-  gst_element_class_set_metadata (gstelement_class,
-      "SRT client sink", "Sink/Network",
-      "Send data over the network via SRT",
-      "Justin Kim <justin.kim@collabora.com>");
-
-  gstbasesink_class->start = GST_DEBUG_FUNCPTR (gst_srt_client_sink_start);
-  gstbasesink_class->stop = GST_DEBUG_FUNCPTR (gst_srt_client_sink_stop);
-
-  gstsrtbasesink_class->send_buffer =
-      GST_DEBUG_FUNCPTR (gst_srt_client_sink_send_buffer);
-}
-
-static void
-gst_srt_client_sink_init (GstSRTClientSink * self)
-{
-  GstSRTClientSinkPrivate *priv = GST_SRT_CLIENT_SINK_GET_PRIVATE (self);
-  priv->poll_timeout = SRT_DEFAULT_POLL_TIMEOUT;
-}
diff --git a/ext/srt/gstsrtclientsink.h b/ext/srt/gstsrtclientsink.h
deleted file mode 100644
index ecc439ec6..000000000
--- a/ext/srt/gstsrtclientsink.h
+++ /dev/null
@@ -1,59 +0,0 @@
-/* GStreamer
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifndef __GST_SRT_CLIENT_SINK_H__
-#define __GST_SRT_CLIENT_SINK_H__
-
-#include "gstsrtbasesink.h"
-
-G_BEGIN_DECLS
-
-#define GST_TYPE_SRT_CLIENT_SINK              (gst_srt_client_sink_get_type ())
-#define GST_IS_SRT_CLIENT_SINK(obj)           (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SRT_CLIENT_SINK))
-#define GST_IS_SRT_CLIENT_SINK_CLASS(klass)   (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SRT_CLIENT_SINK))
-#define GST_SRT_CLIENT_SINK_GET_CLASS(obj)    (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_SRT_CLIENT_SINK, GstSRTClientSinkClass))
-#define GST_SRT_CLIENT_SINK(obj)              (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SRT_CLIENT_SINK, GstSRTClientSink))
-#define GST_SRT_CLIENT_SINK_CLASS(klass)      (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SRT_CLIENT_SINK, GstSRTClientSinkClass))
-#define GST_SRT_CLIENT_SINK_CAST(obj)         ((GstSRTClientSink*)(obj))
-#define GST_SRT_CLIENT_SINK_CLASS_CAST(klass) ((GstSRTClientSinkClass*)(klass))
-
-typedef struct _GstSRTClientSink GstSRTClientSink;
-typedef struct _GstSRTClientSinkClass GstSRTClientSinkClass;
-typedef struct _GstSRTClientSinkPrivate GstSRTClientSinkPrivate;
-
-struct _GstSRTClientSink {
-  GstSRTBaseSink parent;
-
-  /*< private >*/
-  gpointer _gst_reserved[GST_PADDING];
-};
-
-struct _GstSRTClientSinkClass {
-  GstSRTBaseSinkClass parent_class;
-
-  gpointer _gst_reserved[GST_PADDING_LARGE];
-};
-
-GST_EXPORT
-GType gst_srt_client_sink_get_type (void);
-
-G_END_DECLS
-
-#endif /* __GST_SRT_CLIENT_SINK_H__ */
diff --git a/ext/srt/gstsrtclientsrc.c b/ext/srt/gstsrtclientsrc.c
deleted file mode 100644
index ee70fb77f..000000000
--- a/ext/srt/gstsrtclientsrc.c
+++ /dev/null
@@ -1,327 +0,0 @@
-/* GStreamer SRT plugin based on libsrt
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-/**
- * SECTION:element-srtclientsrc
- * @title: srtclientsrc
- *
- * srtclientsrc is a network source that reads <ulink url="http://www.srtalliance.org/">SRT</ulink>
- * packets from the network. Although SRT is a protocol based on UDP, srtclientsrc works like
- * a client socket of connection-oriented protocol.
- *
- * <refsect2>
- * <title>Examples</title>
- * |[
- * gst-launch-1.0 -v srtclientsrc uri="srt://127.0.0.1:7001" ! fakesink
- * ]| This pipeline shows how to connect SRT server by setting #GstSRTClientSrc:uri property.
- *
- * |[
- * gst-launch-1.0 -v srtclientsrc uri="srt://192.168.1.10:7001" rendez-vous ! fakesink
- * ]| This pipeline shows how to connect SRT server by setting #GstSRTClientSrc:uri property and using the rendez-vous mode.
- * </refsect2>
- *
- */
-
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
-#include "gstsrtclientsrc.h"
-#include <srt/srt.h>
-#include <gio/gio.h>
-
-#include "gstsrt.h"
-
-static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
-    GST_PAD_SRC,
-    GST_PAD_ALWAYS,
-    GST_STATIC_CAPS_ANY);
-
-#define GST_CAT_DEFAULT gst_debug_srt_client_src
-GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
-
-struct _GstSRTClientSrcPrivate
-{
-  SRTSOCKET sock;
-  gint poll_id;
-  gint poll_timeout;
-
-  gboolean rendez_vous;
-  gchar *bind_address;
-  guint16 bind_port;
-};
-
-#define GST_SRT_CLIENT_SRC_GET_PRIVATE(obj)  \
-       (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_SRT_CLIENT_SRC, GstSRTClientSrcPrivate))
-
-#define SRT_DEFAULT_POLL_TIMEOUT -1
-enum
-{
-  PROP_POLL_TIMEOUT = 1,
-  PROP_BIND_ADDRESS,
-  PROP_BIND_PORT,
-  PROP_RENDEZ_VOUS,
-
-  /*< private > */
-  PROP_LAST
-};
-
-static GParamSpec *properties[PROP_LAST + 1];
-
-#define gst_srt_client_src_parent_class parent_class
-G_DEFINE_TYPE_WITH_CODE (GstSRTClientSrc, gst_srt_client_src,
-    GST_TYPE_SRT_BASE_SRC, G_ADD_PRIVATE (GstSRTClientSrc)
-    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtclientsrc", 0,
-        "SRT Client Source"));
-
-static void
-gst_srt_client_src_get_property (GObject * object,
-    guint prop_id, GValue * value, GParamSpec * pspec)
-{
-  GstSRTClientSrc *self = GST_SRT_CLIENT_SRC (object);
-  GstSRTClientSrcPrivate *priv = GST_SRT_CLIENT_SRC_GET_PRIVATE (self);
-
-  switch (prop_id) {
-    case PROP_POLL_TIMEOUT:
-      g_value_set_int (value, priv->poll_timeout);
-      break;
-    case PROP_BIND_PORT:
-      g_value_set_int (value, priv->bind_port);
-      break;
-    case PROP_BIND_ADDRESS:
-      g_value_set_string (value, priv->bind_address);
-      break;
-    case PROP_RENDEZ_VOUS:
-      g_value_set_boolean (value, priv->rendez_vous);
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_client_src_set_property (GObject * object,
-    guint prop_id, const GValue * value, GParamSpec * pspec)
-{
-  GstSRTBaseSrc *self = GST_SRT_BASE_SRC (object);
-  GstSRTClientSrcPrivate *priv = GST_SRT_CLIENT_SRC_GET_PRIVATE (self);
-
-  switch (prop_id) {
-    case PROP_POLL_TIMEOUT:
-      priv->poll_timeout = g_value_get_int (value);
-      break;
-    case PROP_BIND_ADDRESS:
-      g_free (priv->bind_address);
-      priv->bind_address = g_value_dup_string (value);
-      break;
-    case PROP_BIND_PORT:
-      priv->bind_port = g_value_get_int (value);
-      break;
-    case PROP_RENDEZ_VOUS:
-      priv->rendez_vous = g_value_get_boolean (value);
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_client_src_finalize (GObject * object)
-{
-  GstSRTClientSrc *self = GST_SRT_CLIENT_SRC (object);
-  GstSRTClientSrcPrivate *priv = GST_SRT_CLIENT_SRC_GET_PRIVATE (self);
-
-  if (priv->poll_id != SRT_ERROR) {
-    srt_epoll_release (priv->poll_id);
-    priv->poll_id = SRT_ERROR;
-  }
-
-  if (priv->sock != SRT_INVALID_SOCK) {
-    srt_close (priv->sock);
-    priv->sock = SRT_INVALID_SOCK;
-  }
-
-  g_free (priv->bind_address);
-
-  G_OBJECT_CLASS (parent_class)->finalize (object);
-}
-
-static GstFlowReturn
-gst_srt_client_src_fill (GstPushSrc * src, GstBuffer * outbuf)
-{
-  GstSRTClientSrc *self = GST_SRT_CLIENT_SRC (src);
-  GstSRTClientSrcPrivate *priv = GST_SRT_CLIENT_SRC_GET_PRIVATE (self);
-  GstFlowReturn ret = GST_FLOW_OK;
-  GstMapInfo info;
-  SRTSOCKET ready[2];
-  gint recv_len;
-
-  if (srt_epoll_wait (priv->poll_id, ready, &(int) {
-          2}, 0, 0, priv->poll_timeout, 0, 0, 0, 0) == -1) {
-
-    /* Assuming that timeout error is normal */
-    if (srt_getlasterror (NULL) != SRT_ETIMEOUT) {
-      GST_ELEMENT_ERROR (src, RESOURCE, READ,
-          (NULL), ("srt_epoll_wait error: %s", srt_getlasterror_str ()));
-      ret = GST_FLOW_ERROR;
-    }
-    srt_clearlasterror ();
-    goto out;
-  }
-
-  if (!gst_buffer_map (outbuf, &info, GST_MAP_WRITE)) {
-    GST_ELEMENT_ERROR (src, RESOURCE, READ,
-        ("Could not map the buffer for writing "), (NULL));
-    ret = GST_FLOW_ERROR;
-    goto out;
-  }
-
-  recv_len = srt_recvmsg (priv->sock, (char *) info.data,
-      gst_buffer_get_size (outbuf));
-
-  gst_buffer_unmap (outbuf, &info);
-
-  if (recv_len == SRT_ERROR) {
-    GST_ELEMENT_ERROR (src, RESOURCE, READ,
-        (NULL), ("srt_recvmsg error: %s", srt_getlasterror_str ()));
-    ret = GST_FLOW_ERROR;
-    goto out;
-  } else if (recv_len == 0) {
-    ret = GST_FLOW_EOS;
-    goto out;
-  }
-
-  gst_buffer_resize (outbuf, 0, recv_len);
-
-  GST_LOG_OBJECT (src, "filled buffer from _get of size %" G_GSIZE_FORMAT,
-      gst_buffer_get_size (outbuf));
-
-out:
-  return ret;
-}
-
-static gboolean
-gst_srt_client_src_start (GstBaseSrc * src)
-{
-  GstSRTClientSrc *self = GST_SRT_CLIENT_SRC (src);
-  GstSRTClientSrcPrivate *priv = GST_SRT_CLIENT_SRC_GET_PRIVATE (self);
-  GstSRTBaseSrc *base = GST_SRT_BASE_SRC (src);
-  GstUri *uri = gst_uri_ref (base->uri);
-  GSocketAddress *socket_address = NULL;
-
-  priv->sock = gst_srt_client_connect (GST_ELEMENT (src), FALSE,
-      gst_uri_get_host (uri), gst_uri_get_port (uri), priv->rendez_vous,
-      priv->bind_address, priv->bind_port, base->latency,
-      &socket_address, &priv->poll_id, base->passphrase, base->key_length);
-
-  g_clear_object (&socket_address);
-  g_clear_pointer (&uri, gst_uri_unref);
-
-  return (priv->sock != SRT_INVALID_SOCK);
-}
-
-static gboolean
-gst_srt_client_src_stop (GstBaseSrc * src)
-{
-  GstSRTClientSrc *self = GST_SRT_CLIENT_SRC (src);
-  GstSRTClientSrcPrivate *priv = GST_SRT_CLIENT_SRC_GET_PRIVATE (self);
-
-  if (priv->poll_id != SRT_ERROR) {
-    if (priv->sock != SRT_INVALID_SOCK)
-      srt_epoll_remove_usock (priv->poll_id, priv->sock);
-    srt_epoll_release (priv->poll_id);
-  }
-  priv->poll_id = SRT_ERROR;
-
-  GST_DEBUG_OBJECT (self, "closing SRT connection");
-  if (priv->sock != SRT_INVALID_SOCK)
-    srt_close (priv->sock);
-  priv->sock = SRT_INVALID_SOCK;
-
-  return TRUE;
-}
-
-static void
-gst_srt_client_src_class_init (GstSRTClientSrcClass * klass)
-{
-  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
-  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
-  GstBaseSrcClass *gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
-  GstPushSrcClass *gstpushsrc_class = GST_PUSH_SRC_CLASS (klass);
-
-  gobject_class->set_property = gst_srt_client_src_set_property;
-  gobject_class->get_property = gst_srt_client_src_get_property;
-  gobject_class->finalize = gst_srt_client_src_finalize;
-
-  /**
-   * GstSRTClientSrc:poll-timeout:
-   *
-   * The timeout(ms) value when polling SRT socket.
-   */
-  properties[PROP_POLL_TIMEOUT] =
-      g_param_spec_int ("poll-timeout", "Poll timeout",
-      "Return poll wait after timeout miliseconds (-1 = infinite)", -1,
-      G_MAXINT32, SRT_DEFAULT_POLL_TIMEOUT,
-      G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_BIND_ADDRESS] =
-      g_param_spec_string ("bind-address", "Bind Address",
-      "Address to bind socket to (required for rendez-vous mode) ", NULL,
-      G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_BIND_PORT] =
-      g_param_spec_int ("bind-port", "Bind Port",
-      "Port to bind socket to (Ignored in rendez-vous mode)", 0,
-      G_MAXUINT16, 0,
-      G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_RENDEZ_VOUS] =
-      g_param_spec_boolean ("rendez-vous", "Rendez Vous",
-      "Work in Rendez-Vous mode instead of client/caller mode", FALSE,
-      G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS);
-
-  g_object_class_install_properties (gobject_class, PROP_LAST, properties);
-
-  gst_element_class_add_static_pad_template (gstelement_class, &src_template);
-  gst_element_class_set_metadata (gstelement_class,
-      "SRT client source", "Source/Network",
-      "Receive data over the network via SRT",
-      "Justin Kim <justin.kim@collabora.com>");
-
-  gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_srt_client_src_start);
-  gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_srt_client_src_stop);
-
-  gstpushsrc_class->fill = GST_DEBUG_FUNCPTR (gst_srt_client_src_fill);
-}
-
-static void
-gst_srt_client_src_init (GstSRTClientSrc * self)
-{
-  GstSRTClientSrcPrivate *priv = GST_SRT_CLIENT_SRC_GET_PRIVATE (self);
-
-  priv->sock = SRT_INVALID_SOCK;
-  priv->poll_id = SRT_ERROR;
-  priv->poll_timeout = SRT_DEFAULT_POLL_TIMEOUT;
-  priv->rendez_vous = FALSE;
-  priv->bind_address = NULL;
-  priv->bind_port = 0;
-}
diff --git a/ext/srt/gstsrtclientsrc.h b/ext/srt/gstsrtclientsrc.h
deleted file mode 100644
index 99c5a0e8f..000000000
--- a/ext/srt/gstsrtclientsrc.h
+++ /dev/null
@@ -1,59 +0,0 @@
-/* GStreamer
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifndef __GST_SRT_CLIENT_SRC_H__
-#define __GST_SRT_CLIENT_SRC_H__
-
-#include "gstsrtbasesrc.h"
-
-G_BEGIN_DECLS
-
-#define GST_TYPE_SRT_CLIENT_SRC              (gst_srt_client_src_get_type ())
-#define GST_IS_SRT_CLIENT_SRC(obj)           (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SRT_CLIENT_SRC))
-#define GST_IS_SRT_CLIENT_SRC_CLASS(klass)   (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SRT_CLIENT_SRC))
-#define GST_SRT_CLIENT_SRC_GET_CLASS(obj)    (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_SRT_CLIENT_SRC, GstSRTClientSrcClass))
-#define GST_SRT_CLIENT_SRC(obj)              (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SRT_CLIENT_SRC, GstSRTClientSrc))
-#define GST_SRT_CLIENT_SRC_CLASS(klass)      (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SRT_CLIENT_SRC, GstSRTClientSrcClass))
-#define GST_SRT_CLIENT_SRC_CAST(obj)         ((GstSRTClientSrc*)(obj))
-#define GST_SRT_CLIENT_SRC_CLASS_CAST(klass) ((GstSRTClientSrcClass*)(klass))
-
-typedef struct _GstSRTClientSrc GstSRTClientSrc;
-typedef struct _GstSRTClientSrcClass GstSRTClientSrcClass;
-typedef struct _GstSRTClientSrcPrivate GstSRTClientSrcPrivate;
-
-struct _GstSRTClientSrc {
-  GstSRTBaseSrc parent;
-
-  /*< private >*/
-  gpointer _gst_reserved[GST_PADDING];
-};
-
-struct _GstSRTClientSrcClass {
-  GstSRTBaseSrcClass parent_class;
-
-  gpointer _gst_reserved[GST_PADDING_LARGE];
-};
-
-GST_EXPORT
-GType gst_srt_client_src_get_type (void);
-
-G_END_DECLS
-
-#endif /* __GST_SRT_CLIENT_SRC_H__ */
diff --git a/ext/srt/gstsrtobject.c b/ext/srt/gstsrtobject.c
new file mode 100644
index 000000000..0716a65db
--- /dev/null
+++ b/ext/srt/gstsrtobject.c
@@ -0,0 +1,1441 @@
+/* GStreamer
+ * Copyright (C) 2018, Collabora Ltd.
+ * Copyright (C) 2018, SK Telecom, Co., Ltd.
+ *   Author: Jeongseok Kim <jeongseok.kim@sk.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include "gstsrtobject.h"
+
+#include <gio/gnetworking.h>
+#include <stdlib.h>
+
+GST_DEBUG_CATEGORY (gst_debug_srtobject);
+#define GST_CAT_DEFAULT gst_debug_srtobject
+
+enum
+{
+  PROP_URI = 1,
+  PROP_MODE,
+  PROP_LOCALADDRESS,
+  PROP_LOCALPORT,
+  PROP_PASSPHRASE,
+  PROP_PBKEYLEN,
+  PROP_POLL_TIMEOUT,
+  PROP_LATENCY,
+  PROP_MSG_SIZE,
+  PROP_STATS,
+  PROP_LAST
+};
+
+typedef struct
+{
+  SRTSOCKET sock;
+  gint poll_id;
+  GSocketAddress *sockaddr;
+  gboolean sent_headers;
+} SRTCaller;
+
+static SRTCaller *
+srt_caller_new (void)
+{
+  SRTCaller *caller = g_new0 (SRTCaller, 1);
+  caller->sock = SRT_INVALID_SOCK;
+  caller->poll_id = SRT_ERROR;
+  caller->sent_headers = FALSE;
+
+  return caller;
+}
+
+static void
+srt_caller_free (SRTCaller * caller)
+{
+  g_return_if_fail (caller != NULL);
+
+  g_clear_object (&caller->sockaddr);
+
+  if (caller->sock != SRT_INVALID_SOCK) {
+    srt_close (caller->sock);
+  }
+
+  if (caller->poll_id != SRT_ERROR) {
+    srt_epoll_release (caller->poll_id);
+  }
+
+  g_free (caller);
+}
+
+static void
+srt_caller_invoke_removed_closure (SRTCaller * caller, GstSRTObject * srtobject)
+{
+  GValue values[2] = { G_VALUE_INIT };
+
+  if (srtobject->caller_removed_closure == NULL) {
+    return;
+  }
+
+  g_value_init (&values[0], G_TYPE_INT);
+  g_value_set_int (&values[0], caller->sock);
+
+  g_value_init (&values[1], G_TYPE_SOCKET_ADDRESS);
+  g_value_set_pointer (&values[1], caller->sockaddr);
+
+  g_closure_invoke (srtobject->caller_removed_closure, NULL, 2, values, NULL);
+
+  g_value_unset (&values[0]);
+  g_value_unset (&values[1]);
+}
+
+struct srt_constant_params
+{
+  const gchar *name;
+  gint param;
+  gint val;
+};
+
+static struct srt_constant_params srt_params[] = {
+  {"SRTO_SNDSYN", SRTO_SNDSYN, 0},      /* 0: non-blocking */
+  {"SRTO_RCVSYN", SRTO_RCVSYN, 0},      /* 0: non-blocking */
+  {"SRTO_LINGER", SRTO_LINGER, 0},
+  {"SRTO_TSBPMODE", SRTO_TSBPDMODE, 1}, /* Timestamp-based Packet Delivery mode must be enabled */
+  {NULL, -1, -1},
+};
+
+static gint srt_init_refcount = 0;
+
+static gboolean
+gst_srt_object_set_default_params (SRTSOCKET sock, GError ** error)
+{
+  struct srt_constant_params *params = srt_params;
+
+  for (; params->name != NULL; params++) {
+    if (srt_setsockopt (sock, 0, params->param, &params->val, sizeof (gint))) {
+      g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_SETTINGS,
+          "failed to set %s (reason: %s)", params->name,
+          srt_getlasterror_str ());
+      return FALSE;
+    }
+  }
+
+  return TRUE;
+}
+
+GstSRTObject *
+gst_srt_object_new (GstElement * element)
+{
+  GstSRTObject *srtobject;
+
+  if (g_atomic_int_get (&srt_init_refcount) == 0) {
+    GST_DEBUG_OBJECT (element, "Starting up SRT");
+    if (srt_startup () != 0) {
+      g_warning ("Failed to initialize SRT (reason: %s)",
+          srt_getlasterror_str ());
+    }
+  }
+
+  g_atomic_int_inc (&srt_init_refcount);
+
+  srtobject = g_new0 (GstSRTObject, 1);
+  srtobject->element = element;
+  srtobject->parameters = gst_structure_new ("application/x-srt-params",
+      "poll-timeout", G_TYPE_INT, GST_SRT_DEFAULT_POLL_TIMEOUT,
+      "latency", G_TYPE_INT, GST_SRT_DEFAULT_LATENCY,
+      "mode", GST_TYPE_SRT_CONNECTION_MODE, GST_SRT_DEFAULT_MODE, NULL);
+
+  srtobject->sock = SRT_INVALID_SOCK;
+  srtobject->poll_id = srt_epoll_create ();
+  srtobject->listener_sock = SRT_INVALID_SOCK;
+  srtobject->listener_poll_id = SRT_ERROR;
+  srtobject->sent_headers = FALSE;
+
+  g_mutex_init (&srtobject->sock_lock);
+  g_cond_init (&srtobject->sock_cond);
+  return srtobject;
+}
+
+void
+gst_srt_object_destroy (GstSRTObject * srtobject)
+{
+  g_return_if_fail (srtobject != NULL);
+
+  if (srtobject->poll_id != SRT_ERROR) {
+    srt_epoll_release (srtobject->poll_id);
+    srtobject->poll_id = SRT_ERROR;
+  }
+
+  g_mutex_clear (&srtobject->sock_lock);
+  g_cond_clear (&srtobject->sock_cond);
+
+  GST_DEBUG_OBJECT (srtobject->element, "Destroying srtobject");
+  gst_structure_free (srtobject->parameters);
+
+  if (g_atomic_int_dec_and_test (&srt_init_refcount)) {
+    srt_cleanup ();
+    GST_DEBUG_OBJECT (srtobject->element, "Cleaning up SRT");
+  }
+
+  g_free (srtobject);
+}
+
+gboolean
+gst_srt_object_set_property_helper (GstSRTObject * srtobject,
+    guint prop_id, const GValue * value, GParamSpec * pspec)
+{
+  switch (prop_id) {
+    case PROP_URI:{
+      gchar *uri = g_value_dup_string (value);
+      gst_srt_object_set_uri (srtobject, uri, NULL);
+      g_free (uri);
+      break;
+    }
+    case PROP_MODE:
+      gst_structure_set_value (srtobject->parameters, "mode", value);
+      break;
+    case PROP_POLL_TIMEOUT:
+      gst_structure_set_value (srtobject->parameters, "poll-timeout", value);
+      break;
+    case PROP_LATENCY:
+      gst_structure_set_value (srtobject->parameters, "latency", value);
+      break;
+    case PROP_MSG_SIZE:
+      gst_structure_set_value (srtobject->parameters, "msg-size", value);
+      break;
+    case PROP_LOCALADDRESS:
+      gst_structure_set_value (srtobject->parameters, "localaddress", value);
+      break;
+    case PROP_LOCALPORT:
+      gst_structure_set_value (srtobject->parameters, "localport", value);
+      break;
+    case PROP_PASSPHRASE:
+      gst_structure_set_value (srtobject->parameters, "passphrase", value);
+      break;
+    case PROP_PBKEYLEN:
+      gst_structure_set_value (srtobject->parameters, "pbkeylen", value);
+      break;
+    default:
+      return FALSE;
+  }
+  return TRUE;
+}
+
+gboolean
+gst_srt_object_get_property_helper (GstSRTObject * srtobject,
+    guint prop_id, GValue * value, GParamSpec * pspec)
+{
+  switch (prop_id) {
+    case PROP_URI:
+      g_value_set_string (value, gst_uri_to_string (srtobject->uri));
+      break;
+    case PROP_MODE:{
+      GstSRTConnectionMode v;
+      if (!gst_structure_get_enum (srtobject->parameters, "mode",
+              GST_TYPE_SRT_CONNECTION_MODE, (gint *) & v)) {
+        GST_WARNING_OBJECT (srtobject->element, "Failed to get 'mode'");
+        v = GST_SRT_CONNECTION_MODE_NONE;
+      }
+      g_value_set_enum (value, v);
+      break;
+    }
+    case PROP_LOCALADDRESS:
+      g_value_set_string (value,
+          gst_structure_get_string (srtobject->parameters, "localaddress"));
+      break;
+    case PROP_LOCALPORT:{
+      guint v;
+      if (!gst_structure_get_uint (srtobject->parameters, "localport", &v)) {
+        GST_WARNING_OBJECT (srtobject->element, "Failed to get 'localport'");
+        v = GST_SRT_DEFAULT_PORT;
+      }
+      g_value_set_uint (value, v);
+      break;
+    }
+    case PROP_PBKEYLEN:{
+      GstSRTKeyLengthBits v;
+      if (!gst_structure_get_enum (srtobject->parameters, "pbkeylen",
+              GST_TYPE_SRT_KEY_LENGTH_BITS, (gint *) & v)) {
+        GST_WARNING_OBJECT (srtobject->element, "Failed to get 'pbkeylen'");
+        v = GST_SRT_KEY_LENGTH_BITS_NO_KEY;
+      }
+      g_value_set_enum (value, v);
+      break;
+    }
+    case PROP_POLL_TIMEOUT:{
+      gint v;
+      if (!gst_structure_get_int (srtobject->parameters, "poll-timeout", &v)) {
+        GST_WARNING_OBJECT (srtobject->element, "Failed to get 'poll-timeout'");
+        v = GST_SRT_DEFAULT_POLL_TIMEOUT;
+      }
+      g_value_set_int (value, v);
+      break;
+    }
+    case PROP_LATENCY:{
+      gint v;
+      if (!gst_structure_get_int (srtobject->parameters, "latency", &v)) {
+        GST_WARNING_OBJECT (srtobject->element, "Failed to get 'latency'");
+        v = GST_SRT_DEFAULT_LATENCY;
+      }
+      g_value_set_int (value, v);
+      break;
+    }
+    case PROP_MSG_SIZE:{
+      gint v;
+      if (!gst_structure_get_int (srtobject->parameters, "msg-size", &v)) {
+        GST_WARNING_OBJECT (srtobject->element, "Failed to get 'msg-size'");
+        v = GST_SRT_DEFAULT_MSG_SIZE;
+      }
+      g_value_set_int (value, v);
+      break;
+    }
+    case PROP_STATS:
+      g_value_take_boxed (value, gst_srt_object_get_stats (srtobject));
+      break;
+    default:
+      return FALSE;
+  }
+
+  return TRUE;
+}
+
+void
+gst_srt_object_install_properties_helper (GObjectClass * gobject_class)
+{
+  /**
+   * GstSRTSrc:uri:
+   *
+   * The URI used by SRT connection. User can specify SRT specific options by URI parameters.
+   * Refer to <a href="https://github.com/Haivision/srt/blob/master/docs/stransmit.md#medium-srt">Mediun: SRT</a>
+   */
+  g_object_class_install_property (gobject_class, PROP_URI,
+      g_param_spec_string ("uri", "URI",
+          "URI in the form of srt://address:port", GST_SRT_DEFAULT_URI,
+          G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY |
+          G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:mode:
+   * 
+   * The SRT connection mode. 
+   * This property can be set by URI parameters.
+   */
+  g_object_class_install_property (gobject_class, PROP_MODE,
+      g_param_spec_enum ("mode", "Connection mode",
+          "SRT connection mode", GST_TYPE_SRT_CONNECTION_MODE,
+          GST_SRT_CONNECTION_MODE_CALLER,
+          G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY |
+          G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:localaddress:
+   * 
+   * The address to bind when #GstSRTSrc:mode is listener or rendezvous.
+   * This property can be set by URI parameters.
+   */
+  g_object_class_install_property (gobject_class, PROP_LOCALADDRESS,
+      g_param_spec_string ("localaddress", "Local address",
+          "Local address to bind", GST_SRT_DEFAULT_LOCALADDRESS,
+          G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY |
+          G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:localport:
+   *
+   * The local port to bind when #GstSRTSrc:mode is listener or rendezvous.
+   * This property can be set by URI parameters.
+   */
+  g_object_class_install_property (gobject_class, PROP_POLL_TIMEOUT,
+      g_param_spec_uint ("localport", "Local port",
+          "Local port to bind", 0,
+          65535, GST_SRT_DEFAULT_PORT,
+          G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY |
+          G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:passphrase:
+   *
+   * The password for the encrypted transmission.
+   * This property can be set by URI parameters.
+   */
+  g_object_class_install_property (gobject_class, PROP_LOCALADDRESS,
+      g_param_spec_string ("passphrase", "Passphrase",
+          "Password for the encrypted transmission", "",
+          G_PARAM_WRITABLE | GST_PARAM_MUTABLE_READY | G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:pbkeylen:
+   * 
+   * The crypto key length.
+   * This property can be set by URI parameters.
+   */
+  g_object_class_install_property (gobject_class, PROP_PBKEYLEN,
+      g_param_spec_enum ("pbkeylen", "Crypto key length",
+          "Crypto key length in bits", GST_TYPE_SRT_KEY_LENGTH_BITS,
+          GST_SRT_DEFAULT_PBKEYLEN,
+          G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY |
+          G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:poll-timeout:
+   * 
+   * The polling timeout used when srt poll is started.
+   * Even if the default value indicates infinite waiting, it can be cancellable according to #GstState
+   * This property can be set by URI parameters.
+   */
+  g_object_class_install_property (gobject_class, PROP_POLL_TIMEOUT,
+      g_param_spec_int ("poll-timeout", "Poll timeout",
+          "Return poll wait after timeout miliseconds (-1 = infinite)", -1,
+          G_MAXINT32, GST_SRT_DEFAULT_POLL_TIMEOUT,
+          G_PARAM_READWRITE | GST_PARAM_MUTABLE_READY |
+          G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:latency:
+   *
+   * The maximum accepted transmission latency.
+   */
+  g_object_class_install_property (gobject_class, PROP_LATENCY,
+      g_param_spec_int ("latency", "latency",
+          "Minimum latency (milliseconds)", 0,
+          G_MAXINT32, GST_SRT_DEFAULT_LATENCY,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:msg-size:
+   * 
+   * The message size of buffer.
+   */
+  g_object_class_install_property (gobject_class, PROP_MSG_SIZE,
+      g_param_spec_int ("msg-size", "message size",
+          "Message size to use with SRT", 1,
+          G_MAXINT32, GST_SRT_DEFAULT_MSG_SIZE,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
+  /**
+   * GstSRTSrc:stats:
+   *
+   * The statistics from SRT.
+   */
+  g_object_class_install_property (gobject_class, PROP_STATS,
+      g_param_spec_boxed ("stats", "Statistics",
+          "SRT Statistics", GST_TYPE_STRUCTURE,
+          G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
+
+}
+
+static void
+gst_srt_object_set_enum_value (GstStructure * s, GType enum_type, gpointer key,
+    gpointer value)
+{
+  GEnumClass *enum_class;
+  GEnumValue *enum_value;
+
+  enum_class = g_type_class_ref (enum_type);
+  enum_value = g_enum_get_value_by_nick (enum_class, value);
+
+  if (enum_value) {
+    GValue v = G_VALUE_INIT;
+    g_value_init (&v, enum_type);
+    g_value_set_enum (&v, enum_value->value);
+    gst_structure_set_value (s, key, &v);
+  }
+
+  g_type_class_unref (enum_class);
+}
+
+static void
+gst_srt_object_set_string_value (GstStructure * s, const gchar * key,
+    const gchar * value)
+{
+  GValue v = G_VALUE_INIT;
+  g_value_init (&v, G_TYPE_STRING);
+  g_value_set_static_string (&v, value);
+  gst_structure_set_value (s, key, &v);
+  g_value_unset (&v);
+}
+
+static void
+gst_srt_object_set_uint_value (GstStructure * s, const gchar * key,
+    const gchar * value)
+{
+  GValue v = G_VALUE_INIT;
+  g_value_init (&v, G_TYPE_UINT);
+  g_value_set_uint (&v, (guint) strtoul (value, NULL, 10));
+  gst_structure_set_value (s, key, &v);
+  g_value_unset (&v);
+}
+
+static void
+gst_srt_object_validate_parameters (GstStructure * s, GstUri * uri)
+{
+  GstSRTConnectionMode connection_mode = GST_SRT_CONNECTION_MODE_NONE;
+
+  gst_structure_get_enum (s, "mode", GST_TYPE_SRT_CONNECTION_MODE,
+      (gint *) & connection_mode);
+
+  if (connection_mode == GST_SRT_CONNECTION_MODE_RENDEZVOUS ||
+      connection_mode == GST_SRT_CONNECTION_MODE_LISTENER) {
+    guint local_port;
+    const gchar *local_address = gst_structure_get_string (s, "localaddress");
+
+
+    if (local_address == NULL) {
+      gst_srt_object_set_string_value (s, "localaddress",
+          GST_SRT_DEFAULT_LOCALADDRESS);
+    }
+
+    if (!gst_structure_get_uint (s, "localport", &local_port)) {
+      gst_srt_object_set_uint_value (s, "localport",
+          G_STRINGIFY (GST_SRT_DEFAULT_PORT));
+    }
+  }
+}
+
+gboolean
+gst_srt_object_set_uri (GstSRTObject * srtobject, const gchar * uri,
+    GError ** err)
+{
+  GHashTable *query_table = NULL;
+  GHashTableIter iter;
+  gpointer key, value;
+
+  if (srtobject->opened) {
+    g_warning
+        ("It's not supported to change the 'uri' property when SRT socket is opened.");
+    g_set_error (err, GST_URI_ERROR, GST_URI_ERROR_BAD_STATE,
+        "It's not supported to change the 'uri' property when SRT socket is opened");
+
+    return FALSE;
+  }
+
+  if (!g_str_has_prefix (uri, GST_SRT_DEFAULT_URI_SCHEME)) {
+    g_warning ("Given uri cannot be used for SRT connection.");
+    g_set_error (err, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
+        "Invalid SRT URI scheme");
+    return FALSE;
+  }
+
+  g_clear_pointer (&srtobject->uri, gst_uri_unref);
+  srtobject->uri = gst_uri_from_string (uri);
+
+  query_table = gst_uri_get_query_table (srtobject->uri);
+
+  GST_DEBUG_OBJECT (srtobject->element,
+      "set uri to (host: %s, port: %d) with %d query strings",
+      gst_uri_get_host (srtobject->uri), gst_uri_get_port (srtobject->uri),
+      query_table == NULL ? 0 : g_hash_table_size (query_table));
+
+  if (!query_table) {
+    GST_DEBUG_OBJECT (srtobject->element, "No parameters from uri");
+    return TRUE;
+  }
+
+  g_hash_table_iter_init (&iter, query_table);
+  while (g_hash_table_iter_next (&iter, &key, &value)) {
+    if (!g_strcmp0 ("mode", key)) {
+      gst_srt_object_set_enum_value (srtobject->parameters,
+          GST_TYPE_SRT_CONNECTION_MODE, key, value);
+    } else if (!g_strcmp0 ("localaddress", key)) {
+      gst_srt_object_set_string_value (srtobject->parameters, key, value);
+    } else if (!g_strcmp0 ("localport", key)) {
+      gst_srt_object_set_uint_value (srtobject->parameters, key, value);
+    } else if (!g_strcmp0 ("passphrase", key)) {
+      gst_srt_object_set_string_value (srtobject->parameters, key, value);
+    } else if (!g_strcmp0 ("pbkeylen", key)) {
+      gst_srt_object_set_enum_value (srtobject->parameters,
+          GST_TYPE_SRT_KEY_LENGTH_BITS, key, value);
+    }
+  }
+
+  g_hash_table_unref (query_table);
+
+  gst_srt_object_validate_parameters (srtobject->parameters, srtobject->uri);
+
+  return TRUE;
+}
+
+static gpointer
+thread_func (gpointer data)
+{
+  GstSRTObject *srtobject = data;
+
+  g_main_loop_run (srtobject->loop);
+
+  return NULL;
+}
+
+static gboolean
+idle_listen_source_cb (gpointer data)
+{
+  GstSRTObject *srtobject = data;
+  SRTSOCKET caller_sock;
+  struct sockaddr caller_sa;
+  gsize caller_sa_len;
+
+  gint poll_timeout;
+
+  SRTSOCKET rsock;
+  gint rsocklen = 1;
+
+  if (!gst_structure_get_int (srtobject->parameters, "poll-timeout",
+          &poll_timeout)) {
+    poll_timeout = GST_SRT_DEFAULT_POLL_TIMEOUT;
+  }
+
+  GST_DEBUG_OBJECT (srtobject->element, "Waiting a request from caller");
+
+  if (srt_epoll_wait (srtobject->listener_poll_id, &rsock,
+          &rsocklen, 0, 0, poll_timeout, NULL, 0, NULL, 0) < 0) {
+    gint srt_errno = srt_getlasterror (NULL);
+
+    if (srt_errno == SRT_ETIMEOUT) {
+      return TRUE;
+    } else {
+      GST_ELEMENT_ERROR (srtobject->element, RESOURCE, FAILED,
+          ("abort polling: %s", srt_getlasterror_str ()), (NULL));
+      return FALSE;
+    }
+  }
+
+  caller_sock =
+      srt_accept (srtobject->listener_sock, &caller_sa, (int *) &caller_sa_len);
+
+  if (caller_sock != SRT_INVALID_SOCK) {
+    SRTCaller *caller;
+    gint flag = SRT_EPOLL_ERR;
+
+    caller = srt_caller_new ();
+    caller->sockaddr =
+        g_socket_address_new_from_native (&caller_sa, caller_sa_len);
+    caller->poll_id = srt_epoll_create ();
+    caller->sock = caller_sock;
+
+    if (gst_uri_handler_get_uri_type (GST_URI_HANDLER
+            (srtobject->element)) == GST_URI_SRC) {
+      flag |= SRT_EPOLL_IN;
+    } else {
+      flag |= SRT_EPOLL_OUT;
+    }
+
+    if (srt_epoll_add_usock (caller->poll_id, caller_sock, &flag)) {
+
+      GST_ELEMENT_ERROR (srtobject->element, RESOURCE, SETTINGS,
+          ("%s", srt_getlasterror_str ()), (NULL));
+
+      srt_caller_free (caller);
+
+      /* try-again */
+      return TRUE;
+    }
+
+    GST_OBJECT_LOCK (srtobject->element);
+    srtobject->callers = g_list_append (srtobject->callers, caller);
+    GST_OBJECT_UNLOCK (srtobject->element);
+
+    g_mutex_lock (&srtobject->sock_lock);
+    g_cond_signal (&srtobject->sock_cond);
+    g_mutex_unlock (&srtobject->sock_lock);
+
+    /* notifying caller-added */
+    if (srtobject->caller_added_closure != NULL) {
+      GValue values[2] = { G_VALUE_INIT };
+
+      g_value_init (&values[0], G_TYPE_INT);
+      g_value_set_int (&values[0], caller->sock);
+
+      g_value_init (&values[1], G_TYPE_SOCKET_ADDRESS);
+      g_value_set_pointer (&values[1], caller->sockaddr);
+
+      g_closure_invoke (srtobject->caller_added_closure, NULL, 2, values, NULL);
+    }
+
+    GST_DEBUG_OBJECT (srtobject->element, "Accept to connect");
+  }
+
+  /* only one caller is allowed if the element is source. */
+  return gst_uri_handler_get_uri_type (GST_URI_HANDLER (srtobject->element)) !=
+      GST_URI_SRC;
+}
+
+static gboolean
+gst_srt_object_wait_connect (GstSRTObject * srtobject,
+    GCancellable * cancellable, gpointer sa, size_t sa_len, GError ** error)
+{
+  SRTSOCKET sock = SRT_INVALID_SOCK;
+  const gchar *local_address = NULL;
+  guint local_port = 0;
+  gint sock_flags = SRT_EPOLL_ERR | SRT_EPOLL_IN;
+
+  gpointer bind_sa;
+  gsize bind_sa_len;
+  GSocketAddress *bind_addr;
+
+  gst_structure_get_uint (srtobject->parameters, "localport", &local_port);
+
+  local_address =
+      gst_structure_get_string (srtobject->parameters, "localaddress");
+
+  bind_addr = g_inet_socket_address_new_from_string (local_address, local_port);
+  bind_sa_len = g_socket_address_get_native_size (bind_addr);
+  bind_sa = g_alloca (bind_sa_len);
+
+  if (!g_socket_address_to_native (bind_addr, bind_sa, bind_sa_len, error)) {
+    goto failed;
+  }
+
+  g_clear_object (&bind_addr);
+
+  sock = srt_socket (AF_INET, SOCK_DGRAM, 0);
+  if (sock == SRT_INVALID_SOCK) {
+    g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_INIT, "%s",
+        srt_getlasterror_str ());
+    goto failed;
+  }
+
+  if (!gst_srt_object_set_default_params (sock, error)) {
+    goto failed;
+  }
+
+  GST_DEBUG_OBJECT (srtobject->element, "Binding to %s (port: %d)",
+      local_address, local_port);
+
+  if (srt_bind (sock, bind_sa, bind_sa_len) == SRT_ERROR) {
+    g_set_error (error, GST_RESOURCE_ERROR,
+        GST_RESOURCE_ERROR_OPEN_READ_WRITE, "Cannot bind to %s:%d - %s",
+        local_address, local_port, srt_getlasterror_str ());
+    goto failed;
+  }
+
+  if (srt_epoll_add_usock (srtobject->listener_poll_id, sock, &sock_flags)) {
+    g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_SETTINGS, "%s",
+        srt_getlasterror_str ());
+    goto failed;
+  }
+
+  GST_DEBUG_OBJECT (srtobject->element, "Starting to listen on bind socket");
+  if (srt_listen (sock, 1) == SRT_ERROR) {
+    g_set_error (error, GST_RESOURCE_ERROR,
+        GST_RESOURCE_ERROR_OPEN_READ_WRITE, "Cannot listen on bind socket: %s",
+        srt_getlasterror_str ());
+
+    goto failed;
+  }
+
+  srtobject->listener_sock = sock;
+
+  srtobject->context = g_main_context_new ();
+  srtobject->loop = g_main_loop_new (srtobject->context, TRUE);
+
+  srtobject->listener_source = g_idle_source_new ();
+  g_source_set_callback (srtobject->listener_source,
+      (GSourceFunc) idle_listen_source_cb, srtobject, NULL);
+
+  g_source_attach (srtobject->listener_source, srtobject->context);
+
+  srtobject->thread =
+      g_thread_try_new ("GstSRTObjectListener", thread_func, srtobject, error);
+
+  if (*error != NULL) {
+    goto failed;
+  }
+
+  return TRUE;
+
+failed:
+
+  g_clear_pointer (&srtobject->loop, g_main_loop_unref);
+  g_clear_pointer (&srtobject->context, g_main_context_unref);
+
+  if (srtobject->listener_poll_id != SRT_ERROR) {
+    srt_epoll_release (srtobject->listener_poll_id);
+  }
+
+  if (sock != SRT_INVALID_SOCK) {
+    srt_close (sock);
+  }
+
+  g_clear_object (&bind_addr);
+
+  srtobject->listener_poll_id = SRT_ERROR;
+  srtobject->listener_sock = SRT_INVALID_SOCK;
+
+  return FALSE;
+}
+
+static gboolean
+gst_srt_object_connect (GstSRTObject * srtobject,
+    GstSRTConnectionMode connection_mode, gpointer sa, size_t sa_len,
+    GError ** error)
+{
+  SRTSOCKET sock;
+  gint option_val = -1;
+  gint sock_flags = SRT_EPOLL_ERR;
+  guint local_port = 0;
+
+  sock = srt_socket (AF_INET, SOCK_DGRAM, 0);
+  if (sock == SRT_INVALID_SOCK) {
+    g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_INIT, "%s",
+        srt_getlasterror_str ());
+    goto failed;
+  }
+
+  if (!gst_srt_object_set_default_params (sock, error)) {
+    goto failed;
+  }
+
+  switch (gst_uri_handler_get_uri_type (GST_URI_HANDLER (srtobject->element))) {
+    case GST_URI_SRC:
+      option_val = 0;
+      sock_flags |= SRT_EPOLL_IN;
+      break;
+    case GST_URI_SINK:
+      option_val = 1;
+      sock_flags |= SRT_EPOLL_OUT;
+      break;
+    default:
+      g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_SETTINGS,
+          "Cannot determine stream direction");
+      goto failed;
+  }
+
+  if (srt_setsockopt (sock, 0, SRTO_SENDER, &option_val, sizeof (gint))) {
+    g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_SETTINGS, "%s",
+        srt_getlasterror_str ());
+    goto failed;
+  }
+
+  option_val = (connection_mode == GST_SRT_CONNECTION_MODE_RENDEZVOUS);
+  if (srt_setsockopt (sock, 0, SRTO_RENDEZVOUS, &option_val, sizeof (gint))) {
+    g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_SETTINGS, "%s",
+        srt_getlasterror_str ());
+    goto failed;
+  }
+
+  gst_structure_get_uint (srtobject->parameters, "localport", &local_port);
+
+  /* According to SRT norm, bind local address and port if specified */
+  if (local_port != 0) {
+    gpointer bind_sa;
+    gsize bind_sa_len;
+    const gchar *local_address =
+        gst_structure_get_string (srtobject->parameters, "localaddress");
+    GSocketAddress *bind_addr =
+        g_inet_socket_address_new_from_string (local_address,
+        local_port);
+
+    bind_sa_len = g_socket_address_get_native_size (bind_addr);
+    bind_sa = g_alloca (bind_sa_len);
+
+    if (!g_socket_address_to_native (bind_addr, bind_sa, bind_sa_len, error)) {
+      g_clear_object (&bind_addr);
+      goto failed;
+    }
+
+    g_clear_object (&bind_addr);
+
+    GST_DEBUG_OBJECT (srtobject->element, "Binding to %s (port: %d)",
+        local_address, local_port);
+
+    if (srt_bind (sock, bind_sa, bind_sa_len) == SRT_ERROR) {
+      g_set_error (error, GST_RESOURCE_ERROR,
+          GST_RESOURCE_ERROR_OPEN_READ_WRITE, "Cannot bind to %s:%d - %s",
+          local_address, local_port, srt_getlasterror_str ());
+      goto failed;
+    }
+  }
+
+  if (srt_epoll_add_usock (srtobject->poll_id, sock, &sock_flags)) {
+    g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_SETTINGS, "%s",
+        srt_getlasterror_str ());
+    goto failed;
+  }
+
+  if (srt_connect (sock, sa, sa_len) == SRT_ERROR) {
+    g_set_error (error, GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_OPEN_READ, "%s",
+        srt_getlasterror_str ());
+    goto failed;
+  }
+
+  srtobject->sock = sock;
+
+  return TRUE;
+
+failed:
+
+  if (srtobject->poll_id != SRT_ERROR) {
+    srt_epoll_release (srtobject->poll_id);
+  }
+
+  if (sock != SRT_INVALID_SOCK) {
+    srt_close (sock);
+  }
+
+  srtobject->poll_id = SRT_ERROR;
+  srtobject->sock = SRT_INVALID_SOCK;
+
+  return FALSE;
+}
+
+static gboolean
+gst_srt_object_open_connection (GstSRTObject * srtobject,
+    GCancellable * cancellable, GstSRTConnectionMode connection_mode,
+    gpointer sa, size_t sa_len, GError ** error)
+{
+  gboolean ret = FALSE;
+
+  if (connection_mode == GST_SRT_CONNECTION_MODE_LISTENER) {
+    ret =
+        gst_srt_object_wait_connect (srtobject, cancellable, sa, sa_len, error);
+  } else {
+    ret =
+        gst_srt_object_connect (srtobject, connection_mode, sa, sa_len, error);
+  }
+
+  return ret;
+}
+
+gboolean
+gst_srt_object_open (GstSRTObject * srtobject, GCancellable * cancellable,
+    GError ** error)
+{
+  return gst_srt_object_open_full (srtobject, NULL, NULL, cancellable, error);
+}
+
+gboolean
+gst_srt_object_open_full (GstSRTObject * srtobject,
+    GstSRTObjectCallerAdded caller_added_func,
+    GstSRTObjectCallerRemoved caller_removed_func,
+    GCancellable * cancellable, GError ** error)
+{
+  GSocketAddress *socket_address = NULL;
+  GstSRTConnectionMode connection_mode = GST_SRT_CONNECTION_MODE_NONE;
+
+  gpointer sa;
+  size_t sa_len;
+
+  srtobject->opened = FALSE;
+
+  if (caller_added_func != NULL) {
+    srtobject->caller_added_closure =
+        g_cclosure_new (G_CALLBACK (caller_added_func), srtobject, NULL);
+  }
+
+  if (caller_removed_func != NULL) {
+    srtobject->caller_removed_closure =
+        g_cclosure_new (G_CALLBACK (caller_removed_func), srtobject, NULL);
+  }
+
+  socket_address =
+      g_inet_socket_address_new_from_string (gst_uri_get_host (srtobject->uri),
+      gst_uri_get_port (srtobject->uri));
+
+  if (socket_address == NULL) {
+    g_set_error (error, GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_OPEN_READ,
+        "Invalid host");
+    goto out;
+  }
+
+  /* FIXME: Unfortunately, SRT doesn't support IPv4 currently. */
+  if (g_socket_address_get_family (socket_address) != G_SOCKET_FAMILY_IPV4) {
+    g_set_error (error, GST_RESOURCE_ERROR, GST_RESOURCE_ERROR_OPEN_READ,
+        "SRT supports IPv4 only");
+    goto out;
+  }
+
+  sa_len = g_socket_address_get_native_size (socket_address);
+  sa = g_alloca (sa_len);
+
+  if (!g_socket_address_to_native (socket_address, sa, sa_len, error)) {
+    goto out;
+  }
+
+  GST_DEBUG_OBJECT (srtobject->element,
+      "Opening SRT socket with parameters: %" GST_PTR_FORMAT,
+      srtobject->parameters);
+
+  if (!gst_structure_get_enum (srtobject->parameters,
+          "mode", GST_TYPE_SRT_CONNECTION_MODE, (gint *) & connection_mode)) {
+    GST_WARNING_OBJECT (srtobject->element,
+        "Cannot get connection mode information." " Use default mode");
+    connection_mode = GST_TYPE_SRT_CONNECTION_MODE;
+  }
+
+  srtobject->listener_poll_id = srt_epoll_create ();
+
+  srtobject->opened =
+      gst_srt_object_open_connection
+      (srtobject, cancellable, connection_mode, sa, sa_len, error);
+
+out:
+  g_clear_object (&socket_address);
+
+  return srtobject->opened;
+}
+
+void
+gst_srt_object_close (GstSRTObject * srtobject)
+{
+  if (srtobject->poll_id != SRT_ERROR) {
+    srt_epoll_remove_usock (srtobject->poll_id, srtobject->sock);
+  }
+
+  if (srtobject->sock != SRT_INVALID_SOCK) {
+
+    GST_DEBUG_OBJECT (srtobject->element, "Closing SRT socket (0x%x)",
+        srtobject->sock);
+
+    srt_close (srtobject->sock);
+    srtobject->sock = SRT_INVALID_SOCK;
+  }
+
+  if (srtobject->loop) {
+    g_main_loop_quit (srtobject->loop);
+
+    if (srtobject->listener_poll_id != SRT_ERROR) {
+      srt_epoll_remove_usock (srtobject->listener_poll_id,
+          srtobject->listener_sock);
+      srtobject->listener_poll_id = SRT_ERROR;
+    }
+
+    g_thread_join (srtobject->thread);
+
+    g_clear_pointer (&srtobject->thread, g_thread_unref);
+    g_clear_pointer (&srtobject->loop, g_main_loop_unref);
+    g_clear_pointer (&srtobject->context, g_main_context_unref);
+  }
+
+  if (srtobject->listener_sock != SRT_INVALID_SOCK) {
+    GST_DEBUG_OBJECT (srtobject->element, "Closing SRT listener socket (0x%x)",
+        srtobject->listener_sock);
+
+    srt_close (srtobject->listener_sock);
+    srtobject->listener_sock = SRT_INVALID_SOCK;
+  }
+
+  g_list_foreach (srtobject->callers, (GFunc) srt_caller_invoke_removed_closure,
+      srtobject);
+  g_list_free_full (srtobject->callers, (GDestroyNotify) srt_caller_free);
+
+  g_clear_pointer (&srtobject->caller_added_closure,
+      (GDestroyNotify) g_closure_unref);
+  g_clear_pointer (&srtobject->caller_removed_closure,
+      (GDestroyNotify) g_closure_unref);
+
+  srtobject->opened = FALSE;
+}
+
+static gboolean
+gst_srt_object_wait_caller (GstSRTObject * srtobject,
+    GCancellable * cancellable, GError ** errorj)
+{
+  GST_DEBUG_OBJECT (srtobject->element, "Waiting connection from caller");
+
+  if (g_cancellable_is_cancelled (cancellable)) {
+    return FALSE;
+  }
+
+  g_mutex_lock (&srtobject->sock_lock);
+  g_cond_wait (&srtobject->sock_cond, &srtobject->sock_lock);
+  g_mutex_unlock (&srtobject->sock_lock);
+
+  return TRUE;
+}
+
+gssize
+gst_srt_object_read (GstSRTObject * srtobject,
+    guint8 * data, gsize size, GCancellable * cancellable, GError ** error)
+{
+  gssize len = 0;
+  gint poll_timeout;
+  gint msg_size;
+  GstSRTConnectionMode connection_mode = GST_SRT_CONNECTION_MODE_NONE;
+  gint poll_id;
+
+  /* Only source element can read data */
+  g_return_val_if_fail (gst_uri_handler_get_uri_type (GST_URI_HANDLER
+          (srtobject->element)) == GST_URI_SRC, -1);
+
+  gst_structure_get_enum (srtobject->parameters, "mode",
+      GST_TYPE_SRT_CONNECTION_MODE, (gint *) & connection_mode);
+
+  if (connection_mode == GST_SRT_CONNECTION_MODE_LISTENER) {
+    SRTCaller *caller;
+
+    if (g_list_length (srtobject->callers) < 1) {
+      if (!gst_srt_object_wait_caller (srtobject, cancellable, error)) {
+        return -1;
+      }
+    }
+
+    caller = srtobject->callers->data;
+    poll_id = caller->poll_id;
+
+  } else {
+    poll_id = srtobject->poll_id;
+  }
+
+  if (!gst_structure_get_int (srtobject->parameters, "poll-timeout",
+          &poll_timeout)) {
+    poll_timeout = GST_SRT_DEFAULT_POLL_TIMEOUT;
+  }
+
+  if (!gst_structure_get_int (srtobject->parameters, "msg-size", &msg_size)) {
+    msg_size = GST_SRT_DEFAULT_MSG_SIZE;
+  }
+
+  while (!g_cancellable_is_cancelled (cancellable)) {
+
+    SRTSOCKET rsock;
+    gint rsocklen = 1;
+
+    if (srt_epoll_wait (poll_id, &rsock,
+            &rsocklen, 0, 0, poll_timeout, NULL, 0, NULL, 0) < 0) {
+      continue;
+    }
+
+    if (rsocklen < 0) {
+      GST_WARNING_OBJECT (srtobject->element,
+          "abnormal SRT socket is detected");
+      srt_close (rsock);
+    }
+
+    switch (srt_getsockstate (rsock)) {
+      case SRTS_BROKEN:
+      case SRTS_NONEXIST:
+      case SRTS_CLOSED:
+        if (connection_mode == GST_SRT_CONNECTION_MODE_LISTENER) {
+          /* Caller has been disappeared. */
+          return 0;
+        } else {
+          GST_WARNING_OBJECT (srtobject->element,
+              "Invalid SRT socket. Trying to reconnect");
+          gst_srt_object_close (srtobject);
+          if (!gst_srt_object_open (srtobject, cancellable, error)) {
+            return -1;
+          }
+          continue;
+        }
+      case SRTS_CONNECTED:
+        /* good to go */
+        break;
+      default:
+        /* not-ready */
+        continue;
+    }
+
+    while (len < size) {
+      gint recv;
+      gint rest = size - len;
+
+      /* Workaround for SRT being unhappy about buffers that
+       * are less than the chunk size */
+      if (rest < msg_size)
+        goto out;
+
+      recv = srt_recvmsg (rsock, (char *) (data + len), rest);
+
+      if (recv <= 0)
+        goto out;
+
+      len += recv;
+    }
+  }
+
+out:
+  return len;
+}
+
+void
+gst_srt_object_wakeup (GstSRTObject * srtobject)
+{
+  GstSRTConnectionMode connection_mode = GST_SRT_CONNECTION_MODE_NONE;
+
+  GST_DEBUG_OBJECT (srtobject->element, "waking up SRT");
+
+  /* Removing all socket descriptors from the monitoring list
+   * wakes up SRT's threads. We only have one to remove. */
+  srt_epoll_remove_usock (srtobject->poll_id, srtobject->sock);
+
+  gst_structure_get_enum (srtobject->parameters, "mode",
+      GST_TYPE_SRT_CONNECTION_MODE, (gint *) & connection_mode);
+
+  if (connection_mode == GST_SRT_CONNECTION_MODE_LISTENER) {
+    g_mutex_lock (&srtobject->sock_lock);
+    g_cond_signal (&srtobject->sock_cond);
+    g_mutex_unlock (&srtobject->sock_lock);
+  }
+}
+
+static gboolean
+gst_srt_object_send_headers (GstSRTObject * srtobject, SRTSOCKET sock,
+    gint poll_id, gint poll_timeout, GstBufferList * headers,
+    GCancellable * cancellable)
+{
+  guint size, i;
+
+  if (!headers)
+    return TRUE;
+
+  size = gst_buffer_list_length (headers);
+
+  GST_DEBUG_OBJECT (srtobject->element, "Sending %u stream headers", size);
+
+  for (i = 0; i < size; i++) {
+    SRTSOCKET wsock = sock;
+    gint wsocklen = 1;
+
+    GstBuffer *buffer = gst_buffer_list_get (headers, i);
+    GstMapInfo mapinfo;
+
+    if (g_cancellable_is_cancelled (cancellable)) {
+      return FALSE;
+    }
+
+    if (poll_id > 0 && srt_epoll_wait (poll_id, 0, 0, &wsock,
+            &wsocklen, poll_timeout, NULL, 0, NULL, 0) < 0) {
+      continue;
+    }
+
+    GST_TRACE_OBJECT (srtobject->element, "sending header %u %" GST_PTR_FORMAT,
+        i, buffer);
+
+    if (!gst_buffer_map (buffer, &mapinfo, GST_MAP_READ)) {
+      GST_ELEMENT_ERROR (srtobject->element, RESOURCE, READ,
+          ("Could not map the input stream"), (NULL));
+      return FALSE;
+    }
+
+    if (srt_sendmsg2 (wsock, (char *) mapinfo.data, mapinfo.size,
+            0) == SRT_ERROR) {
+      GST_ELEMENT_ERROR (srtobject->element, RESOURCE, WRITE, NULL,
+          ("%s", srt_getlasterror_str ()));
+      gst_buffer_unmap (buffer, &mapinfo);
+      return FALSE;
+    }
+
+    gst_buffer_unmap (buffer, &mapinfo);
+  }
+
+  return TRUE;
+}
+
+static gssize
+gst_srt_object_write_to_callers (GstSRTObject * srtobject,
+    GstBufferList * headers,
+    const GstMapInfo * mapinfo, GCancellable * cancellable, GError ** error)
+{
+  GList *callers = srtobject->callers;
+
+  GST_OBJECT_LOCK (srtobject->element);
+  while (callers != NULL) {
+    gssize len = 0;
+    const guint8 *msg = mapinfo->data;
+    gint sent;
+
+    SRTCaller *caller = callers->data;
+    callers = callers->next;
+
+    if (g_cancellable_is_cancelled (cancellable)) {
+      GST_OBJECT_UNLOCK (srtobject->element);
+      return -1;
+    }
+
+    if (!caller->sent_headers) {
+      if (!gst_srt_object_send_headers (srtobject, caller->sock, -1,
+              -1, headers, cancellable)) {
+        goto err;
+      }
+      caller->sent_headers = TRUE;
+    }
+
+    while (len < mapinfo->size) {
+      gint rest = mapinfo->size - len;
+      sent = srt_sendmsg2 (caller->sock, (char *) (msg + len), rest, 0);
+      if (sent < 0) {
+        goto err;
+      }
+      len += sent;
+    }
+
+    continue;
+
+  err:
+    srtobject->callers = g_list_remove (srtobject->callers, caller);
+    srt_caller_invoke_removed_closure (caller, srtobject);
+    GST_OBJECT_UNLOCK (srtobject->element);
+    srt_caller_free (caller);
+    GST_OBJECT_LOCK (srtobject->element);
+  }
+
+  GST_OBJECT_UNLOCK (srtobject->element);
+
+  return mapinfo->size;
+}
+
+static gssize
+gst_srt_object_write_one (GstSRTObject * srtobject,
+    GstBufferList * headers,
+    const GstMapInfo * mapinfo, GCancellable * cancellable, GError ** error)
+{
+  gssize len = 0;
+  gint poll_timeout;
+  const guint8 *msg = mapinfo->data;
+
+  if (!gst_structure_get_int (srtobject->parameters, "poll-timeout",
+          &poll_timeout)) {
+    poll_timeout = GST_SRT_DEFAULT_POLL_TIMEOUT;
+  }
+
+  if (!srtobject->sent_headers) {
+    if (!gst_srt_object_send_headers (srtobject, srtobject->sock,
+            srtobject->poll_id, poll_timeout, headers, cancellable)) {
+      return -1;
+    }
+    srtobject->sent_headers = TRUE;
+  }
+
+  while (len < mapinfo->size) {
+    SRTSOCKET wsock;
+    gint wsocklen = 1;
+
+    gint sent;
+    gint rest = mapinfo->size - len;
+
+    if (g_cancellable_is_cancelled (cancellable)) {
+      break;
+    }
+
+    if (srt_epoll_wait (srtobject->poll_id, 0, 0, &wsock,
+            &wsocklen, poll_timeout, NULL, 0, NULL, 0) < 0) {
+      continue;
+    }
+
+    switch (srt_getsockstate (wsock)) {
+      case SRTS_BROKEN:
+      case SRTS_NONEXIST:
+      case SRTS_CLOSED:
+        GST_WARNING_OBJECT (srtobject->element,
+            "Invalid SRT socket. Trying to reconnect");
+        gst_srt_object_close (srtobject);
+        if (!gst_srt_object_open (srtobject, cancellable, error)) {
+          return -1;
+        }
+        continue;
+      case SRTS_CONNECTED:
+        /* good to go */
+        GST_WARNING_OBJECT (srtobject->element, "good to go");
+        break;
+      default:
+        GST_WARNING_OBJECT (srtobject->element, "not ready");
+        /* not-ready */
+        continue;
+    }
+
+    sent = srt_sendmsg2 (wsock, (char *) (msg + len), rest, 0);
+    if (sent < 0) {
+      GST_ELEMENT_ERROR (srtobject->element, RESOURCE, WRITE, NULL,
+          ("%s", srt_getlasterror_str ()));
+      break;
+    }
+    len += sent;
+  }
+
+  return len;
+}
+
+gssize
+gst_srt_object_write (GstSRTObject * srtobject,
+    GstBufferList * headers,
+    const GstMapInfo * mapinfo, GCancellable * cancellable, GError ** error)
+{
+  gssize len = 0;
+  GstSRTConnectionMode connection_mode = GST_SRT_CONNECTION_MODE_NONE;
+
+  /* Only sink element can write data */
+  g_return_val_if_fail (gst_uri_handler_get_uri_type (GST_URI_HANDLER
+          (srtobject->element)) == GST_URI_SINK, -1);
+
+  gst_structure_get_enum (srtobject->parameters, "mode",
+      GST_TYPE_SRT_CONNECTION_MODE, (gint *) & connection_mode);
+
+  if (connection_mode == GST_SRT_CONNECTION_MODE_LISTENER) {
+    if (g_list_length (srtobject->callers) < 1) {
+      if (!gst_srt_object_wait_caller (srtobject, cancellable, error)) {
+        return -1;
+      }
+    }
+    len =
+        gst_srt_object_write_to_callers (srtobject, headers, mapinfo,
+        cancellable, error);
+  } else {
+    len =
+        gst_srt_object_write_one (srtobject, headers, mapinfo, cancellable,
+        error);
+  }
+
+  return len;
+}
+
+GstStructure *
+gst_srt_object_get_stats (GstSRTObject * srtobject)
+{
+  SRT_TRACEBSTATS stats;
+  int ret;
+  GstStructure *s = gst_structure_new_empty ("application/x-srt-statistics");
+
+  /* FIXME: what if ruinning on listener mode */
+  if (srtobject->sock == SRT_INVALID_SOCK)
+    return s;
+
+  ret = srt_bstats (srtobject->sock, &stats, 0);
+
+  if (ret >= 0) {
+    gst_structure_set (s,
+        /* number of sent data packets, including retransmissions */
+        "packets-sent", G_TYPE_INT64, stats.pktSent,
+        /* number of lost packets (sender side) */
+        "packets-sent-lost", G_TYPE_INT, stats.pktSndLoss,
+        /* number of retransmitted packets */
+        "packets-retransmitted", G_TYPE_INT, stats.pktRetrans,
+        /* number of received ACK packets */
+        "packet-ack-received", G_TYPE_INT, stats.pktRecvACK,
+        /* number of received NAK packets */
+        "packet-nack-received", G_TYPE_INT, stats.pktRecvNAK,
+        /* time duration when UDT is sending data (idle time exclusive) */
+        "send-duration-us", G_TYPE_INT64, stats.usSndDuration,
+        /* number of sent data bytes, including retransmissions */
+        "bytes-sent", G_TYPE_UINT64, stats.byteSent,
+        /* number of retransmitted bytes */
+        "bytes-retransmitted", G_TYPE_UINT64, stats.byteRetrans,
+        /* number of too-late-to-send dropped bytes */
+        "bytes-sent-dropped", G_TYPE_UINT64, stats.byteSndDrop,
+        /* number of too-late-to-send dropped packets */
+        "packets-sent-dropped", G_TYPE_INT, stats.pktSndDrop,
+        /* sending rate in Mb/s */
+        "send-rate-mbps", G_TYPE_DOUBLE, stats.msRTT,
+        /* estimated bandwidth, in Mb/s */
+        "bandwidth-mbps", G_TYPE_DOUBLE, stats.mbpsBandwidth,
+        /* busy sending time (i.e., idle time exclusive) */
+        "send-duration-us", G_TYPE_UINT64, stats.usSndDuration,
+        "rtt-ms", G_TYPE_DOUBLE, stats.msRTT,
+        "negotiated-latency-ms", G_TYPE_INT, stats.msSndTsbPdDelay, NULL);
+  }
+
+  return s;
+}
diff --git a/ext/srt/gstsrtobject.h b/ext/srt/gstsrtobject.h
new file mode 100644
index 000000000..8dad97536
--- /dev/null
+++ b/ext/srt/gstsrtobject.h
@@ -0,0 +1,127 @@
+/* GStreamer
+ * Copyright (C) 2018, Collabora Ltd.
+ * Copyright (C) 2018, SK Telecom, Co., Ltd.
+ *   Author: Jeongseok Kim <jeongseok.kim@sk.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __GST_SRT_OBJECT_H__
+#define __GST_SRT_OBJECT_H__
+
+#include "gstsrt-enums.h"
+#include "gstsrt-enumtypes.h"
+
+#include <gio/gio.h>
+#include <srt/srt.h>
+
+G_BEGIN_DECLS
+
+#define GST_SRT_DEFAULT_URI_SCHEME "srt"
+#define GST_SRT_DEFAULT_PORT 7001
+#define GST_SRT_DEFAULT_HOST "127.0.0.1"
+#define GST_SRT_DEFAULT_LOCALADDRESS "0.0.0.0"
+#define GST_SRT_DEFAULT_URI GST_SRT_DEFAULT_URI_SCHEME"://"GST_SRT_DEFAULT_HOST":"G_STRINGIFY(GST_SRT_DEFAULT_PORT)
+
+#define GST_SRT_DEFAULT_MODE GST_SRT_CONNECTION_MODE_CALLER
+#define GST_SRT_DEFAULT_PBKEYLEN GST_SRT_KEY_LENGTH_BITS_128
+#define GST_SRT_DEFAULT_POLL_TIMEOUT -1
+#define GST_SRT_DEFAULT_LATENCY 125
+#define GST_SRT_DEFAULT_MSG_SIZE 1316
+#define GST_SRT_DEFAULT_KEY_LENGTH 16
+
+typedef struct _GstSRTObject GstSRTObject;
+
+struct _GstSRTObject
+{
+  GstElement                   *element;
+  GstUri                       *uri;
+
+  GstStructure                 *parameters;
+  gboolean                      opened;
+  SRTSOCKET                     sock;
+  gint                          poll_id;
+  gboolean                      sent_headers;
+
+  GMutex                        sock_lock;
+  GCond                         sock_cond;
+
+  GTask                        *listener_task;
+  SRTSOCKET                     listener_sock;
+  gint                          listener_poll_id;
+
+  GMainLoop                    *loop;
+  GMainContext                 *context;
+  GSource                      *listener_source;
+  GThread                      *thread;
+
+  GList                        *callers;
+
+  GClosure                     *caller_added_closure;
+  GClosure                     *caller_removed_closure;
+};
+
+
+typedef void (*GstSRTObjectCallerAdded)         (int sock, GSocketAddress *addr, GstSRTObject * srtobject);
+
+typedef void (*GstSRTObjectCallerRemoved)       (int sock, GSocketAddress *addr, GstSRTObject * srtobject);
+
+GstSRTObject   *gst_srt_object_new              (GstElement *element);
+
+void            gst_srt_object_destroy          (GstSRTObject *srtobject);
+
+gboolean        gst_srt_object_open             (GstSRTObject *srtobject,
+                                                 GCancellable *cancellable,
+                                                 GError **error);
+
+gboolean        gst_srt_object_open_full        (GstSRTObject *srtobject,
+                                                 GstSRTObjectCallerAdded caller_added_func,
+                                                 GstSRTObjectCallerRemoved caller_removed_func,
+                                                 GCancellable *cancellable,
+                                                 GError **error);
+
+void            gst_srt_object_close            (GstSRTObject *srtobject);
+
+gboolean        gst_srt_object_set_property_helper (GstSRTObject *srtobject, 
+                                                    guint prop_id, const GValue * value,
+                                                    GParamSpec * pspec);
+
+gboolean        gst_srt_object_get_property_helper (GstSRTObject *srtobject, 
+                                                    guint prop_id, GValue * value,
+                                                    GParamSpec * pspec);
+
+void            gst_srt_object_install_properties_helper (GObjectClass *gobject_class);
+
+gboolean        gst_srt_object_set_uri (GstSRTObject * srtobject, const gchar *uri, GError ** err);
+
+gssize          gst_srt_object_read     (GstSRTObject * srtobject, 
+                                         guint8 *data, gsize size,
+                                         GCancellable *cancellable,
+                                         GError **err);
+
+gssize          gst_srt_object_write    (GstSRTObject * srtobject, 
+                                         GstBufferList * headers,
+                                         const GstMapInfo * mapinfo,
+                                         GCancellable *cancellable,
+                                         GError **err);
+
+void            gst_srt_object_wakeup   (GstSRTObject * srtobject);
+
+GstStructure   *gst_srt_object_get_stats        (GstSRTObject * srtobject);
+
+G_END_DECLS
+
+#endif // __GST_SRT_OBJECT_H__
diff --git a/ext/srt/gstsrtserversink.c b/ext/srt/gstsrtserversink.c
deleted file mode 100644
index 66cf91916..000000000
--- a/ext/srt/gstsrtserversink.c
+++ /dev/null
@@ -1,516 +0,0 @@
-/* GStreamer SRT plugin based on libsrt
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-/**
- * SECTION:element-srtserversink
- * @title: srtserversink
- *
- * srtserversink is a network sink that sends <ulink url="http://www.srtalliance.org/">SRT</ulink>
- * packets to the network. Although SRT is an UDP-based protocol, srtserversink works like
- * a server socket of connection-oriented protocol.
- *
- * <refsect2>
- * <title>Examples</title>
- * |[
- * gst-launch-1.0 -v audiotestsrc ! srtserversink
- * ]| This pipeline shows how to serve SRT packets through the default port.
- * </refsect2>
- *
- */
-
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
-#include "gstsrtserversink.h"
-#include "gstsrt.h"
-#include <srt/srt.h>
-#include <gio/gio.h>
-
-#define SRT_DEFAULT_POLL_TIMEOUT -1
-
-static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
-    GST_PAD_SINK,
-    GST_PAD_ALWAYS,
-    GST_STATIC_CAPS_ANY);
-
-#define GST_CAT_DEFAULT gst_debug_srt_server_sink
-GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
-
-struct _GstSRTServerSinkPrivate
-{
-  gboolean cancelled;
-
-  SRTSOCKET sock;
-  gint poll_id;
-  gint poll_timeout;
-
-  GMainLoop *loop;
-  GMainContext *context;
-  GSource *server_source;
-  GThread *thread;
-
-  GList *clients;
-};
-
-#define GST_SRT_SERVER_SINK_GET_PRIVATE(obj)  \
-       (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_SRT_SERVER_SINK, GstSRTServerSinkPrivate))
-
-enum
-{
-  PROP_POLL_TIMEOUT = 1,
-  PROP_STATS,
-  /*< private > */
-  PROP_LAST
-};
-
-static GParamSpec *properties[PROP_LAST];
-
-enum
-{
-  SIG_CLIENT_ADDED,
-  SIG_CLIENT_REMOVED,
-
-  LAST_SIGNAL
-};
-
-static guint signals[LAST_SIGNAL] = { 0 };
-
-#define gst_srt_server_sink_parent_class parent_class
-G_DEFINE_TYPE_WITH_CODE (GstSRTServerSink, gst_srt_server_sink,
-    GST_TYPE_SRT_BASE_SINK, G_ADD_PRIVATE (GstSRTServerSink)
-    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtserversink", 0,
-        "SRT Server Sink"));
-
-typedef struct
-{
-  int sock;
-  GSocketAddress *sockaddr;
-  gboolean sent_headers;
-} SRTClient;
-
-static SRTClient *
-srt_client_new (void)
-{
-  SRTClient *client = g_new0 (SRTClient, 1);
-  client->sock = SRT_INVALID_SOCK;
-  return client;
-}
-
-static void
-srt_client_free (SRTClient * client)
-{
-  g_return_if_fail (client != NULL);
-
-  g_clear_object (&client->sockaddr);
-
-  if (client->sock != SRT_INVALID_SOCK) {
-    srt_close (client->sock);
-  }
-
-  g_free (client);
-}
-
-static void
-srt_emit_client_removed (SRTClient * client, gpointer user_data)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (user_data);
-  g_return_if_fail (client != NULL && GST_IS_SRT_SERVER_SINK (self));
-
-  g_signal_emit (self, signals[SIG_CLIENT_REMOVED], 0, client->sock,
-      client->sockaddr);
-}
-
-static void
-gst_srt_server_sink_get_property (GObject * object,
-    guint prop_id, GValue * value, GParamSpec * pspec)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (object);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-
-  switch (prop_id) {
-    case PROP_POLL_TIMEOUT:
-      g_value_set_int (value, priv->poll_timeout);
-      break;
-    case PROP_STATS:
-    {
-      GList *item;
-
-      GST_OBJECT_LOCK (self);
-      for (item = priv->clients; item; item = item->next) {
-        SRTClient *client = item->data;
-        GValue tmp = G_VALUE_INIT;
-
-        g_value_init (&tmp, GST_TYPE_STRUCTURE);
-        g_value_take_boxed (&tmp, gst_srt_base_sink_get_stats (client->sockaddr,
-                client->sock));
-        gst_value_array_append_and_take_value (value, &tmp);
-      }
-      GST_OBJECT_UNLOCK (self);
-      break;
-    }
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_server_sink_set_property (GObject * object,
-    guint prop_id, const GValue * value, GParamSpec * pspec)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (object);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-
-  switch (prop_id) {
-    case PROP_POLL_TIMEOUT:
-      priv->poll_timeout = g_value_get_int (value);
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static gboolean
-idle_listen_callback (gpointer data)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (data);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-  gboolean ret = TRUE;
-
-  SRTClient *client;
-  SRTSOCKET ready[2];
-  struct sockaddr sa;
-  int sa_len;
-
-  if (srt_epoll_wait (priv->poll_id, ready, &(int) {
-          2}, 0, 0, priv->poll_timeout, 0, 0, 0, 0) == -1) {
-    int srt_errno = srt_getlasterror (NULL);
-
-    if (srt_errno != SRT_ETIMEOUT) {
-      GST_ELEMENT_ERROR (self, RESOURCE, FAILED,
-          ("SRT error: %s", srt_getlasterror_str ()), (NULL));
-      ret = FALSE;
-      goto out;
-    }
-
-    /* Mimicking cancellable */
-    if (srt_errno == SRT_ETIMEOUT && priv->cancelled) {
-      GST_DEBUG_OBJECT (self, "Cancelled waiting for client");
-      ret = FALSE;
-      goto out;
-    }
-  }
-
-  client = srt_client_new ();
-  client->sock = srt_accept (priv->sock, &sa, &sa_len);
-
-  if (client->sock == SRT_INVALID_SOCK) {
-    GST_WARNING_OBJECT (self, "detected invalid SRT client socket (reason: %s)",
-        srt_getlasterror_str ());
-    srt_clearlasterror ();
-    srt_client_free (client);
-    ret = FALSE;
-    goto out;
-  }
-
-  client->sockaddr = g_socket_address_new_from_native (&sa, sa_len);
-
-  GST_OBJECT_LOCK (self);
-  priv->clients = g_list_append (priv->clients, client);
-  GST_OBJECT_UNLOCK (self);
-
-  g_signal_emit (self, signals[SIG_CLIENT_ADDED], 0, client->sock,
-      client->sockaddr);
-  GST_DEBUG_OBJECT (self, "client added");
-
-out:
-  return ret;
-}
-
-static gpointer
-thread_func (gpointer data)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (data);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-
-  g_main_loop_run (priv->loop);
-
-  return NULL;
-}
-
-static gboolean
-gst_srt_server_sink_start (GstBaseSink * sink)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (sink);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-  GstSRTBaseSink *base = GST_SRT_BASE_SINK (sink);
-  GstUri *uri = gst_uri_ref (GST_SRT_BASE_SINK (self)->uri);
-  GError *error = NULL;
-  gboolean ret = TRUE;
-  const gchar *host;
-
-  if (gst_uri_get_port (uri) == GST_URI_NO_PORT) {
-    GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_WRITE, NULL, (("Invalid port")));
-    return FALSE;
-  }
-
-  host = gst_uri_get_host (uri);
-
-  priv->sock = gst_srt_server_listen (GST_ELEMENT (self),
-      TRUE, host, gst_uri_get_port (uri),
-      base->latency, &priv->poll_id, base->passphrase, base->key_length);
-
-  if (priv->sock == SRT_INVALID_SOCK) {
-    GST_ERROR_OBJECT (sink, "Failed to create srt socket");
-    goto failed;
-  }
-
-  priv->context = g_main_context_new ();
-
-  priv->server_source = g_idle_source_new ();
-  g_source_set_callback (priv->server_source,
-      (GSourceFunc) idle_listen_callback, gst_object_ref (self),
-      (GDestroyNotify) gst_object_unref);
-
-  g_source_attach (priv->server_source, priv->context);
-  priv->loop = g_main_loop_new (priv->context, TRUE);
-
-  priv->thread = g_thread_try_new ("srtserversink", thread_func, self, &error);
-  if (error != NULL) {
-    GST_WARNING_OBJECT (self, "failed to create thread (reason: %s)",
-        error->message);
-    ret = FALSE;
-  }
-
-  g_clear_pointer (&uri, gst_uri_unref);
-
-  return ret;
-
-failed:
-  if (priv->poll_id != SRT_ERROR) {
-    srt_epoll_release (priv->poll_id);
-    priv->poll_id = SRT_ERROR;
-  }
-
-  if (priv->sock != SRT_INVALID_SOCK) {
-    srt_close (priv->sock);
-    priv->sock = SRT_INVALID_SOCK;
-  }
-
-  g_clear_error (&error);
-  g_clear_pointer (&uri, gst_uri_unref);
-
-  return FALSE;
-}
-
-static gboolean
-send_buffer_internal (GstSRTBaseSink * sink,
-    const GstMapInfo * mapinfo, gpointer user_data)
-{
-  SRTClient *client = user_data;
-
-  if (srt_sendmsg2 (client->sock, (char *) mapinfo->data, mapinfo->size,
-          0) == SRT_ERROR) {
-    GST_WARNING_OBJECT (sink, "%s", srt_getlasterror_str ());
-    return FALSE;
-  }
-
-  return TRUE;
-}
-
-static gboolean
-gst_srt_server_sink_send_buffer (GstSRTBaseSink * sink,
-    const GstMapInfo * mapinfo)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (sink);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-  GList *clients = priv->clients;
-
-  GST_OBJECT_LOCK (sink);
-  while (clients != NULL) {
-    SRTClient *client = clients->data;
-    clients = clients->next;
-
-    if (!client->sent_headers) {
-      if (!gst_srt_base_sink_send_headers (sink, send_buffer_internal, client))
-        goto err;
-
-      client->sent_headers = TRUE;
-    }
-
-    if (!send_buffer_internal (sink, mapinfo, client))
-      goto err;
-
-    continue;
-
-  err:
-    priv->clients = g_list_remove (priv->clients, client);
-    GST_OBJECT_UNLOCK (sink);
-    g_signal_emit (self, signals[SIG_CLIENT_REMOVED], 0, client->sock,
-        client->sockaddr);
-    srt_client_free (client);
-    GST_OBJECT_LOCK (sink);
-  }
-  GST_OBJECT_UNLOCK (sink);
-
-  return TRUE;
-}
-
-static gboolean
-gst_srt_server_sink_stop (GstBaseSink * sink)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (sink);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-  GList *clients;
-
-  GST_DEBUG_OBJECT (self, "closing client sockets");
-
-  GST_OBJECT_LOCK (sink);
-  clients = priv->clients;
-  priv->clients = NULL;
-  GST_OBJECT_UNLOCK (sink);
-
-  g_list_foreach (clients, (GFunc) srt_emit_client_removed, self);
-  g_list_free_full (clients, (GDestroyNotify) srt_client_free);
-
-  GST_DEBUG_OBJECT (self, "closing SRT connection");
-  srt_epoll_remove_usock (priv->poll_id, priv->sock);
-  srt_epoll_release (priv->poll_id);
-  srt_close (priv->sock);
-
-  if (priv->loop) {
-    g_main_loop_quit (priv->loop);
-    g_thread_join (priv->thread);
-    g_clear_pointer (&priv->loop, g_main_loop_unref);
-    g_clear_pointer (&priv->thread, g_thread_unref);
-  }
-
-  if (priv->server_source) {
-    g_source_destroy (priv->server_source);
-    g_clear_pointer (&priv->server_source, g_source_unref);
-  }
-
-  g_clear_pointer (&priv->context, g_main_context_unref);
-
-  return GST_BASE_SINK_CLASS (parent_class)->stop (sink);
-}
-
-static gboolean
-gst_srt_server_sink_unlock (GstBaseSink * sink)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (sink);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-
-  priv->cancelled = TRUE;
-
-  return TRUE;
-}
-
-static gboolean
-gst_srt_server_sink_unlock_stop (GstBaseSink * sink)
-{
-  GstSRTServerSink *self = GST_SRT_SERVER_SINK (sink);
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-
-  priv->cancelled = FALSE;
-
-  return TRUE;
-}
-
-static void
-gst_srt_server_sink_class_init (GstSRTServerSinkClass * klass)
-{
-  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
-  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
-  GstBaseSinkClass *gstbasesink_class = GST_BASE_SINK_CLASS (klass);
-  GstSRTBaseSinkClass *gstsrtbasesink_class = GST_SRT_BASE_SINK_CLASS (klass);
-
-  gobject_class->set_property = gst_srt_server_sink_set_property;
-  gobject_class->get_property = gst_srt_server_sink_get_property;
-
-  properties[PROP_POLL_TIMEOUT] =
-      g_param_spec_int ("poll-timeout", "Poll Timeout",
-      "Return poll wait after timeout miliseconds (-1 = infinite)", -1,
-      G_MAXINT32, SRT_DEFAULT_POLL_TIMEOUT,
-      G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  properties[PROP_STATS] = gst_param_spec_array ("stats", "Statistics",
-      "Array of GstStructures containing SRT statistics",
-      g_param_spec_boxed ("stats", "Statistics",
-          "Statistics for one client", GST_TYPE_STRUCTURE,
-          G_PARAM_READABLE | G_PARAM_STATIC_STRINGS),
-      G_PARAM_READABLE | G_PARAM_STATIC_STRINGS);
-
-  g_object_class_install_properties (gobject_class, PROP_LAST, properties);
-
-  /**
-   * GstSRTServerSink::client-added:
-   * @gstsrtserversink: the srtserversink element that emitted this signal
-   * @sock: the client socket descriptor that was added to srtserversink
-   * @addr: the pointer of "struct sockaddr" that describes the @sock
-   * @addr_len: the length of @addr
-   *
-   * The given socket descriptor was added to srtserversink.
-   */
-  signals[SIG_CLIENT_ADDED] =
-      g_signal_new ("client-added", G_TYPE_FROM_CLASS (klass),
-      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTServerSinkClass, client_added),
-      NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
-      2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
-
-  /**
-   * GstSRTServerSink::client-removed:
-   * @gstsrtserversink: the srtserversink element that emitted this signal
-   * @sock: the client socket descriptor that was added to srtserversink
-   * @addr: the pointer of "struct sockaddr" that describes the @sock
-   * @addr_len: the length of @addr
-   *
-   * The given socket descriptor was removed from srtserversink.
-   */
-  signals[SIG_CLIENT_REMOVED] =
-      g_signal_new ("client-removed", G_TYPE_FROM_CLASS (klass),
-      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTServerSinkClass,
-          client_removed), NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
-      2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
-
-  gst_element_class_add_static_pad_template (gstelement_class, &sink_template);
-  gst_element_class_set_metadata (gstelement_class,
-      "SRT server sink", "Sink/Network",
-      "Send data over the network via SRT",
-      "Justin Kim <justin.kim@collabora.com>");
-
-  gstbasesink_class->start = GST_DEBUG_FUNCPTR (gst_srt_server_sink_start);
-  gstbasesink_class->stop = GST_DEBUG_FUNCPTR (gst_srt_server_sink_stop);
-  gstbasesink_class->unlock = GST_DEBUG_FUNCPTR (gst_srt_server_sink_unlock);
-  gstbasesink_class->unlock_stop =
-      GST_DEBUG_FUNCPTR (gst_srt_server_sink_unlock_stop);
-
-  gstsrtbasesink_class->send_buffer =
-      GST_DEBUG_FUNCPTR (gst_srt_server_sink_send_buffer);
-}
-
-static void
-gst_srt_server_sink_init (GstSRTServerSink * self)
-{
-  GstSRTServerSinkPrivate *priv = GST_SRT_SERVER_SINK_GET_PRIVATE (self);
-  priv->poll_timeout = SRT_DEFAULT_POLL_TIMEOUT;
-}
diff --git a/ext/srt/gstsrtserversink.h b/ext/srt/gstsrtserversink.h
deleted file mode 100644
index 537dd0b0a..000000000
--- a/ext/srt/gstsrtserversink.h
+++ /dev/null
@@ -1,62 +0,0 @@
-/* GStreamer
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifndef __GST_SRT_SERVER_SINK_H__
-#define __GST_SRT_SERVER_SINK_H__
-
-#include "gstsrtbasesink.h"
-
-G_BEGIN_DECLS
-
-#define GST_TYPE_SRT_SERVER_SINK              (gst_srt_server_sink_get_type ())
-#define GST_IS_SRT_SERVER_SINK(obj)           (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SRT_SERVER_SINK))
-#define GST_IS_SRT_SERVER_SINK_CLASS(klass)   (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SRT_SERVER_SINK))
-#define GST_SRT_SERVER_SINK_GET_CLASS(obj)    (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_SRT_SERVER_SINK, GstSRTServerSinkClass))
-#define GST_SRT_SERVER_SINK(obj)              (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SRT_SERVER_SINK, GstSRTServerSink))
-#define GST_SRT_SERVER_SINK_CLASS(klass)      (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SRT_SERVER_SINK, GstSRTServerSinkClass))
-#define GST_SRT_SERVER_SINK_CAST(obj)         ((GstSRTServerSink*)(obj))
-#define GST_SRT_SERVER_SINK_CLASS_CAST(klass) ((GstSRTServerSinkClass*)(klass))
-
-typedef struct _GstSRTServerSink GstSRTServerSink;
-typedef struct _GstSRTServerSinkClass GstSRTServerSinkClass;
-typedef struct _GstSRTServerSinkPrivate GstSRTServerSinkPrivate;
-
-struct _GstSRTServerSink {
-  GstSRTBaseSink parent;
-
-  /*< private >*/
-  gpointer _gst_reserved[GST_PADDING];
-};
-
-struct _GstSRTServerSinkClass {
-  GstSRTBaseSinkClass parent_class;
-
-  void (*client_added)      (GstSRTServerSink *self, int sock, GSocketAddress *addr);
-  void (*client_removed)    (GstSRTServerSink *self, int sock, GSocketAddress *addr);
-
-  gpointer _gst_reserved[GST_PADDING_LARGE];
-};
-
-GST_EXPORT
-GType gst_srt_server_sink_get_type (void);
-
-G_END_DECLS
-
-#endif /* __GST_SRT_SERVER_SINK_H__ */
diff --git a/ext/srt/gstsrtserversrc.c b/ext/srt/gstsrtserversrc.c
deleted file mode 100644
index 0b4f4a7cf..000000000
--- a/ext/srt/gstsrtserversrc.c
+++ /dev/null
@@ -1,426 +0,0 @@
-/* GStreamer SRT plugin based on libsrt
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-/**
- * SECTION:element-srtserversrc
- * @title: srtserversrc
- *
- * srtserversrc is a network source that reads <ulink url="http://www.srtalliance.org/">SRT</ulink>
- * packets from the network. Although SRT is a protocol based on UDP, srtserversrc works like
- * a server socket of connection-oriented protocol, but it accepts to only one client connection.
- *
- * <refsect2>
- * <title>Examples</title>
- * |[
- * gst-launch-1.0 -v srtserversrc uri="srt://:7001" ! fakesink
- * ]| This pipeline shows how to bind SRT server by setting #GstSRTServerSrc:uri property.
- * </refsect2>
- *
- */
-
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
-#include "gstsrtserversrc.h"
-#include "gstsrt.h"
-#include <gio/gio.h>
-
-#define SRT_DEFAULT_POLL_TIMEOUT 100
-
-static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
-    GST_PAD_SRC,
-    GST_PAD_ALWAYS,
-    GST_STATIC_CAPS_ANY);
-
-#define GST_CAT_DEFAULT gst_debug_srt_server_src
-GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
-
-struct _GstSRTServerSrcPrivate
-{
-  SRTSOCKET sock;
-  SRTSOCKET client_sock;
-  GSocketAddress *client_sockaddr;
-
-  gint poll_id;
-  gint poll_timeout;
-
-  gboolean has_client;
-  gboolean cancelled;
-};
-
-#define GST_SRT_SERVER_SRC_GET_PRIVATE(obj)  \
-       (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_SRT_SERVER_SRC, GstSRTServerSrcPrivate))
-
-enum
-{
-  PROP_POLL_TIMEOUT = 1,
-
-  /*< private > */
-  PROP_LAST
-};
-
-static GParamSpec *properties[PROP_LAST];
-
-enum
-{
-  SIG_CLIENT_ADDED,
-  SIG_CLIENT_CLOSED,
-
-  LAST_SIGNAL
-};
-
-static guint signals[LAST_SIGNAL] = { 0 };
-
-#define gst_srt_server_src_parent_class parent_class
-G_DEFINE_TYPE_WITH_CODE (GstSRTServerSrc, gst_srt_server_src,
-    GST_TYPE_SRT_BASE_SRC, G_ADD_PRIVATE (GstSRTServerSrc)
-    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtserversrc", 0,
-        "SRT Server Source"));
-
-static void
-gst_srt_server_src_get_property (GObject * object,
-    guint prop_id, GValue * value, GParamSpec * pspec)
-{
-  GstSRTServerSrc *self = GST_SRT_SERVER_SRC (object);
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-
-  switch (prop_id) {
-    case PROP_POLL_TIMEOUT:
-      g_value_set_int (value, priv->poll_timeout);
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_server_src_set_property (GObject * object,
-    guint prop_id, const GValue * value, GParamSpec * pspec)
-{
-  GstSRTServerSrc *self = GST_SRT_SERVER_SRC (object);
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-
-  switch (prop_id) {
-    case PROP_POLL_TIMEOUT:
-      priv->poll_timeout = g_value_get_int (value);
-      break;
-    default:
-      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
-      break;
-  }
-}
-
-static void
-gst_srt_server_src_finalize (GObject * object)
-{
-  GstSRTServerSrc *self = GST_SRT_SERVER_SRC (object);
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-
-  if (priv->poll_id != SRT_ERROR) {
-    srt_epoll_release (priv->poll_id);
-    priv->poll_id = SRT_ERROR;
-  }
-
-  if (priv->sock != SRT_ERROR) {
-    srt_close (priv->sock);
-    priv->sock = SRT_ERROR;
-  }
-
-  G_OBJECT_CLASS (parent_class)->finalize (object);
-}
-
-static GstFlowReturn
-gst_srt_server_src_fill (GstPushSrc * src, GstBuffer * outbuf)
-{
-  GstSRTServerSrc *self = GST_SRT_SERVER_SRC (src);
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-  GstFlowReturn ret = GST_FLOW_OK;
-  GstMapInfo info;
-  SRTSOCKET ready[2];
-  gint recv_len;
-  struct sockaddr client_sa;
-  size_t client_sa_len;
-
-  while (!priv->has_client) {
-    GST_DEBUG_OBJECT (self, "poll wait (timeout: %d)", priv->poll_timeout);
-
-    if (srt_epoll_wait (priv->poll_id, ready, &(int) {
-            2}, 0, 0, priv->poll_timeout, 0, 0, 0, 0) == -1) {
-      int srt_errno = srt_getlasterror (NULL);
-
-      /* Assuming that timeout error is normal */
-      if (srt_errno != SRT_ETIMEOUT) {
-        GST_ELEMENT_ERROR (src, RESOURCE, FAILED,
-            ("SRT error: %s", srt_getlasterror_str ()), (NULL));
-
-        return GST_FLOW_ERROR;
-      }
-
-      /* Mimicking cancellable */
-      if (srt_errno == SRT_ETIMEOUT && priv->cancelled) {
-        GST_DEBUG_OBJECT (self, "Cancelled waiting for client");
-        return GST_FLOW_FLUSHING;
-      }
-
-      continue;
-    }
-
-    priv->client_sock =
-        srt_accept (priv->sock, &client_sa, (int *) &client_sa_len);
-
-    GST_DEBUG_OBJECT (self, "checking client sock");
-    if (priv->client_sock == SRT_INVALID_SOCK) {
-      GST_WARNING_OBJECT (self,
-          "detected invalid SRT client socket (reason: %s)",
-          srt_getlasterror_str ());
-      srt_clearlasterror ();
-    } else {
-      priv->has_client = TRUE;
-      g_clear_object (&priv->client_sockaddr);
-      priv->client_sockaddr = g_socket_address_new_from_native (&client_sa,
-          client_sa_len);
-      g_signal_emit (self, signals[SIG_CLIENT_ADDED], 0,
-          priv->client_sock, priv->client_sockaddr);
-    }
-  }
-
-  GST_DEBUG_OBJECT (self, "filling buffer");
-
-  if (!gst_buffer_map (outbuf, &info, GST_MAP_WRITE)) {
-    GST_ELEMENT_ERROR (src, RESOURCE, WRITE,
-        ("Could not map the output stream"), (NULL));
-    ret = GST_FLOW_ERROR;
-    goto out;
-  }
-
-  recv_len = srt_recvmsg (priv->client_sock, (char *) info.data,
-      gst_buffer_get_size (outbuf));
-
-  gst_buffer_unmap (outbuf, &info);
-
-  if (recv_len == SRT_ERROR) {
-    GST_WARNING_OBJECT (self, "%s", srt_getlasterror_str ());
-
-    g_signal_emit (self, signals[SIG_CLIENT_CLOSED], 0,
-        priv->client_sock, priv->client_sockaddr);
-
-    srt_close (priv->client_sock);
-    priv->client_sock = SRT_INVALID_SOCK;
-    g_clear_object (&priv->client_sockaddr);
-    priv->has_client = FALSE;
-    gst_buffer_resize (outbuf, 0, 0);
-    ret = GST_FLOW_OK;
-    goto out;
-  } else if (recv_len == 0) {
-    ret = GST_FLOW_EOS;
-    goto out;
-  }
-
-  gst_buffer_resize (outbuf, 0, recv_len);
-
-  GST_LOG_OBJECT (src, "filled buffer from _get of size %" G_GSIZE_FORMAT,
-      gst_buffer_get_size (outbuf));
-
-out:
-  return ret;
-}
-
-static gboolean
-gst_srt_server_src_start (GstBaseSrc * src)
-{
-  GstSRTServerSrc *self = GST_SRT_SERVER_SRC (src);
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-  GstSRTBaseSrc *base = GST_SRT_BASE_SRC (src);
-  GstUri *uri = gst_uri_ref (base->uri);
-  const gchar *host;
-
-  if (gst_uri_get_port (uri) == GST_URI_NO_PORT) {
-    GST_ELEMENT_ERROR (src, RESOURCE, OPEN_WRITE, NULL, (("Invalid port")));
-    return FALSE;
-  }
-
-  host = gst_uri_get_host (uri);
-
-  priv->sock = gst_srt_server_listen (GST_ELEMENT (self),
-      FALSE, host, gst_uri_get_port (uri),
-      base->latency, &priv->poll_id, base->passphrase, base->key_length);
-
-  if (priv->sock == SRT_INVALID_SOCK) {
-    GST_ERROR_OBJECT (src, "Failed to create srt socket");
-    goto failed;
-  }
-
-  g_clear_pointer (&uri, gst_uri_unref);
-
-  return TRUE;
-
-failed:
-  if (priv->poll_id != SRT_ERROR) {
-    srt_epoll_release (priv->poll_id);
-    priv->poll_id = SRT_ERROR;
-  }
-
-  if (priv->sock != SRT_ERROR) {
-    srt_close (priv->sock);
-    priv->sock = SRT_ERROR;
-  }
-
-  g_clear_pointer (&uri, gst_uri_unref);
-
-  return FALSE;
-}
-
-static gboolean
-gst_srt_server_src_stop (GstBaseSrc * src)
-{
-  GstSRTServerSrc *self = GST_SRT_SERVER_SRC (src);
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-
-  if (priv->client_sock != SRT_INVALID_SOCK) {
-    g_signal_emit (self, signals[SIG_CLIENT_CLOSED], 0,
-        priv->client_sock, priv->client_sockaddr);
-    srt_close (priv->client_sock);
-    g_clear_object (&priv->client_sockaddr);
-    priv->client_sock = SRT_INVALID_SOCK;
-    priv->has_client = FALSE;
-  }
-
-  if (priv->poll_id != SRT_ERROR) {
-    srt_epoll_remove_usock (priv->poll_id, priv->sock);
-    srt_epoll_release (priv->poll_id);
-    priv->poll_id = SRT_ERROR;
-  }
-
-  if (priv->sock != SRT_INVALID_SOCK) {
-    GST_DEBUG_OBJECT (self, "closing SRT connection");
-    srt_close (priv->sock);
-    priv->sock = SRT_INVALID_SOCK;
-  }
-
-  priv->cancelled = FALSE;
-
-  return TRUE;
-}
-
-static gboolean
-gst_srt_server_src_unlock (GstBaseSrc * src)
-{
-  GstSRTServerSrc *self = GST_SRT_SERVER_SRC (src);
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-
-  priv->cancelled = TRUE;
-
-  return TRUE;
-}
-
-static gboolean
-gst_srt_server_src_unlock_stop (GstBaseSrc * src)
-{
-  GstSRTServerSrc *self = GST_SRT_SERVER_SRC (src);
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-
-  priv->cancelled = FALSE;
-
-  return TRUE;
-}
-
-static void
-gst_srt_server_src_class_init (GstSRTServerSrcClass * klass)
-{
-  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
-  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
-  GstBaseSrcClass *gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
-  GstPushSrcClass *gstpushsrc_class = GST_PUSH_SRC_CLASS (klass);
-
-  gobject_class->set_property = gst_srt_server_src_set_property;
-  gobject_class->get_property = gst_srt_server_src_get_property;
-  gobject_class->finalize = gst_srt_server_src_finalize;
-
-  /**
-   * GstSRTServerSrc:poll-timeout:
-   *
-   * The timeout(ms) value when polling SRT socket. For #GstSRTServerSrc,
-   * this value shouldn't be set as -1 (infinite) because "srt_epoll_wait"
-   * isn't cancellable unless closing the socket.
-   */
-  properties[PROP_POLL_TIMEOUT] =
-      g_param_spec_int ("poll-timeout", "Poll timeout",
-      "Return poll wait after timeout miliseconds", 0, G_MAXINT32,
-      SRT_DEFAULT_POLL_TIMEOUT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
-
-  g_object_class_install_properties (gobject_class, PROP_LAST, properties);
-
-  /**
-   * GstSRTServerSrc::client-added:
-   * @gstsrtserversrc: the srtserversrc element that emitted this signal
-   * @sock: the client socket descriptor that was added to srtserversrc
-   * @addr: the pointer of "struct sockaddr" that describes the @sock
-   * @addr_len: the length of @addr
-   *
-   * The given socket descriptor was added to srtserversrc.
-   */
-  signals[SIG_CLIENT_ADDED] =
-      g_signal_new ("client-added", G_TYPE_FROM_CLASS (klass),
-      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTServerSrcClass, client_added),
-      NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
-      2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
-
-  /**
-   * GstSRTServerSrc::client-closed:
-   * @gstsrtserversrc: the srtserversrc element that emitted this signal
-   * @sock: the client socket descriptor that was added to srtserversrc
-   * @addr: the pointer of "struct sockaddr" that describes the @sock
-   * @addr_len: the length of @addr
-   *
-   * The given socket descriptor was closed.
-   */
-  signals[SIG_CLIENT_CLOSED] =
-      g_signal_new ("client-closed", G_TYPE_FROM_CLASS (klass),
-      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTServerSrcClass, client_closed),
-      NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
-      2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
-
-  gst_element_class_add_static_pad_template (gstelement_class, &src_template);
-  gst_element_class_set_metadata (gstelement_class,
-      "SRT Server source", "Source/Network",
-      "Receive data over the network via SRT",
-      "Justin Kim <justin.kim@collabora.com>");
-
-  gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_srt_server_src_start);
-  gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_srt_server_src_stop);
-  gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_srt_server_src_unlock);
-  gstbasesrc_class->unlock_stop =
-      GST_DEBUG_FUNCPTR (gst_srt_server_src_unlock_stop);
-
-  gstpushsrc_class->fill = GST_DEBUG_FUNCPTR (gst_srt_server_src_fill);
-}
-
-static void
-gst_srt_server_src_init (GstSRTServerSrc * self)
-{
-  GstSRTServerSrcPrivate *priv = GST_SRT_SERVER_SRC_GET_PRIVATE (self);
-
-  priv->sock = SRT_INVALID_SOCK;
-  priv->client_sock = SRT_INVALID_SOCK;
-  priv->poll_id = SRT_ERROR;
-  priv->poll_timeout = SRT_DEFAULT_POLL_TIMEOUT;
-}
diff --git a/ext/srt/gstsrtserversrc.h b/ext/srt/gstsrtserversrc.h
deleted file mode 100644
index e8a3f554b..000000000
--- a/ext/srt/gstsrtserversrc.h
+++ /dev/null
@@ -1,64 +0,0 @@
-/* GStreamer
- * Copyright (C) 2017, Collabora Ltd.
- *   Author:Justin Kim <justin.kim@collabora.com>
- *
- * This library is free software; you can redistribute it and/or
- * modify it under the terms of the GNU Library General Public
- * License as published by the Free Software Foundation; either
- * version 2 of the License, or (at your option) any later version.
- *
- * This library is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
- * Library General Public License for more details.
- *
- * You should have received a copy of the GNU Library General Public
- * License along with this library; if not, write to the
- * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
- * Boston, MA 02110-1301, USA.
- */
-
-#ifndef __GST_SRT_SERVER_SRC_H__
-#define __GST_SRT_SERVER_SRC_H__
-
-#include "gstsrtbasesrc.h"
-#include <gio/gio.h>
-#include <srt/srt.h>
-
-G_BEGIN_DECLS
-
-#define GST_TYPE_SRT_SERVER_SRC              (gst_srt_server_src_get_type ())
-#define GST_IS_SRT_SERVER_SRC(obj)           (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SRT_SERVER_SRC))
-#define GST_IS_SRT_SERVER_SRC_CLASS(klass)   (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SRT_SERVER_SRC))
-#define GST_SRT_SERVER_SRC_GET_CLASS(obj)    (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_SRT_SERVER_SRC, GstSRTServerSrcClass))
-#define GST_SRT_SERVER_SRC(obj)              (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SRT_SERVER_SRC, GstSRTServerSrc))
-#define GST_SRT_SERVER_SRC_CLASS(klass)      (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SRT_SERVER_SRC, GstSRTServerSrcClass))
-#define GST_SRT_SERVER_SRC_CAST(obj)         ((GstSRTServerSrc*)(obj))
-#define GST_SRT_SERVER_SRC_CLASS_CAST(klass) ((GstSRTServerSrcClass*)(klass))
-
-typedef struct _GstSRTServerSrc GstSRTServerSrc;
-typedef struct _GstSRTServerSrcClass GstSRTServerSrcClass;
-typedef struct _GstSRTServerSrcPrivate GstSRTServerSrcPrivate;
-
-struct _GstSRTServerSrc {
-  GstSRTBaseSrc parent;
-
-  /*< private >*/
-  gpointer _gst_reserved[GST_PADDING];
-};
-
-struct _GstSRTServerSrcClass {
-  GstSRTBaseSrcClass parent_class;
-
-  void (*client_added)      (GstSRTServerSrc *self, int sock, GSocketAddress *addr);
-  void (*client_closed)     (GstSRTServerSrc *self, int sock, GSocketAddress *addr);
-
-  gpointer _gst_reserved[GST_PADDING_LARGE];
-};
-
-GST_EXPORT
-GType gst_srt_server_src_get_type (void);
-
-G_END_DECLS
-
-#endif /* __GST_SRT_SERVER_SRC_H__ */
diff --git a/ext/srt/gstsrtsink.c b/ext/srt/gstsrtsink.c
new file mode 100644
index 000000000..14a4452b5
--- /dev/null
+++ b/ext/srt/gstsrtsink.c
@@ -0,0 +1,391 @@
+/* GStreamer
+ * Copyright (C) 2018, Collabora Ltd.
+ * Copyright (C) 2018, SK Telecom, Co., Ltd.
+ *   Author: Jeongseok Kim <jeongseok.kim@sk.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+/**
+ * SECTION:element-srtsink
+ * @title: srtsink
+ *
+ * srtsink is a network sink that sends <ulink url="http://www.srtalliance.org/">SRT</ulink>
+ * packets to the network.
+ *
+ * <refsect2>
+ * <title>Examples</title>
+ * |[
+ * gst-launch-1.0 -v audiotestsrc ! srtsink uri://host?mode=caller
+ * ]| This pipeline shows how to serve SRT packets through the default port.
+ *
+ * |[
+ * gst-launch-1.0 -v audiotestsrc ! srtsink uri://host:port?mode=listener
+ * ]| This pipeline shows how to wait SRT callers.
+ * </refsect2>
+ * 
+ */
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include "gstsrtsink.h"
+
+static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
+    GST_PAD_SINK,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY);
+
+#define GST_CAT_DEFAULT gst_debug_srt_sink
+GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
+
+enum
+{
+  SIG_CALLER_ADDED,
+  SIG_CALLER_REMOVED,
+
+  LAST_SIGNAL
+};
+
+static guint signals[LAST_SIGNAL] = { 0 };
+
+static void gst_srt_sink_uri_handler_init (gpointer g_iface,
+    gpointer iface_data);
+static gchar *gst_srt_sink_uri_get_uri (GstURIHandler * handler);
+static gboolean gst_srt_sink_uri_set_uri (GstURIHandler * handler,
+    const gchar * uri, GError ** error);
+
+#define gst_srt_sink_parent_class parent_class
+G_DEFINE_TYPE_WITH_CODE (GstSRTSink, gst_srt_sink,
+    GST_TYPE_BASE_SINK,
+    G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_srt_sink_uri_handler_init)
+    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtsink", 0, "SRT Sink"));
+
+static void
+gst_srt_sink_set_property (GObject * object,
+    guint prop_id, const GValue * value, GParamSpec * pspec)
+{
+  GstSRTSink *self = GST_SRT_SINK (object);
+
+  if (!gst_srt_object_set_property_helper (self->srtobject, prop_id, value,
+          pspec)) {
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+  }
+}
+
+static void
+gst_srt_sink_get_property (GObject * object,
+    guint prop_id, GValue * value, GParamSpec * pspec)
+{
+  GstSRTSink *self = GST_SRT_SINK (object);
+
+  if (!gst_srt_object_get_property_helper (self->srtobject, prop_id, value,
+          pspec)) {
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+  }
+}
+
+static void
+gst_srt_sink_finalize (GObject * object)
+{
+  GstSRTSink *self = GST_SRT_SINK (object);
+
+  g_clear_object (&self->cancellable);
+  gst_srt_object_destroy (self->srtobject);
+
+  G_OBJECT_CLASS (parent_class)->finalize (object);
+}
+
+static void
+gst_srt_sink_init (GstSRTSink * self)
+{
+  self->srtobject = gst_srt_object_new (GST_ELEMENT (self));
+  self->cancellable = g_cancellable_new ();
+
+  gst_srt_object_set_uri (self->srtobject, GST_SRT_DEFAULT_URI, NULL);
+}
+
+static void
+gst_srt_sink_caller_added_cb (int sock, GSocketAddress * addr,
+    GstSRTObject * srtobject)
+{
+  g_signal_emit (srtobject->element, signals[SIG_CALLER_ADDED], 0, sock, addr);
+}
+
+static void
+gst_srt_sink_caller_removed_cb (int sock, GSocketAddress * addr,
+    GstSRTObject * srtobject)
+{
+  g_signal_emit (srtobject->element, signals[SIG_CALLER_REMOVED], 0, sock,
+      addr);
+}
+
+static gboolean
+gst_srt_sink_start (GstBaseSink * bsink)
+{
+  GstSRTSink *self = GST_SRT_SINK (bsink);
+  GstSRTConnectionMode connection_mode = GST_SRT_CONNECTION_MODE_NONE;
+
+  GError *error = NULL;
+  gboolean ret = FALSE;
+
+  gst_structure_get_enum (self->srtobject->parameters, "mode",
+      GST_TYPE_SRT_CONNECTION_MODE, (gint *) & connection_mode);
+
+  if (connection_mode == GST_SRT_CONNECTION_MODE_LISTENER) {
+    ret =
+        gst_srt_object_open_full (self->srtobject, gst_srt_sink_caller_added_cb,
+        gst_srt_sink_caller_removed_cb, self->cancellable, &error);
+  } else {
+    ret = gst_srt_object_open (self->srtobject, self->cancellable, &error);
+  }
+
+  if (!ret) {
+    GST_WARNING_OBJECT (self, "Failed to open SRT: %s", error->message);
+    g_clear_error (&error);
+  }
+
+  return ret;
+}
+
+static gboolean
+gst_srt_sink_stop (GstBaseSink * bsink)
+{
+  GstSRTSink *self = GST_SRT_SINK (bsink);
+
+  gst_srt_object_close (self->srtobject);
+
+  return TRUE;
+}
+
+static GstFlowReturn
+gst_srt_sink_render (GstBaseSink * sink, GstBuffer * buffer)
+{
+  GstSRTSink *self = GST_SRT_SINK (sink);
+  GstFlowReturn ret = GST_FLOW_OK;
+  GstMapInfo info;
+  GError *error = NULL;
+
+  if (g_cancellable_is_cancelled (self->cancellable)) {
+    ret = GST_FLOW_FLUSHING;
+  }
+
+  if (self->headers && GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_HEADER)) {
+    GST_DEBUG_OBJECT (self, "Have streamheaders,"
+        " ignoring header %" GST_PTR_FORMAT, buffer);
+    return GST_FLOW_OK;
+  }
+
+  if (!gst_buffer_map (buffer, &info, GST_MAP_READ)) {
+    GST_ELEMENT_ERROR (self, RESOURCE, READ,
+        ("Could not map the input stream"), (NULL));
+    return GST_FLOW_ERROR;
+  }
+
+  if (gst_srt_object_write (self->srtobject, self->headers, &info,
+          self->cancellable, &error) < 0) {
+    ret = GST_FLOW_ERROR;
+  }
+
+  gst_buffer_unmap (buffer, &info);
+
+  GST_TRACE_OBJECT (self, "sending buffer %p, offset %"
+      G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT
+      ", timestamp %" GST_TIME_FORMAT ", duration %" GST_TIME_FORMAT
+      ", size %" G_GSIZE_FORMAT,
+      buffer, GST_BUFFER_OFFSET (buffer),
+      GST_BUFFER_OFFSET_END (buffer),
+      GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buffer)),
+      GST_TIME_ARGS (GST_BUFFER_DURATION (buffer)),
+      gst_buffer_get_size (buffer));
+
+  return ret;
+}
+
+static gboolean
+gst_srt_sink_unlock (GstBaseSink * bsink)
+{
+  GstSRTSink *self = GST_SRT_SINK (bsink);
+
+  g_cancellable_cancel (self->cancellable);
+  gst_srt_object_wakeup (self->srtobject);
+
+  return TRUE;
+}
+
+static gboolean
+gst_srt_sink_unlock_stop (GstBaseSink * bsink)
+{
+  GstSRTSink *self = GST_SRT_SINK (bsink);
+
+  g_cancellable_reset (self->cancellable);
+
+  return TRUE;
+}
+
+static gboolean
+gst_srt_sink_set_caps (GstBaseSink * bsink, GstCaps * caps)
+{
+  GstSRTSink *self = GST_SRT_SINK (bsink);
+  GstStructure *s;
+  const GValue *streamheader;
+
+  GST_DEBUG_OBJECT (self, "setcaps %" GST_PTR_FORMAT, caps);
+
+  g_clear_pointer (&self->headers, gst_buffer_list_unref);
+
+  s = gst_caps_get_structure (caps, 0);
+  streamheader = gst_structure_get_value (s, "streamheader");
+
+  if (!streamheader) {
+    GST_DEBUG_OBJECT (self, "'streamheader' field not present");
+  } else if (GST_VALUE_HOLDS_BUFFER (streamheader)) {
+    GST_DEBUG_OBJECT (self, "'streamheader' field holds buffer");
+    self->headers = gst_buffer_list_new_sized (1);
+    gst_buffer_list_add (self->headers, g_value_dup_boxed (streamheader));
+  } else if (GST_VALUE_HOLDS_ARRAY (streamheader)) {
+    guint i, size;
+
+    GST_DEBUG_OBJECT (self, "'streamheader' field holds array");
+
+    size = gst_value_array_get_size (streamheader);
+    self->headers = gst_buffer_list_new_sized (size);
+
+    for (i = 0; i < size; i++) {
+      const GValue *v = gst_value_array_get_value (streamheader, i);
+      if (!GST_VALUE_HOLDS_BUFFER (v)) {
+        GST_ERROR_OBJECT (self, "'streamheader' item of unexpected type '%s'",
+            G_VALUE_TYPE_NAME (v));
+        return FALSE;
+      }
+
+      gst_buffer_list_add (self->headers, g_value_dup_boxed (v));
+    }
+  } else {
+    GST_ERROR_OBJECT (self, "'streamheader' field has unexpected type '%s'",
+        G_VALUE_TYPE_NAME (streamheader));
+    return FALSE;
+  }
+
+  GST_DEBUG_OBJECT (self, "Collected streamheaders: %u buffers",
+      self->headers ? gst_buffer_list_length (self->headers) : 0);
+
+  return TRUE;
+}
+
+static void
+gst_srt_sink_class_init (GstSRTSinkClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
+  GstBaseSinkClass *gstbasesink_class = GST_BASE_SINK_CLASS (klass);
+
+  gobject_class->set_property = gst_srt_sink_set_property;
+  gobject_class->get_property = gst_srt_sink_get_property;
+  gobject_class->finalize = gst_srt_sink_finalize;
+
+  /**
+   * GstSRTSink::caller-added:
+   * @gstsrtsink: the srtsink element that emitted this signal
+   * @sock: the client socket descriptor that was added to srtsink
+   * @addr: the #GSocketAddress that describes the @sock
+   * 
+   * The given socket descriptor was added to srtsink.
+   */
+  signals[SIG_CALLER_ADDED] =
+      g_signal_new ("caller-added", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTSinkClass, caller_added),
+      NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
+      2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
+
+  /**
+   * GstSRTSink::caller-removed:
+   * @gstsrtsink: the srtsink element that emitted this signal
+   * @sock: the client socket descriptor that was added to srtsink
+   * @addr: the #GSocketAddress that describes the @sock
+   *
+   * The given socket descriptor was removed from srtsink.
+   */
+  signals[SIG_CALLER_REMOVED] =
+      g_signal_new ("caller-removed", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTSinkClass,
+          caller_added), NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
+      2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
+
+  gst_srt_object_install_properties_helper (gobject_class);
+
+  gst_element_class_add_static_pad_template (gstelement_class, &sink_template);
+  gst_element_class_set_metadata (gstelement_class,
+      "SRT sink", "Sink/Network",
+      "Send data over the network via SRT",
+      "Justin Kim <justin.joy.9to5@gmail.com>");
+
+  gstbasesink_class->start = GST_DEBUG_FUNCPTR (gst_srt_sink_start);
+  gstbasesink_class->stop = GST_DEBUG_FUNCPTR (gst_srt_sink_stop);
+  gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_srt_sink_render);
+  gstbasesink_class->unlock = GST_DEBUG_FUNCPTR (gst_srt_sink_unlock);
+  gstbasesink_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_srt_sink_unlock_stop);
+  gstbasesink_class->set_caps = GST_DEBUG_FUNCPTR (gst_srt_sink_set_caps);
+
+}
+
+static GstURIType
+gst_srt_sink_uri_get_type (GType type)
+{
+  return GST_URI_SINK;
+}
+
+static const gchar *const *
+gst_srt_sink_uri_get_protocols (GType type)
+{
+  static const gchar *protocols[] = { GST_SRT_DEFAULT_URI_SCHEME, NULL };
+
+  return protocols;
+}
+
+static gchar *
+gst_srt_sink_uri_get_uri (GstURIHandler * handler)
+{
+  gchar *uri_str;
+  GstSRTSink *self = GST_SRT_SINK (handler);
+
+  GST_OBJECT_LOCK (self);
+  uri_str = gst_uri_to_string (self->srtobject->uri);
+  GST_OBJECT_UNLOCK (self);
+
+  return uri_str;
+}
+
+static gboolean
+gst_srt_sink_uri_set_uri (GstURIHandler * handler,
+    const gchar * uri, GError ** error)
+{
+  GstSRTSink *self = GST_SRT_SINK (handler);
+
+  return gst_srt_object_set_uri (self->srtobject, uri, error);
+}
+
+static void
+gst_srt_sink_uri_handler_init (gpointer g_iface, gpointer iface_data)
+{
+  GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
+
+  iface->get_type = gst_srt_sink_uri_get_type;
+  iface->get_protocols = gst_srt_sink_uri_get_protocols;
+  iface->get_uri = gst_srt_sink_uri_get_uri;
+  iface->set_uri = gst_srt_sink_uri_set_uri;
+}
diff --git a/ext/srt/gstsrtsink.h b/ext/srt/gstsrtsink.h
new file mode 100644
index 000000000..d39faf4e4
--- /dev/null
+++ b/ext/srt/gstsrtsink.h
@@ -0,0 +1,65 @@
+/* GStreamer
+ * Copyright (C) 2018, Collabora Ltd.
+ * Copyright (C) 2018, SK Telecom, Co., Ltd.
+ *   Author: Jeongseok Kim <jeongseok.kim@sk.com>
+ * 
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __GST_SRT_SINK_H__
+#define __GST_SRT_SINK_H__
+
+#include <gst/gst.h>
+#include <gst/base/gstbasesink.h>
+#include <gio/gio.h>
+
+#include "gstsrtobject.h"
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_SRT_SINK              (gst_srt_sink_get_type ())
+#define GST_IS_SRT_SINK(obj)           (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SRT_SINK))
+#define GST_IS_SRT_SINK_CLASS(klass)   (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SRT_SINK))
+#define GST_SRT_SINK_GET_CLASS(obj)    (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_SRT_SINK, GstSRTSinkClass))
+#define GST_SRT_SINK(obj)              (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SRT_SINK, GstSRTSink))
+#define GST_SRT_SINK_CLASS(klass)      (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SRT_SINK, GstSRTSinkClass))
+#define GST_SRT_SINK_CAST(obj)         ((GstSRTSink*)(obj))
+#define GST_SRT_SINK_CLASS_CAST(klass) ((GstSRTSinkClass*)(klass))
+
+typedef struct _GstSRTSink GstSRTSink;
+typedef struct _GstSRTSinkClass GstSRTSinkClass;
+
+struct _GstSRTSink {
+  GstBaseSink parent;
+
+  GstBufferList *headers;
+
+  GstSRTObject *srtobject;
+  GCancellable *cancellable;
+};
+
+struct _GstSRTSinkClass {
+  GstBaseSinkClass parent_class;
+
+  void (*caller_added)      (GstSRTSink *self, int sock, GSocketAddress * addr);
+  void (*caller_removed)    (GstSRTSink *self, int sock, GSocketAddress * addr);
+};
+
+GType gst_srt_sink_get_type (void);
+
+G_END_DECLS
+
+#endif // __GST_SRT_SINK_H__
diff --git a/ext/srt/gstsrtsrc.c b/ext/srt/gstsrtsrc.c
new file mode 100644
index 000000000..3931bfcab
--- /dev/null
+++ b/ext/srt/gstsrtsrc.c
@@ -0,0 +1,360 @@
+/* GStreamer
+ * Copyright (C) 2018, Collabora Ltd.
+ * Copyright (C) 2018, SK Telecom, Co., Ltd.
+ *   Author: Jeongseok Kim <jeongseok.kim@sk.com>
+ * 
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+/**
+ * SECTION:element-srtsrc
+ * @title: srtsrc
+ *
+ * srtsrc is a network source that reads <ulink url="http://www.srtalliance.org/">SRT</ulink>
+ * packets from the network.
+ *
+ * <refsect2>
+ * <title>Examples</title>
+ * |[
+ * gst-launch-1.0 -v srtsrc uri="srt://127.0.0.1:7001" ! fakesink
+ * ]| This pipeline shows how to connect SRT server by setting #GstSRTSrc:uri property.
+ *
+ * |[
+ * gst-launch-1.0 -v srtsrc uri="srt://127.0.0.1:7001?mode=listener" ! fakesink
+ * ]| This pipeline shows how to wait SRT connection by setting #GstSRTSrc:uri property.
+ *
+ * |[
+ * gst-launch-1.0 -v srtclientsrc uri="srt://192.168.1.10:7001?mode=rendez-vous" ! fakesink
+ * ]| This pipeline shows how to connect SRT server by setting #GstSRTSrc:uri property and using the rendez-vous mode.
+ * </refsect2>
+ *
+ */
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include "gstsrtsrc.h"
+
+static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
+    GST_PAD_SRC,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS_ANY);
+
+#define GST_CAT_DEFAULT gst_debug_srt_src
+GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
+
+enum
+{
+  SIG_CALLER_ADDED,
+  SIG_CALLER_REMOVED,
+
+  LAST_SIGNAL
+};
+
+static guint signals[LAST_SIGNAL] = { 0 };
+
+static void gst_srt_src_uri_handler_init (gpointer g_iface,
+    gpointer iface_data);
+static gchar *gst_srt_src_uri_get_uri (GstURIHandler * handler);
+static gboolean gst_srt_src_uri_set_uri (GstURIHandler * handler,
+    const gchar * uri, GError ** error);
+
+#define gst_srt_src_parent_class parent_class
+G_DEFINE_TYPE_WITH_CODE (GstSRTSrc, gst_srt_src,
+    GST_TYPE_PUSH_SRC,
+    G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_srt_src_uri_handler_init)
+    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtsrc", 0, "SRT Source"));
+
+static void
+gst_srt_src_caller_added_cb (int sock, GSocketAddress * addr,
+    GstSRTObject * srtobject)
+{
+  g_signal_emit (srtobject->element, signals[SIG_CALLER_ADDED], 0, sock, addr);
+}
+
+static void
+gst_srt_src_caller_removed_cb (int sock, GSocketAddress * addr,
+    GstSRTObject * srtobject)
+{
+  g_signal_emit (srtobject->element, signals[SIG_CALLER_REMOVED], 0, sock,
+      addr);
+}
+
+static gboolean
+gst_srt_src_start (GstBaseSrc * bsrc)
+{
+  GstSRTSrc *self = GST_SRT_SRC (bsrc);
+  GError *error = NULL;
+  gboolean ret = FALSE;
+  GstSRTConnectionMode connection_mode = GST_SRT_CONNECTION_MODE_NONE;
+
+  gst_structure_get_enum (self->srtobject->parameters, "mode",
+      GST_TYPE_SRT_CONNECTION_MODE, (gint *) & connection_mode);
+
+  if (connection_mode == GST_SRT_CONNECTION_MODE_LISTENER) {
+    ret =
+        gst_srt_object_open_full (self->srtobject, gst_srt_src_caller_added_cb,
+        gst_srt_src_caller_removed_cb, self->cancellable, &error);
+  } else {
+    ret = gst_srt_object_open (self->srtobject, self->cancellable, &error);
+  }
+
+  if (!ret) {
+    GST_WARNING_OBJECT (self, "Failed to open SRT: %s", error->message);
+    g_clear_error (&error);
+  }
+
+  return ret;
+}
+
+static gboolean
+gst_srt_src_stop (GstBaseSrc * bsrc)
+{
+  GstSRTSrc *self = GST_SRT_SRC (bsrc);
+
+  gst_srt_object_close (self->srtobject);
+
+  return TRUE;
+}
+
+static GstFlowReturn
+gst_srt_src_fill (GstPushSrc * src, GstBuffer * outbuf)
+{
+  GstSRTSrc *self = GST_SRT_SRC (src);
+  GstFlowReturn ret = GST_FLOW_OK;
+  GstMapInfo info;
+  GError *err = NULL;
+  gssize recv_len;
+
+  if (g_cancellable_is_cancelled (self->cancellable)) {
+    ret = GST_FLOW_FLUSHING;
+  }
+
+  if (!gst_buffer_map (outbuf, &info, GST_MAP_WRITE)) {
+    GST_ELEMENT_ERROR (src, RESOURCE, READ,
+        ("Could not map the buffer for writing "), (NULL));
+    ret = GST_FLOW_ERROR;
+    goto out;
+  }
+
+  recv_len = gst_srt_object_read (self->srtobject, info.data,
+      gst_buffer_get_size (outbuf), self->cancellable, &err);
+
+  gst_buffer_unmap (outbuf, &info);
+
+  if (g_cancellable_is_cancelled (self->cancellable)) {
+    ret = GST_FLOW_FLUSHING;
+    goto out;
+  }
+
+  if (recv_len < 0) {
+    GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), ("%s", err->message));
+    ret = GST_FLOW_ERROR;
+    g_clear_error (&err);
+    goto out;
+  } else if (recv_len == 0) {
+    ret = GST_FLOW_EOS;
+    goto out;
+  }
+
+  gst_buffer_resize (outbuf, 0, recv_len);
+
+  GST_LOG_OBJECT (src,
+      "filled buffer from _get of size %" G_GSIZE_FORMAT ", ts %"
+      GST_TIME_FORMAT ", dur %" GST_TIME_FORMAT
+      ", offset %" G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT,
+      gst_buffer_get_size (outbuf),
+      GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (outbuf)),
+      GST_TIME_ARGS (GST_BUFFER_DURATION (outbuf)),
+      GST_BUFFER_OFFSET (outbuf), GST_BUFFER_OFFSET_END (outbuf));
+
+out:
+  return ret;
+}
+
+static void
+gst_srt_src_init (GstSRTSrc * self)
+{
+  self->srtobject = gst_srt_object_new (GST_ELEMENT (self));
+  self->cancellable = g_cancellable_new ();
+
+  gst_base_src_set_format (GST_BASE_SRC (self), GST_FORMAT_TIME);
+  gst_base_src_set_live (GST_BASE_SRC (self), TRUE);
+  gst_base_src_set_do_timestamp (GST_BASE_SRC (self), TRUE);
+
+  gst_srt_object_set_uri (self->srtobject, GST_SRT_DEFAULT_URI, NULL);
+
+}
+
+static void
+gst_srt_src_finalize (GObject * object)
+{
+  GstSRTSrc *self = GST_SRT_SRC (object);
+
+  g_clear_object (&self->cancellable);
+  gst_srt_object_destroy (self->srtobject);
+
+  G_OBJECT_CLASS (parent_class)->finalize (object);
+}
+
+static gboolean
+gst_srt_src_unlock (GstBaseSrc * bsrc)
+{
+  GstSRTSrc *self = GST_SRT_SRC (bsrc);
+
+  g_cancellable_cancel (self->cancellable);
+  gst_srt_object_wakeup (self->srtobject);
+
+  return TRUE;
+}
+
+static gboolean
+gst_srt_src_unlock_stop (GstBaseSrc * bsrc)
+{
+  GstSRTSrc *self = GST_SRT_SRC (bsrc);
+
+  g_cancellable_reset (self->cancellable);
+
+  return TRUE;
+}
+
+static void
+gst_srt_src_set_property (GObject * object,
+    guint prop_id, const GValue * value, GParamSpec * pspec)
+{
+  GstSRTSrc *self = GST_SRT_SRC (object);
+
+  if (!gst_srt_object_set_property_helper (self->srtobject, prop_id, value,
+          pspec)) {
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+  }
+}
+
+static void
+gst_srt_src_get_property (GObject * object,
+    guint prop_id, GValue * value, GParamSpec * pspec)
+{
+  GstSRTSrc *self = GST_SRT_SRC (object);
+
+  if (!gst_srt_object_get_property_helper (self->srtobject, prop_id, value,
+          pspec)) {
+    G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+  }
+}
+
+static void
+gst_srt_src_class_init (GstSRTSrcClass * klass)
+{
+  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
+  GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
+  GstBaseSrcClass *gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
+  GstPushSrcClass *gstpushsrc_class = GST_PUSH_SRC_CLASS (klass);
+
+  gobject_class->set_property = gst_srt_src_set_property;
+  gobject_class->get_property = gst_srt_src_get_property;
+  gobject_class->finalize = gst_srt_src_finalize;
+
+  /**
+   * GstSRTSrc::caller-added:
+   * @gstsrtsink: the srtsink element that emitted this signal
+   * @sock: the client socket descriptor that was added to srtsink
+   * @addr: the #GSocketAddress that describes the @sock
+   * 
+   * The given socket descriptor was added to srtsink.
+   */
+  signals[SIG_CALLER_ADDED] =
+      g_signal_new ("caller-added", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTSrcClass, caller_added),
+      NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
+      2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
+
+  /**
+   * GstSRTSrc::caller-removed:
+   * @gstsrtsink: the srtsink element that emitted this signal
+   * @sock: the client socket descriptor that was added to srtsink
+   * @addr: the #GSocketAddress that describes the @sock
+   *
+   * The given socket descriptor was removed from srtsink.
+   */
+  signals[SIG_CALLER_REMOVED] =
+      g_signal_new ("caller-removed", G_TYPE_FROM_CLASS (klass),
+      G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTSrcClass,
+          caller_added), NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE,
+      2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
+
+  gst_srt_object_install_properties_helper (gobject_class);
+
+  gst_element_class_add_static_pad_template (gstelement_class, &src_template);
+  gst_element_class_set_metadata (gstelement_class,
+      "SRT source", "Source/Network",
+      "Receive data over the network via SRT",
+      "Justin Kim <justin.joy.9to5@gmail.com>");
+
+  gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_srt_src_start);
+  gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_srt_src_stop);
+  gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_srt_src_unlock);
+  gstbasesrc_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_srt_src_unlock_stop);
+
+  gstpushsrc_class->fill = GST_DEBUG_FUNCPTR (gst_srt_src_fill);
+}
+
+static GstURIType
+gst_srt_src_uri_get_type (GType type)
+{
+  return GST_URI_SRC;
+}
+
+static const gchar *const *
+gst_srt_src_uri_get_protocols (GType type)
+{
+  static const gchar *protocols[] = { GST_SRT_DEFAULT_URI_SCHEME, NULL };
+
+  return protocols;
+}
+
+static gchar *
+gst_srt_src_uri_get_uri (GstURIHandler * handler)
+{
+  gchar *uri_str;
+  GstSRTSrc *self = GST_SRT_SRC (handler);
+
+  GST_OBJECT_LOCK (self);
+  uri_str = gst_uri_to_string (self->srtobject->uri);
+  GST_OBJECT_UNLOCK (self);
+
+  return uri_str;
+}
+
+static gboolean
+gst_srt_src_uri_set_uri (GstURIHandler * handler,
+    const gchar * uri, GError ** error)
+{
+  GstSRTSrc *self = GST_SRT_SRC (handler);
+
+  return gst_srt_object_set_uri (self->srtobject, uri, error);
+}
+
+static void
+gst_srt_src_uri_handler_init (gpointer g_iface, gpointer iface_data)
+{
+  GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
+
+  iface->get_type = gst_srt_src_uri_get_type;
+  iface->get_protocols = gst_srt_src_uri_get_protocols;
+  iface->get_uri = gst_srt_src_uri_get_uri;
+  iface->set_uri = gst_srt_src_uri_set_uri;
+}
diff --git a/ext/srt/gstsrtsrc.h b/ext/srt/gstsrtsrc.h
new file mode 100644
index 000000000..057366e11
--- /dev/null
+++ b/ext/srt/gstsrtsrc.h
@@ -0,0 +1,65 @@
+/* GStreamer
+ * Copyright (C) 2018, Collabora Ltd.
+ * Copyright (C) 2018, SK Telecom, Co., Ltd.
+ *   Author: Jeongseok Kim <jeongseok.kim@sk.com>
+ * 
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#ifndef __GST_SRT_SRC_H__
+#define __GST_SRT_SRC_H__
+
+#include <gio/gio.h>
+#include <gst/gst.h>
+#include <gst/base/gstpushsrc.h>
+
+#include "gstsrtobject.h"
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_SRT_SRC              (gst_srt_src_get_type ())
+#define GST_IS_SRT_SRC(obj)           (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_TYPE_SRT_SRC))
+#define GST_IS_SRT_SRC_CLASS(klass)   (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_TYPE_SRT_SRC))
+#define GST_SRT_SRC_GET_CLASS(obj)    (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_TYPE_SRT_SRC, GstSRTSrcClass))
+#define GST_SRT_SRC(obj)              (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_TYPE_SRT_SRC, GstSRTSrc))
+#define GST_SRT_SRC_CLASS(klass)      (G_TYPE_CHECK_CLASS_CAST ((klass), GST_TYPE_SRT_SRC, GstSRTSrcClass))
+#define GST_SRT_SRC_CAST(obj)         ((GstSRTSrc*)(obj))
+#define GST_SRT_SRC_CLASS_CAST(klass) ((GstSRTSrcClass*)(klass))
+
+typedef struct _GstSRTSrc GstSRTSrc;
+typedef struct _GstSRTSrcClass GstSRTSrcClass;
+
+struct _GstSRTSrc {
+  GstPushSrc parent;
+
+  GstCaps      *caps;
+
+  GstSRTObject *srtobject;
+  GCancellable *cancellable;
+};
+
+struct _GstSRTSrcClass {
+  GstPushSrcClass parent_class;
+
+  void (*caller_added)      (GstSRTSrc *self, int sock, GSocketAddress * addr);
+  void (*caller_removed)    (GstSRTSrc *self, int sock, GSocketAddress * addr);
+};
+
+GType   gst_srt_src_get_type (void);
+
+G_END_DECLS
+
+#endif // __GST_SRT_SRC_H__
diff --git a/ext/srt/meson.build b/ext/srt/meson.build
index 0f328ccb9..0ad86db97 100644
--- a/ext/srt/meson.build
+++ b/ext/srt/meson.build
@@ -1,11 +1,8 @@
 srt_sources = [
   'gstsrt.c',
-  'gstsrtbasesrc.c',
-  'gstsrtclientsrc.c',
-  'gstsrtserversrc.c',
-  'gstsrtbasesink.c',
-  'gstsrtclientsink.c',
-  'gstsrtserversink.c',
+  'gstsrtobject.c',
+  'gstsrtsink.c',
+  'gstsrtsrc.c'
 ]
 srt_option = get_option('srt')
 if srt_option.disabled()
@@ -21,8 +18,13 @@ if not srt_dep.found() and srt_option.enabled()
 endif
 
 if srt_dep.found()
+  gstsrt_enums = gnome.mkenums_simple('gstsrt-enumtypes',
+    sources: ['gstsrt-enums.h'],
+    decorator : 'G_GNUC_INTERNAL',
+    install_header: false)
+
   gstsrt = library('gstsrt',
-    srt_sources,
+    srt_sources, gstsrt_enums,
     c_args : gst_plugins_bad_args,
     link_args : noseh_link_args,
     include_directories : [configinc, libsinc],