Commit 53a45b12 authored by Håvard Graff's avatar Håvard Graff Committed by Mathieu Duponchelle

Initial commit of GstRtpFunnel

For funneling together rtp-streams into a single session.
Use-cases include multiplexing and bundle.
parent afa4be4b
...@@ -14,7 +14,8 @@ libgstrtpmanager_la_SOURCES = gstrtpmanager.c \ ...@@ -14,7 +14,8 @@ libgstrtpmanager_la_SOURCES = gstrtpmanager.c \
rtpsession.c \ rtpsession.c \
rtpsource.c \ rtpsource.c \
rtpstats.c \ rtpstats.c \
gstrtpsession.c gstrtpsession.c \
gstrtpfunnel.c
noinst_HEADERS = gstrtpbin.h \ noinst_HEADERS = gstrtpbin.h \
gstrtpdtmfmux.h \ gstrtpdtmfmux.h \
...@@ -29,7 +30,9 @@ noinst_HEADERS = gstrtpbin.h \ ...@@ -29,7 +30,9 @@ noinst_HEADERS = gstrtpbin.h \
rtpsession.h \ rtpsession.h \
rtpsource.h \ rtpsource.h \
rtpstats.h \ rtpstats.h \
gstrtpsession.h gstrtpsession.h \
gstrtpfunnel.h
libgstrtpmanager_la_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(GST_CFLAGS) \ libgstrtpmanager_la_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(GST_CFLAGS) \
$(GST_NET_CFLAGS) $(WARNING_CFLAGS) $(ERROR_CFLAGS) $(GST_NET_CFLAGS) $(WARNING_CFLAGS) $(ERROR_CFLAGS)
......
This diff is collapsed.
/* RTP funnel element for GStreamer
*
* gstrtpfunnel.h:
*
* Copyright (C) <2017> Pexip.
* Contact: Havard Graff <havard@pexip.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_RTP_FUNNEL_H__
#define __GST_RTP_FUNNEL_H__
#include <gst/gst.h>
G_BEGIN_DECLS
G_DECLARE_FINAL_TYPE (GstRtpFunnel, gst_rtp_funnel, GST, RTP_FUNNEL, GstElement)
#define GST_TYPE_RTP_FUNNEL (gst_rtp_funnel_get_type())
#define GST_RTP_FUNNEL_CAST(obj) ((GstRtpFunnel *)(obj))
G_DECLARE_FINAL_TYPE (GstRtpFunnelPad, gst_rtp_funnel_pad, GST, RTP_FUNNEL_PAD, GstPad)
#define GST_TYPE_RTP_FUNNEL_PAD (gst_rtp_funnel_pad_get_type())
#define GST_RTP_FUNNEL_PAD_CAST(obj) ((GstRtpFunnelPad *)(obj))
G_END_DECLS
#endif /* __GST_RTP_FUNNEL_H__ */
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#include "gstrtpssrcdemux.h" #include "gstrtpssrcdemux.h"
#include "gstrtpdtmfmux.h" #include "gstrtpdtmfmux.h"
#include "gstrtpmux.h" #include "gstrtpmux.h"
#include "gstrtpfunnel.h"
static gboolean static gboolean
plugin_init (GstPlugin * plugin) plugin_init (GstPlugin * plugin)
...@@ -69,6 +70,10 @@ plugin_init (GstPlugin * plugin) ...@@ -69,6 +70,10 @@ plugin_init (GstPlugin * plugin)
if (!gst_rtp_dtmf_mux_plugin_init (plugin)) if (!gst_rtp_dtmf_mux_plugin_init (plugin))
return FALSE; return FALSE;
if (!gst_element_register (plugin, "rtpfunnel", GST_RANK_NONE,
GST_TYPE_RTP_FUNNEL))
return FALSE;
return TRUE; return TRUE;
} }
......
...@@ -14,6 +14,7 @@ rtpmanager_sources = [ ...@@ -14,6 +14,7 @@ rtpmanager_sources = [
'rtpsource.c', 'rtpsource.c',
'rtpstats.c', 'rtpstats.c',
'gstrtpsession.c', 'gstrtpsession.c',
'gstrtpfunnel.c',
] ]
gstrtpmanager = library('gstrtpmanager', gstrtpmanager = library('gstrtpmanager',
......
...@@ -254,7 +254,8 @@ check_rtpmanager = \ ...@@ -254,7 +254,8 @@ check_rtpmanager = \
elements/rtpsession \ elements/rtpsession \
elements/rtpstorage \ elements/rtpstorage \
elements/rtpred \ elements/rtpred \
elements/rtpulpfec elements/rtpulpfec \
elements/rtpfunnel
else else
check_rtpmanager = check_rtpmanager =
endif endif
...@@ -605,6 +606,9 @@ elements_rtprtx_LDADD = $(GST_PLUGINS_BASE_LIBS) -lgstrtp-$(GST_API_VERSION) $(L ...@@ -605,6 +606,9 @@ elements_rtprtx_LDADD = $(GST_PLUGINS_BASE_LIBS) -lgstrtp-$(GST_API_VERSION) $(L
elements_rtpsession_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(CFLAGS) $(AM_CFLAGS) elements_rtpsession_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(CFLAGS) $(AM_CFLAGS)
elements_rtpsession_LDADD = $(GST_PLUGINS_BASE_LIBS) $(GST_NET_LIBS) -lgstrtp-$(GST_API_VERSION) -lgstvideo-$(GST_API_VERSION) $(GIO_LIBS) $(LDADD) elements_rtpsession_LDADD = $(GST_PLUGINS_BASE_LIBS) $(GST_NET_LIBS) -lgstrtp-$(GST_API_VERSION) -lgstvideo-$(GST_API_VERSION) $(GIO_LIBS) $(LDADD)
elements_rtpfunnel_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(GST_BASE_CFLAGS) $(AM_CFLAGS)
elements_rtpfunnel_LDADD = $(GST_PLUGINS_BASE_LIBS) -lgstrtp-$(GST_API_VERSION) $(GST_BASE_LIBS) $(LDADD)
elements_rtpcollision_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(CFLAGS) $(AM_CFLAGS) elements_rtpcollision_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(CFLAGS) $(AM_CFLAGS)
elements_rtpcollision_LDADD = $(GST_PLUGINS_BASE_LIBS) $(GST_NET_LIBS) -lgstrtp-$(GST_API_VERSION) $(GIO_LIBS) $(LDADD) elements_rtpcollision_LDADD = $(GST_PLUGINS_BASE_LIBS) $(GST_NET_LIBS) -lgstrtp-$(GST_API_VERSION) $(GIO_LIBS) $(LDADD)
......
/* GStreamer
*
* unit test for rtpfunnel
*
* Copyright (C) <2017> Pexip.
* Contact: Havard Graff <havard@pexip.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#include <gst/check/gstcheck.h>
#include <gst/check/gstharness.h>
GST_START_TEST (rtpfunnel_ssrc_demuxing)
{
GstHarness *h = gst_harness_new_with_padnames ("rtpfunnel", NULL, "src");
GstHarness *h0 = gst_harness_new_with_element (h->element, "sink_0", NULL);
GstHarness *h1 = gst_harness_new_with_element (h->element, "sink_1", NULL);
gst_harness_set_src_caps_str (h0, "application/x-rtp, ssrc=(uint)123");
gst_harness_set_src_caps_str (h1, "application/x-rtp, ssrc=(uint)321");
/* unref latency events */
gst_event_unref (gst_harness_pull_upstream_event (h0));
gst_event_unref (gst_harness_pull_upstream_event (h1));
fail_unless_equals_int (1, gst_harness_upstream_events_received (h0));
fail_unless_equals_int (1, gst_harness_upstream_events_received (h1));
/* send to pad 0 */
gst_harness_push_upstream_event (h,
gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
gst_structure_new ("GstForceKeyUnit",
"ssrc", G_TYPE_UINT, 123, NULL)));
fail_unless_equals_int (2, gst_harness_upstream_events_received (h0));
fail_unless_equals_int (1, gst_harness_upstream_events_received (h1));
/* send to pad 1 */
gst_harness_push_upstream_event (h,
gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
gst_structure_new ("GstForceKeyUnit",
"ssrc", G_TYPE_UINT, 321, NULL)));
fail_unless_equals_int (2, gst_harness_upstream_events_received (h0));
fail_unless_equals_int (2, gst_harness_upstream_events_received (h1));
/* unknown ssrc, we drop it */
gst_harness_push_upstream_event (h,
gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
gst_structure_new ("GstForceKeyUnit",
"ssrc", G_TYPE_UINT, 666, NULL)));
fail_unless_equals_int (2, gst_harness_upstream_events_received (h0));
fail_unless_equals_int (2, gst_harness_upstream_events_received (h1));
/* no ssrc, we send to all */
gst_harness_push_upstream_event (h,
gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
gst_structure_new_empty ("GstForceKeyUnit")));
fail_unless_equals_int (3, gst_harness_upstream_events_received (h0));
fail_unless_equals_int (3, gst_harness_upstream_events_received (h1));
/* remove pad 0, and send an event referencing the now dead ssrc */
gst_harness_teardown (h0);
gst_harness_push_upstream_event (h,
gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
gst_structure_new ("GstForceKeyUnit",
"ssrc", G_TYPE_UINT, 123, NULL)));
fail_unless_equals_int (3, gst_harness_upstream_events_received (h1));
gst_harness_teardown (h);
gst_harness_teardown (h1);
}
GST_END_TEST
GST_START_TEST (rtpfunnel_ssrc_downstream_not_leaking_through)
{
GstHarness *h = gst_harness_new_with_padnames ("rtpfunnel",
"sink_0", "src");
GstCaps *caps;
const GstStructure *s;
gst_harness_set_sink_caps_str (h, "application/x-rtp, ssrc=(uint)123");
caps = gst_pad_peer_query_caps (h->srcpad, NULL);
s = gst_caps_get_structure (caps, 0);
fail_unless (!gst_structure_has_field (s, "ssrc"));
gst_caps_unref (caps);
gst_harness_teardown (h);
}
GST_END_TEST
GST_START_TEST (rtpfunnel_common_ts_offset)
{
GstHarness *h = gst_harness_new_with_padnames ("rtpfunnel",
"sink_0", "src");
GstCaps *caps;
const GstStructure *s;
const guint expected_ts_offset = 12345;
guint ts_offset;
g_object_set (h->element, "common-ts-offset", expected_ts_offset, NULL);
caps = gst_pad_peer_query_caps (h->srcpad, NULL);
s = gst_caps_get_structure (caps, 0);
fail_unless (gst_structure_get_uint (s, "timestamp-offset", &ts_offset));
fail_unless_equals_int (expected_ts_offset, ts_offset);
gst_caps_unref (caps);
gst_harness_teardown (h);
}
GST_END_TEST
GST_START_TEST (rtpfunnel_stress)
{
GstHarness *h = gst_harness_new_with_padnames ("rtpfunnel",
"sink_0", "src");
GstHarness *h1 = gst_harness_new_with_element (h->element, "sink_1", NULL);
GstPadTemplate *templ =
gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS (h->element),
"sink_%u");
GstCaps *caps = gst_caps_from_string ("application/x-rtp, ssrc=(uint)123");
GstBuffer *buf = gst_buffer_new_allocate (NULL, 0, NULL);
GstSegment segment;
GstHarnessThread *statechange, *push, *req, *push1;
gst_check_add_log_filter ("GStreamer", G_LOG_LEVEL_WARNING,
g_regex_new ("Got data flow before (stream-start|segment) event",
(GRegexCompileFlags) 0, (GRegexMatchFlags) 0, NULL),
NULL, NULL, NULL);
gst_check_add_log_filter ("GStreamer", G_LOG_LEVEL_WARNING,
g_regex_new ("Sticky event misordering",
(GRegexCompileFlags) 0, (GRegexMatchFlags) 0, NULL),
NULL, NULL, NULL);
gst_segment_init (&segment, GST_FORMAT_TIME);
statechange = gst_harness_stress_statechange_start (h);
push = gst_harness_stress_push_buffer_start (h, caps, &segment, buf);
req = gst_harness_stress_requestpad_start (h, templ, NULL, NULL, TRUE);
push1 = gst_harness_stress_push_buffer_start (h1, caps, &segment, buf);
gst_caps_unref (caps);
gst_buffer_unref (buf);
/* test-length */
g_usleep (G_USEC_PER_SEC * 1);
gst_harness_stress_thread_stop (push1);
gst_harness_stress_thread_stop (req);
gst_harness_stress_thread_stop (push);
gst_harness_stress_thread_stop (statechange);
gst_harness_teardown (h1);
gst_harness_teardown (h);
gst_check_clear_log_filter ();
}
GST_END_TEST;
static Suite *
rtpfunnel_suite (void)
{
Suite *s = suite_create ("rtpfunnel");
TCase *tc_chain = tcase_create ("general");
suite_add_tcase (s, tc_chain);
tcase_add_test (tc_chain, rtpfunnel_ssrc_demuxing);
tcase_add_test (tc_chain, rtpfunnel_ssrc_downstream_not_leaking_through);
tcase_add_test (tc_chain, rtpfunnel_common_ts_offset);
tcase_add_test (tc_chain, rtpfunnel_stress);
return s;
}
GST_CHECK_MAIN (rtpfunnel)
...@@ -75,6 +75,7 @@ good_tests = [ ...@@ -75,6 +75,7 @@ good_tests = [
[ 'elements/rtpbin_buffer_list' ], [ 'elements/rtpbin_buffer_list' ],
[ 'elements/rtpbundle' ], [ 'elements/rtpbundle' ],
[ 'elements/rtpcollision' ], [ 'elements/rtpcollision' ],
[ 'elements/rtpfunnel' ],
[ 'elements/rtpjitterbuffer' ], [ 'elements/rtpjitterbuffer' ],
[ 'elements/rtpmux' ], [ 'elements/rtpmux' ],
[ 'elements/rtprtx' ], [ 'elements/rtprtx' ],
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment