mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2025-01-11 01:45:33 +00:00
Code Issues 1 Projects Releases Wiki Activity

Add new SCTP plugins (sctpenc/sctpdec)

Browse source 
https://bugzilla.gnome.org/show_bug.cgi?id=744863
This commit is contained in:
George Kiagiadakis 2015-02-04 17:12:48 +01:00 committed by Matthew Waters
parent 45fe050286
commit e2f06326ea
19 changed files with 3158 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
configure.ac
View file

@ -2356,6 +2356,22 @@ AG_GST_CHECK_FEATURE(WEBRTC, [WebRTC], webrtc, [
])
])
dnl *** usrsctp ***
translit(dnm, m, l) AM_CONDITIONAL(USE_SCTP, true)
AG_GST_CHECK_FEATURE(SCTP, [sctp plug-in], sctp, [
AC_CHECK_HEADER(usrsctp.h, [
AC_CHECK_LIB(usrsctp, usrsctp_init, [
USRSCTP_LIBS="-lusrsctp"
AC_SUBST(USRSCTP_LIBS)
HAVE_SCTP=yes
], [
HAVE_SCTP=no
])
], [
HAVE_SCTP=no
])
])
else
dnl not building plugins with external dependencies,
@ -2440,6 +2456,7 @@ AM_CONDITIONAL(USE_X265, false)
AM_CONDITIONAL(USE_DTLS, false)
AM_CONDITIONAL(USE_VULKAN, false)
AM_CONDITIONAL(USE_TTML, false)
AM_CONDITIONAL(USE_SCTP, false)
fi dnl of EXT plugins
@ -2612,6 +2629,7 @@ gst-libs/gst/interfaces/Makefile
gst-libs/gst/isoff/Makefile
gst-libs/gst/codecparsers/Makefile
gst-libs/gst/mpegts/Makefile
gst-libs/gst/sctp/Makefile
gst-libs/gst/uridownloader/Makefile
gst-libs/gst/wayland/Makefile
gst-libs/gst/webrtc/Makefile
@ -2714,6 +2732,7 @@ ext/rsvg/Makefile
ext/resindvd/Makefile
ext/rtmp/Makefile
ext/sbc/Makefile
ext/sctp/Makefile
ext/smoothstreaming/Makefile
ext/sndfile/Makefile
ext/soundtouch/Makefile
@ -2750,6 +2769,8 @@ pkgconfig/gstreamer-mpegts.pc
pkgconfig/gstreamer-mpegts-uninstalled.pc
pkgconfig/gstreamer-player.pc
pkgconfig/gstreamer-player-uninstalled.pc
pkgconfig/gstreamer-sctp.pc
pkgconfig/gstreamer-sctp-uninstalled.pc
pkgconfig/gstreamer-wayland.pc
pkgconfig/gstreamer-wayland-uninstalled.pc
pkgconfig/gstreamer-webrtc.pc

8
ext/Makefile.am
View file

@ -286,6 +286,12 @@ else
SBC_DIR=
endif
if USE_SCTP
SCTP_DIR=sctp
else
SCTP_DIR=
endif
if USE_SMOOTHSTREAMING
SMOOTHSTREAMING_DIR = smoothstreaming
else
@ -455,6 +461,7 @@ SUBDIRS=\
$(OPUS_DIR) \
$(RSVG_DIR) \
$(SBC_DIR) \
$(SCTP_DIR) \
$(SMOOTHSTREAMING_DIR) \
$(SMOOTHWAVE_DIR) \
$(SNDFILE_DIR) \
@ -523,6 +530,7 @@ DIST_SUBDIRS = \
rsvg \
resindvd \
sbc \
sctp \
smoothstreaming \
sndfile \
soundtouch \

21
ext/sctp/Makefile.am Normal file
View file

@ -0,0 +1,21 @@
plugin_LTLIBRARIES = libgstsctp.la
libgstsctp_la_SOURCES = \
gstsctpplugin.c \
sctpassociation.c \
gstsctpenc.c \
gstsctpdec.c
libgstsctp_la_CFLAGS = \
$(GST_PLUGINS_BASE_CFLAGS) \
$(GST_BASE_CFLAGS) \
$(GST_CFLAGS) \
$(USRSCTP_CFLAGS) \
-I$(top_srcdir)/gst-libs
libgstsctp_la_LIBADD = $(GST_LIBS) $(GST_BASE_LIBS) $(USRSCTP_LIBS) $(top_builddir)/gst-libs/gst/sctp/libgstsctp-1.0.la
noinst_HEADERS = \
sctpassociation.h \
gstsctpenc.h \
gstsctpdec.h

655
ext/sctp/gstsctpdec.c Normal file
View file

@ -0,0 +1,655 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "gstsctpdec.h"
#include <gst/sctp/sctpreceivemeta.h>
#include <gst/base/gstdataqueue.h>
#include <stdio.h>
#include <stdlib.h>
GST_DEBUG_CATEGORY_STATIC (gst_sctp_dec_debug_category);
#define GST_CAT_DEFAULT gst_sctp_dec_debug_category
#define gst_sctp_dec_parent_class parent_class
G_DEFINE_TYPE (GstSctpDec, gst_sctp_dec, GST_TYPE_ELEMENT);
static GstStaticPadTemplate sink_template =
GST_STATIC_PAD_TEMPLATE ("sink", GST_PAD_SINK,
GST_PAD_ALWAYS, GST_STATIC_CAPS ("application/x-sctp"));
static GstStaticPadTemplate src_template =
GST_STATIC_PAD_TEMPLATE ("src_%u", GST_PAD_SRC,
GST_PAD_SOMETIMES, GST_STATIC_CAPS_ANY);
enum
{
SIGNAL_RESET_STREAM,
NUM_SIGNALS
};
static guint signals[NUM_SIGNALS];
enum
{
PROP_0,
PROP_GST_SCTP_ASSOCIATION_ID,
PROP_LOCAL_SCTP_PORT,
NUM_PROPERTIES
};
static GParamSpec *properties[NUM_PROPERTIES];
#define DEFAULT_GST_SCTP_ASSOCIATION_ID 1
#define DEFAULT_LOCAL_SCTP_PORT 0
#define MAX_SCTP_PORT 65535
#define MAX_GST_SCTP_ASSOCIATION_ID 65535
#define MAX_STREAM_ID 65535
GType gst_sctp_dec_pad_get_type (void);
#define GST_TYPE_SCTP_DEC_PAD (gst_sctp_dec_pad_get_type())
#define GST_SCTP_DEC_PAD(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj), GST_TYPE_SCTP_DEC_PAD, GstSctpDecPad))
#define GST_SCTP_DEC_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass), GST_TYPE_SCTP_DEC_PAD, GstSctpDecPadClass))
#define GST_IS_SCTP_DEC_PAD(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj), GST_TYPE_SCTP_DEC_PAD))
#define GST_IS_SCTP_DEC_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass), GST_TYPE_SCTP_DEC_PAD))
typedef struct _GstSctpDecPad GstSctpDecPad;
typedef GstPadClass GstSctpDecPadClass;
struct _GstSctpDecPad
{
GstPad parent;
GstDataQueue *packet_queue;
};
G_DEFINE_TYPE (GstSctpDecPad, gst_sctp_dec_pad, GST_TYPE_PAD);
static void
gst_sctp_dec_pad_finalize (GObject * object)
{
GstSctpDecPad *self = GST_SCTP_DEC_PAD (object);
gst_object_unref (self->packet_queue);
G_OBJECT_CLASS (gst_sctp_dec_pad_parent_class)->finalize (object);
}
static gboolean
data_queue_check_full_cb (GstDataQueue * queue, guint visible, guint bytes,
guint64 time, gpointer user_data)
{
/* FIXME: Are we full at some point and block? */
return FALSE;
}
static void
data_queue_empty_cb (GstDataQueue * queue, gpointer user_data)
{
}
static void
data_queue_full_cb (GstDataQueue * queue, gpointer user_data)
{
}
static void
gst_sctp_dec_pad_class_init (GstSctpDecPadClass * klass)
{
GObjectClass *gobject_class;
gobject_class = G_OBJECT_CLASS (klass);
gobject_class->finalize = gst_sctp_dec_pad_finalize;
}
static void
gst_sctp_dec_pad_init (GstSctpDecPad * self)
{
self->packet_queue = gst_data_queue_new (data_queue_check_full_cb,
data_queue_full_cb, data_queue_empty_cb, NULL);
}
static void gst_sctp_dec_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec);
static void gst_sctp_dec_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
static GstStateChangeReturn gst_sctp_dec_change_state (GstElement * element,
GstStateChange transition);
static GstFlowReturn gst_sctp_dec_packet_chain (GstPad * pad, GstSctpDec * self,
GstBuffer * buf);
static gboolean gst_sctp_dec_packet_event (GstPad * pad, GstSctpDec * self,
GstEvent * event);
static void gst_sctp_data_srcpad_loop (GstPad * pad);
static gboolean configure_association (GstSctpDec * self);
static void on_gst_sctp_association_stream_reset (GstSctpAssociation *
gst_sctp_association, guint16 stream_id, GstSctpDec * self);
static void on_receive (GstSctpAssociation * gst_sctp_association, guint8 * buf,
gsize length, guint16 stream_id, guint ppid, gpointer user_data);
static void stop_srcpad_task (GstPad * pad);
static void stop_all_srcpad_tasks (GstSctpDec * self);
static void sctpdec_cleanup (GstSctpDec * self);
static GstPad *get_pad_for_stream_id (GstSctpDec * self, guint16 stream_id);
static void remove_pad (GstElement * element, GstPad * pad);
static void on_reset_stream (GstSctpDec * self, guint stream_id);
static void
gst_sctp_dec_class_init (GstSctpDecClass * klass)
{
GObjectClass *gobject_class;
GstElementClass *element_class;
gobject_class = G_OBJECT_CLASS (klass);
element_class = GST_ELEMENT_CLASS (klass);
GST_DEBUG_CATEGORY_INIT (gst_sctp_dec_debug_category,
"sctpdec", 0, "debug category for sctpdec element");
gst_element_class_add_pad_template (element_class,
gst_static_pad_template_get (&src_template));
gst_element_class_add_pad_template (element_class,
gst_static_pad_template_get (&sink_template));
gobject_class->set_property = gst_sctp_dec_set_property;
gobject_class->get_property = gst_sctp_dec_get_property;
element_class->change_state = GST_DEBUG_FUNCPTR (gst_sctp_dec_change_state);
klass->on_reset_stream = on_reset_stream;
properties[PROP_GST_SCTP_ASSOCIATION_ID] =
g_param_spec_uint ("sctp-association-id",
"SCTP Association ID",
"Every encoder/decoder pair should have the same, unique, sctp-association-id. "
"This value must be set before any pads are requested.",
0, MAX_GST_SCTP_ASSOCIATION_ID, DEFAULT_GST_SCTP_ASSOCIATION_ID,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
properties[PROP_LOCAL_SCTP_PORT] =
g_param_spec_uint ("local-sctp-port",
"Local SCTP port",
"Local sctp port for the sctp association. The remote port is configured via the "
"GstSctpEnc element.",
0, MAX_SCTP_PORT, DEFAULT_LOCAL_SCTP_PORT,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
g_object_class_install_properties (gobject_class, NUM_PROPERTIES, properties);
signals[SIGNAL_RESET_STREAM] = g_signal_new ("reset-stream",
G_TYPE_FROM_CLASS (gobject_class), G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstSctpDecClass, on_reset_stream), NULL, NULL,
g_cclosure_marshal_generic, G_TYPE_NONE, 1, G_TYPE_UINT);
gst_element_class_set_static_metadata (element_class,
"SCTP Decoder",
"Decoder/Network/SCTP",
"Decodes packets with SCTP",
"George Kiagiadakis <george.kiagiadakis@collabora.com>");
}
static void
gst_sctp_dec_init (GstSctpDec * self)
{
self->sctp_association_id = DEFAULT_GST_SCTP_ASSOCIATION_ID;
self->local_sctp_port = DEFAULT_LOCAL_SCTP_PORT;
self->sink_pad = gst_pad_new_from_static_template (&sink_template, "sink");
gst_pad_set_chain_function (self->sink_pad,
GST_DEBUG_FUNCPTR ((GstPadChainFunction) gst_sctp_dec_packet_chain));
gst_pad_set_event_function (self->sink_pad,
GST_DEBUG_FUNCPTR ((GstPadEventFunction) gst_sctp_dec_packet_event));
gst_element_add_pad (GST_ELEMENT (self), self->sink_pad);
}
static void
gst_sctp_dec_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec)
{
GstSctpDec *self = GST_SCTP_DEC (object);
switch (prop_id) {
case PROP_GST_SCTP_ASSOCIATION_ID:
self->sctp_association_id = g_value_get_uint (value);
break;
case PROP_LOCAL_SCTP_PORT:
self->local_sctp_port = g_value_get_uint (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (self, prop_id, pspec);
break;
}
}
static void
gst_sctp_dec_get_property (GObject * object, guint prop_id, GValue * value,
GParamSpec * pspec)
{
GstSctpDec *self = GST_SCTP_DEC (object);
switch (prop_id) {
case PROP_GST_SCTP_ASSOCIATION_ID:
g_value_set_uint (value, self->sctp_association_id);
break;
case PROP_LOCAL_SCTP_PORT:
g_value_set_uint (value, self->local_sctp_port);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (self, prop_id, pspec);
break;
}
}
static GstStateChangeReturn
gst_sctp_dec_change_state (GstElement * element, GstStateChange transition)
{
GstSctpDec *self = GST_SCTP_DEC (element);
GstStateChangeReturn ret = GST_STATE_CHANGE_SUCCESS;
switch (transition) {
case GST_STATE_CHANGE_READY_TO_PAUSED:
if (!configure_association (self))
ret = GST_STATE_CHANGE_FAILURE;
break;
case GST_STATE_CHANGE_PAUSED_TO_READY:
sctpdec_cleanup (self);
break;
default:
break;
}
if (ret != GST_STATE_CHANGE_FAILURE)
ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
return ret;
}
static GstFlowReturn
gst_sctp_dec_packet_chain (GstPad * pad, GstSctpDec * self, GstBuffer * buf)
{
GstMapInfo map;
if (!gst_buffer_map (buf, &map, GST_MAP_READ)) {
GST_WARNING_OBJECT (self, "Could not map GstBuffer");
gst_buffer_unref (buf);
return GST_FLOW_ERROR;
}
gst_sctp_association_incoming_packet (self->sctp_association,
(guint8 *) map.data, (guint32) map.size);
gst_buffer_unmap (buf, &map);
gst_buffer_unref (buf);
return GST_FLOW_OK;
}
static void
flush_srcpad (const GValue * item, gpointer user_data)
{
GstSctpDecPad *sctpdec_pad = g_value_get_object (item);
gboolean flush = GPOINTER_TO_INT (user_data);
if (flush) {
gst_data_queue_set_flushing (sctpdec_pad->packet_queue, TRUE);
gst_data_queue_flush (sctpdec_pad->packet_queue);
} else {
gst_data_queue_set_flushing (sctpdec_pad->packet_queue, FALSE);
gst_pad_start_task (GST_PAD (sctpdec_pad),
(GstTaskFunction) gst_sctp_data_srcpad_loop, sctpdec_pad, NULL);
}
}
static gboolean
gst_sctp_dec_packet_event (GstPad * pad, GstSctpDec * self, GstEvent * event)
{
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_STREAM_START:
case GST_EVENT_CAPS:
/* We create our own stream-start events and the caps event does not
* make sense */
gst_event_unref (event);
return TRUE;
case GST_EVENT_EOS:
/* Drop this, we're never EOS until shut down */
gst_event_unref (event);
return TRUE;
case GST_EVENT_FLUSH_START:{
GstIterator *it;
it = gst_element_iterate_src_pads (GST_ELEMENT (self));
while (gst_iterator_foreach (it, flush_srcpad,
GINT_TO_POINTER (TRUE)) == GST_ITERATOR_RESYNC)
gst_iterator_resync (it);
gst_iterator_free (it);
return gst_pad_event_default (pad, GST_OBJECT (self), event);
}
case GST_EVENT_FLUSH_STOP:{
GstIterator *it;
it = gst_element_iterate_src_pads (GST_ELEMENT (self));
while (gst_iterator_foreach (it, flush_srcpad,
GINT_TO_POINTER (FALSE)) == GST_ITERATOR_RESYNC)
gst_iterator_resync (it);
gst_iterator_free (it);
return gst_pad_event_default (pad, GST_OBJECT (self), event);
}
default:
return gst_pad_event_default (pad, GST_OBJECT (self), event);
}
}
static void
gst_sctp_data_srcpad_loop (GstPad * pad)
{
GstSctpDecPad *sctpdec_pad = GST_SCTP_DEC_PAD (pad);
GstDataQueueItem *item;
if (gst_data_queue_pop (sctpdec_pad->packet_queue, &item)) {
GstFlowReturn flow_ret;
flow_ret = gst_pad_push (pad, GST_BUFFER (item->object));
item->object = NULL;
if (G_UNLIKELY (flow_ret == GST_FLOW_FLUSHING
|| flow_ret == GST_FLOW_NOT_LINKED)) {
GST_DEBUG_OBJECT (pad, "Push failed on packet source pad. Error: %s",
gst_flow_get_name (flow_ret));
} else if (G_UNLIKELY (flow_ret != GST_FLOW_OK)) {
GST_ERROR_OBJECT (pad, "Push failed on packet source pad. Error: %s",
gst_flow_get_name (flow_ret));
}
if (G_UNLIKELY (flow_ret != GST_FLOW_OK)) {
GST_DEBUG_OBJECT (pad, "Pausing task because of an error");
gst_data_queue_set_flushing (sctpdec_pad->packet_queue, TRUE);
gst_data_queue_flush (sctpdec_pad->packet_queue);
gst_pad_pause_task (pad);
}
item->destroy (item);
} else {
GST_DEBUG_OBJECT (pad, "Pausing task because we're flushing");
gst_pad_pause_task (pad);
}
}
static gboolean
configure_association (GstSctpDec * self)
{
gint state;
self->sctp_association = gst_sctp_association_get (self->sctp_association_id);
g_object_get (self->sctp_association, "state", &state, NULL);
if (state != GST_SCTP_ASSOCIATION_STATE_NEW) {
GST_WARNING_OBJECT (self,
"Could not configure SCTP association. Association already in use!");
g_object_unref (self->sctp_association);
self->sctp_association = NULL;
goto error;
}
self->signal_handler_stream_reset =
g_signal_connect_object (self->sctp_association, "stream-reset",
G_CALLBACK (on_gst_sctp_association_stream_reset), self, 0);
g_object_bind_property (self, "local-sctp-port", self->sctp_association,
"local-port", G_BINDING_SYNC_CREATE);
gst_sctp_association_set_on_packet_received (self->sctp_association,
on_receive, self);
return TRUE;
error:
return FALSE;
}
static gboolean
gst_sctp_dec_src_event (GstPad * pad, GstSctpDec * self, GstEvent * event)
{
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_RECONFIGURE:
case GST_EVENT_FLUSH_STOP:{
GstSctpDecPad *sctpdec_pad = GST_SCTP_DEC_PAD (pad);
/* Unflush and start task again */
gst_data_queue_set_flushing (sctpdec_pad->packet_queue, FALSE);
gst_pad_start_task (pad, (GstTaskFunction) gst_sctp_data_srcpad_loop, pad,
NULL);
return gst_pad_event_default (pad, GST_OBJECT (self), event);
}
case GST_EVENT_FLUSH_START:{
GstSctpDecPad *sctpdec_pad = GST_SCTP_DEC_PAD (pad);
gst_data_queue_set_flushing (sctpdec_pad->packet_queue, TRUE);
gst_data_queue_flush (sctpdec_pad->packet_queue);
return gst_pad_event_default (pad, GST_OBJECT (self), event);
}
default:
return gst_pad_event_default (pad, GST_OBJECT (self), event);
}
}
static gboolean
copy_sticky_events (GstPad * pad, GstEvent ** event, gpointer user_data)
{
GstPad *new_pad = user_data;
if (GST_EVENT_TYPE (*event) != GST_EVENT_CAPS
&& GST_EVENT_TYPE (*event) != GST_EVENT_STREAM_START)
gst_pad_store_sticky_event (new_pad, *event);
return TRUE;
}
static GstPad *
get_pad_for_stream_id (GstSctpDec * self, guint16 stream_id)
{
GstPad *new_pad = NULL;
gint state;
gchar *pad_name, *pad_stream_id;
GstPadTemplate *template;
pad_name = g_strdup_printf ("src_%hu", stream_id);
new_pad = gst_element_get_static_pad (GST_ELEMENT (self), pad_name);
if (new_pad) {
g_free (pad_name);
return new_pad;
}
g_object_get (self->sctp_association, "state", &state, NULL);
if (state != GST_SCTP_ASSOCIATION_STATE_CONNECTED) {
GST_WARNING_OBJECT (self,
"The SCTP association must be established before a new stream can be created");
return NULL;
}
if (stream_id > MAX_STREAM_ID)
return NULL;
template = gst_static_pad_template_get (&src_template);
new_pad = g_object_new (GST_TYPE_SCTP_DEC_PAD, "name", pad_name,
"direction", template->direction, "template", template, NULL);
g_free (pad_name);
gst_pad_set_event_function (new_pad,
GST_DEBUG_FUNCPTR ((GstPadEventFunction) gst_sctp_dec_src_event));
if (!gst_pad_set_active (new_pad, TRUE))
goto error_cleanup;
pad_stream_id =
gst_pad_create_stream_id_printf (new_pad, GST_ELEMENT (self), "%hu",
stream_id);
gst_pad_push_event (new_pad, gst_event_new_stream_start (pad_stream_id));
g_free (pad_stream_id);
gst_pad_sticky_events_foreach (self->sink_pad, copy_sticky_events, new_pad);
if (!gst_element_add_pad (GST_ELEMENT (self), new_pad))
goto error_cleanup;
gst_pad_start_task (new_pad, (GstTaskFunction) gst_sctp_data_srcpad_loop,
new_pad, NULL);
gst_object_ref (new_pad);
return new_pad;
error_cleanup:
gst_object_unref (new_pad);
return NULL;
}
static void
remove_pad (GstElement * element, GstPad * pad)
{
stop_srcpad_task (pad);
gst_pad_set_active (pad, FALSE);
gst_element_remove_pad (element, pad);
}
static void
on_gst_sctp_association_stream_reset (GstSctpAssociation * gst_sctp_association,
guint16 stream_id, GstSctpDec * self)
{
gchar *pad_name;
GstPad *srcpad;
pad_name = g_strdup_printf ("src_%hu", stream_id);
srcpad = gst_element_get_static_pad (GST_ELEMENT (self), pad_name);
g_free (pad_name);
if (!srcpad) {
GST_WARNING_OBJECT (self, "Reset called on stream without a srcpad");
return;
}
remove_pad (GST_ELEMENT (self), srcpad);
gst_object_unref (srcpad);
}
static void
data_queue_item_free (GstDataQueueItem * item)
{
if (item->object)
gst_mini_object_unref (item->object);
g_free (item);
}
static void
on_receive (GstSctpAssociation * sctp_association, guint8 * buf, gsize length,
guint16 stream_id, guint ppid, gpointer user_data)
{
GstSctpDec *self = user_data;
GstSctpDecPad *sctpdec_pad;
GstPad *src_pad;
GstDataQueueItem *item;
GstBuffer *gstbuf;
src_pad = get_pad_for_stream_id (self, stream_id);
g_assert (src_pad);
sctpdec_pad = GST_SCTP_DEC_PAD (src_pad);
gstbuf = gst_buffer_new_wrapped (buf, length);
gst_sctp_buffer_add_receive_meta (gstbuf, ppid);
item = g_new0 (GstDataQueueItem, 1);
item->object = GST_MINI_OBJECT (gstbuf);
item->size = length;
item->visible = TRUE;
item->destroy = (GDestroyNotify) data_queue_item_free;
if (!gst_data_queue_push (sctpdec_pad->packet_queue, item)) {
item->destroy (item);
GST_DEBUG_OBJECT (src_pad, "Failed to push item because we're flushing");
}
gst_object_unref (src_pad);
}
static void
stop_srcpad_task (GstPad * pad)
{
GstSctpDecPad *sctpdec_pad = GST_SCTP_DEC_PAD (pad);
gst_data_queue_set_flushing (sctpdec_pad->packet_queue, TRUE);
gst_data_queue_flush (sctpdec_pad->packet_queue);
gst_pad_stop_task (pad);
}
static void
remove_pad_it (const GValue * item, gpointer user_data)
{
GstPad *pad = g_value_get_object (item);
GstSctpDec *self = user_data;
remove_pad (GST_ELEMENT (self), pad);
}
static void
stop_all_srcpad_tasks (GstSctpDec * self)
{
GstIterator *it;
it = gst_element_iterate_src_pads (GST_ELEMENT (self));
while (gst_iterator_foreach (it, remove_pad_it, self) == GST_ITERATOR_RESYNC)
gst_iterator_resync (it);
gst_iterator_free (it);
}
static void
sctpdec_cleanup (GstSctpDec * self)
{
if (self->sctp_association) {
gst_sctp_association_set_on_packet_received (self->sctp_association, NULL,
NULL);
g_signal_handler_disconnect (self->sctp_association,
self->signal_handler_stream_reset);
stop_all_srcpad_tasks (self);
gst_sctp_association_force_close (self->sctp_association);
g_object_unref (self->sctp_association);
self->sctp_association = NULL;
}
}
static void
on_reset_stream (GstSctpDec * self, guint stream_id)
{
if (self->sctp_association) {
gst_sctp_association_reset_stream (self->sctp_association, stream_id);
on_gst_sctp_association_stream_reset (self->sctp_association, stream_id,
self);
}
}

66
ext/sctp/gstsctpdec.h Normal file
View file

@ -0,0 +1,66 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifndef __GST_SCTP_DEC_H__
#define __GST_SCTP_DEC_H__
#include <gst/gst.h>
#include "sctpassociation.h"
G_BEGIN_DECLS
#define GST_TYPE_SCTP_DEC (gst_sctp_dec_get_type())
#define GST_SCTP_DEC(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj), GST_TYPE_SCTP_DEC, GstSctpDec))
#define GST_SCTP_DEC_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass), GST_TYPE_SCTP_DEC, GstSctpDecClass))
#define GST_IS_SCTP_DEC(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj), GST_TYPE_SCTP_DEC))
#define GST_IS_SCTP_DEC_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass), GST_TYPE_SCTP_DEC))
typedef struct _GstSctpDec GstSctpDec;
typedef struct _GstSctpDecClass GstSctpDecClass;
struct _GstSctpDec
{
GstElement element;
GstPad *sink_pad;
guint sctp_association_id;
guint local_sctp_port;
GstSctpAssociation *sctp_association;
gulong signal_handler_stream_reset;
};
struct _GstSctpDecClass
{
GstElementClass parent_class;
void (*on_reset_stream) (GstSctpDec * sctp_dec, guint stream_id);
};
GType gst_sctp_dec_get_type (void);
G_END_DECLS
#endif /* __GST_SCTP_DEC_H__ */

937
ext/sctp/gstsctpenc.c Normal file
View file

@ -0,0 +1,937 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "gstsctpenc.h"
#include <gst/sctp/sctpsendmeta.h>
#include <stdio.h>
GST_DEBUG_CATEGORY_STATIC (gst_sctp_enc_debug_category);
#define GST_CAT_DEFAULT gst_sctp_enc_debug_category
#define gst_sctp_enc_parent_class parent_class
G_DEFINE_TYPE (GstSctpEnc, gst_sctp_enc, GST_TYPE_ELEMENT);
static GstStaticPadTemplate sink_template =
GST_STATIC_PAD_TEMPLATE ("sink_%u", GST_PAD_SINK,
GST_PAD_REQUEST, GST_STATIC_CAPS_ANY);
static GstStaticPadTemplate src_template =
GST_STATIC_PAD_TEMPLATE ("src", GST_PAD_SRC,
GST_PAD_ALWAYS, GST_STATIC_CAPS ("application/x-sctp"));
enum
{
SIGNAL_SCTP_ASSOCIATION_ESTABLISHED,
SIGNAL_GET_STREAM_BYTES_SENT,
NUM_SIGNALS
};
static guint signals[NUM_SIGNALS];
enum
{
PROP_0,
PROP_GST_SCTP_ASSOCIATION_ID,
PROP_REMOTE_SCTP_PORT,
PROP_USE_SOCK_STREAM,
NUM_PROPERTIES
};
static GParamSpec *properties[NUM_PROPERTIES];
#define DEFAULT_GST_SCTP_ASSOCIATION_ID 1
#define DEFAULT_REMOTE_SCTP_PORT 0
#define DEFAULT_GST_SCTP_ORDERED TRUE
#define DEFAULT_SCTP_PPID 1
#define DEFAULT_USE_SOCK_STREAM FALSE
#define BUFFER_FULL_SLEEP_TIME 100000
GType gst_sctp_enc_pad_get_type (void);
#define GST_TYPE_SCTP_ENC_PAD (gst_sctp_enc_pad_get_type())
#define GST_SCTP_ENC_PAD(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj), GST_TYPE_SCTP_ENC_PAD, GstSctpEncPad))
#define GST_SCTP_ENC_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass), GST_TYPE_SCTP_ENC_PAD, GstSctpEncPadClass))
#define GST_IS_SCTP_ENC_PAD(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj), GST_TYPE_SCTP_ENC_PAD))
#define GST_IS_SCTP_ENC_PAD_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass), GST_TYPE_SCTP_ENC_PAD))
typedef struct _GstSctpEncPad GstSctpEncPad;
typedef GstPadClass GstSctpEncPadClass;
struct _GstSctpEncPad
{
GstPad parent;
guint16 stream_id;
gboolean ordered;
guint32 ppid;
GstSctpAssociationPartialReliability reliability;
guint32 reliability_param;
guint64 bytes_sent;
GMutex lock;
GCond cond;
gboolean flushing;
};
G_DEFINE_TYPE (GstSctpEncPad, gst_sctp_enc_pad, GST_TYPE_PAD);
static void
gst_sctp_enc_pad_finalize (GObject * object)
{
GstSctpEncPad *self = GST_SCTP_ENC_PAD (object);
g_cond_clear (&self->cond);
g_mutex_clear (&self->lock);
G_OBJECT_CLASS (gst_sctp_enc_pad_parent_class)->finalize (object);
}
static void
gst_sctp_enc_pad_class_init (GstSctpEncPadClass * klass)
{
GObjectClass *gobject_class = (GObjectClass *) klass;
gobject_class->finalize = gst_sctp_enc_pad_finalize;
}
static void
gst_sctp_enc_pad_init (GstSctpEncPad * self)
{
g_mutex_init (&self->lock);
g_cond_init (&self->cond);
self->flushing = FALSE;
}
static void gst_sctp_enc_finalize (GObject * object);
static void gst_sctp_enc_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec);
static void gst_sctp_enc_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
static GstStateChangeReturn gst_sctp_enc_change_state (GstElement * element,
GstStateChange transition);
static GstPad *gst_sctp_enc_request_new_pad (GstElement * element,
GstPadTemplate * template, const gchar * name, const GstCaps * caps);
static void gst_sctp_enc_release_pad (GstElement * element, GstPad * pad);
static void gst_sctp_enc_srcpad_loop (GstPad * pad);
static GstFlowReturn gst_sctp_enc_sink_chain (GstPad * pad, GstObject * parent,
GstBuffer * buffer);
static gboolean gst_sctp_enc_sink_event (GstPad * pad, GstObject * parent,
GstEvent * event);
static gboolean gst_sctp_enc_src_event (GstPad * pad, GstObject * parent,
GstEvent * event);
static void on_sctp_association_state_changed (GstSctpAssociation *
sctp_association, GParamSpec * pspec, GstSctpEnc * self);
static gboolean configure_association (GstSctpEnc * self);
static void on_sctp_packet_out (GstSctpAssociation * sctp_association,
const guint8 * buf, gsize length, gpointer user_data);
static void stop_srcpad_task (GstPad * pad, GstSctpEnc * self);
static void sctpenc_cleanup (GstSctpEnc * self);
static void get_config_from_caps (const GstCaps * caps, gboolean * ordered,
GstSctpAssociationPartialReliability * reliability,
guint32 * reliability_param, guint32 * ppid, gboolean * ppid_available);
static guint64 on_get_stream_bytes_sent (GstSctpEnc * self, guint stream_id);
static void
gst_sctp_enc_class_init (GstSctpEncClass * klass)
{
GObjectClass *gobject_class;
GstElementClass *element_class;
gobject_class = (GObjectClass *) klass;
element_class = (GstElementClass *) klass;
GST_DEBUG_CATEGORY_INIT (gst_sctp_enc_debug_category,
"sctpenc", 0, "debug category for sctpenc element");
gst_element_class_add_pad_template (GST_ELEMENT_CLASS (klass),
gst_static_pad_template_get (&src_template));
gst_element_class_add_pad_template (GST_ELEMENT_CLASS (klass),
gst_static_pad_template_get (&sink_template));
gobject_class->finalize = GST_DEBUG_FUNCPTR (gst_sctp_enc_finalize);
gobject_class->set_property = GST_DEBUG_FUNCPTR (gst_sctp_enc_set_property);
gobject_class->get_property = GST_DEBUG_FUNCPTR (gst_sctp_enc_get_property);
element_class->change_state = GST_DEBUG_FUNCPTR (gst_sctp_enc_change_state);
element_class->request_new_pad =
GST_DEBUG_FUNCPTR (gst_sctp_enc_request_new_pad);
element_class->release_pad = GST_DEBUG_FUNCPTR (gst_sctp_enc_release_pad);
properties[PROP_GST_SCTP_ASSOCIATION_ID] =
g_param_spec_uint ("sctp-association-id",
"SCTP Association ID",
"Every encoder/decoder pair should have the same, unique, sctp-association-id. "
"This value must be set before any pads are requested.",
0, G_MAXUINT, DEFAULT_GST_SCTP_ASSOCIATION_ID,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
properties[PROP_REMOTE_SCTP_PORT] =
g_param_spec_uint ("remote-sctp-port",
"Remote SCTP port",
"Sctp remote sctp port for the sctp association. The local port is configured via the "
"GstSctpDec element.",
0, G_MAXUSHORT, DEFAULT_REMOTE_SCTP_PORT,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
properties[PROP_USE_SOCK_STREAM] =
g_param_spec_boolean ("use-sock-stream",
"Use sock-stream",
"When set to TRUE, a sequenced, reliable, connection-based connection is used."
"When TRUE the partial reliability parameters of the channel are ignored.",
DEFAULT_USE_SOCK_STREAM, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
g_object_class_install_properties (gobject_class, NUM_PROPERTIES, properties);
signals[SIGNAL_SCTP_ASSOCIATION_ESTABLISHED] =
g_signal_new ("sctp-association-established",
G_TYPE_FROM_CLASS (gobject_class), G_SIGNAL_RUN_LAST,
G_STRUCT_OFFSET (GstSctpEncClass, on_sctp_association_is_established),
NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 1, G_TYPE_BOOLEAN);
signals[SIGNAL_GET_STREAM_BYTES_SENT] = g_signal_new ("bytes-sent",
G_TYPE_FROM_CLASS (gobject_class), G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstSctpEncClass, on_get_stream_bytes_sent), NULL, NULL,
g_cclosure_marshal_generic, G_TYPE_UINT64, 1, G_TYPE_UINT);
klass->on_get_stream_bytes_sent =
GST_DEBUG_FUNCPTR (on_get_stream_bytes_sent);
gst_element_class_set_static_metadata (element_class,
"SCTP Encoder",
"Encoder/Network/SCTP",
"Encodes packets with SCTP",
"George Kiagiadakis <george.kiagiadakis@collabora.com>");
}
static gboolean
data_queue_check_full_cb (GstDataQueue * queue, guint visible, guint bytes,
guint64 time, gpointer user_data)
{
/* TODO: When are we considered full? */
return FALSE;
}
static void
data_queue_empty_cb (GstDataQueue * queue, gpointer user_data)
{
}
static void
data_queue_full_cb (GstDataQueue * queue, gpointer user_data)
{
}
static void
gst_sctp_enc_init (GstSctpEnc * self)
{
self->sctp_association_id = DEFAULT_GST_SCTP_ASSOCIATION_ID;
self->remote_sctp_port = DEFAULT_REMOTE_SCTP_PORT;
self->sctp_association = NULL;
self->outbound_sctp_packet_queue =
gst_data_queue_new (data_queue_check_full_cb, data_queue_full_cb,
data_queue_empty_cb, NULL);
self->src_pad = gst_pad_new_from_static_template (&src_template, "src");
gst_pad_set_event_function (self->src_pad,
GST_DEBUG_FUNCPTR ((GstPadEventFunction) gst_sctp_enc_src_event));
gst_element_add_pad (GST_ELEMENT (self), self->src_pad);
g_queue_init (&self->pending_pads);
}
static void
gst_sctp_enc_finalize (GObject * object)
{
GstSctpEnc *self = GST_SCTP_ENC (object);
g_queue_clear (&self->pending_pads);
gst_object_unref (self->outbound_sctp_packet_queue);
G_OBJECT_CLASS (parent_class)->finalize (object);
}
static void
gst_sctp_enc_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec)
{
GstSctpEnc *self = GST_SCTP_ENC (object);
switch (prop_id) {
case PROP_GST_SCTP_ASSOCIATION_ID:
self->sctp_association_id = g_value_get_uint (value);
break;
case PROP_REMOTE_SCTP_PORT:
self->remote_sctp_port = g_value_get_uint (value);
break;
case PROP_USE_SOCK_STREAM:
self->use_sock_stream = g_value_get_boolean (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (self, prop_id, pspec);
break;
}
}
static void
gst_sctp_enc_get_property (GObject * object, guint prop_id, GValue * value,
GParamSpec * pspec)
{
GstSctpEnc *self = GST_SCTP_ENC (object);
switch (prop_id) {
case PROP_GST_SCTP_ASSOCIATION_ID:
g_value_set_uint (value, self->sctp_association_id);
break;
case PROP_REMOTE_SCTP_PORT:
g_value_set_uint (value, self->remote_sctp_port);
break;
case PROP_USE_SOCK_STREAM:
g_value_set_boolean (value, self->use_sock_stream);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (self, prop_id, pspec);
break;
}
}
static GstStateChangeReturn
gst_sctp_enc_change_state (GstElement * element, GstStateChange transition)
{
GstSctpEnc *self = GST_SCTP_ENC (element);
GstStateChangeReturn ret = GST_STATE_CHANGE_FAILURE;
gboolean res = TRUE;
switch (transition) {
case GST_STATE_CHANGE_NULL_TO_READY:
break;
case GST_STATE_CHANGE_READY_TO_PAUSED:
self->need_segment = self->need_stream_start_caps = TRUE;
gst_data_queue_set_flushing (self->outbound_sctp_packet_queue, FALSE);
gst_pad_start_task (self->src_pad,
(GstTaskFunction) gst_sctp_enc_srcpad_loop, self->src_pad, NULL);
res = configure_association (self);
break;
case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
break;
case GST_STATE_CHANGE_PAUSED_TO_READY:
sctpenc_cleanup (self);
break;
case GST_STATE_CHANGE_READY_TO_NULL:
break;
default:
break;
}
if (res)
ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
return ret;
}
static GstPad *
gst_sctp_enc_request_new_pad (GstElement * element, GstPadTemplate * template,
const gchar * new_pad_name, const GstCaps * caps)
{
GstSctpEnc *self = GST_SCTP_ENC (element);
GstPad *new_pad = NULL;
GstSctpEncPad *sctpenc_pad;
guint32 stream_id;
gint state;
guint32 new_ppid;
gboolean is_new_ppid;
g_object_get (self->sctp_association, "state", &state, NULL);
if (state != GST_SCTP_ASSOCIATION_STATE_CONNECTED) {
g_warning
("The SCTP association must be established before a new stream can be created");
goto invalid_state;
}
if (!template)
goto invalid_parameter;
if (!new_pad_name || (sscanf (new_pad_name, "sink_%u", &stream_id) != 1)
|| stream_id > 65534) /* 65535 is not a valid stream id */
goto invalid_parameter;
new_pad = gst_element_get_static_pad (element, new_pad_name);
if (new_pad) {
gst_object_unref (new_pad);
new_pad = NULL;
goto invalid_parameter;
}
new_pad =
g_object_new (GST_TYPE_SCTP_ENC_PAD, "name", new_pad_name, "direction",
template->direction, "template", template, NULL);
gst_pad_set_chain_function (new_pad,
GST_DEBUG_FUNCPTR (gst_sctp_enc_sink_chain));
gst_pad_set_event_function (new_pad,
GST_DEBUG_FUNCPTR (gst_sctp_enc_sink_event));
sctpenc_pad = GST_SCTP_ENC_PAD (new_pad);
sctpenc_pad->stream_id = stream_id;
sctpenc_pad->ppid = DEFAULT_SCTP_PPID;
get_config_from_caps (caps, &sctpenc_pad->ordered, &sctpenc_pad->reliability,
&sctpenc_pad->reliability_param, &new_ppid, &is_new_ppid);
if (is_new_ppid)
sctpenc_pad->ppid = new_ppid;
sctpenc_pad->flushing = FALSE;
if (!gst_pad_set_active (new_pad, TRUE))
goto error_cleanup;
if (!gst_element_add_pad (element, new_pad))
goto error_cleanup;
invalid_state:
invalid_parameter:
return new_pad;
error_cleanup:
gst_object_unref (new_pad);
return NULL;
}
static void
gst_sctp_enc_release_pad (GstElement * element, GstPad * pad)
{
GstSctpEncPad *sctpenc_pad = GST_SCTP_ENC_PAD (pad);
GstSctpEnc *self;
guint stream_id = 0;
self = GST_SCTP_ENC (element);
g_mutex_lock (&sctpenc_pad->lock);
sctpenc_pad->flushing = TRUE;
g_cond_signal (&sctpenc_pad->cond);
g_mutex_unlock (&sctpenc_pad->lock);
stream_id = sctpenc_pad->stream_id;
gst_pad_set_active (pad, FALSE);
if (self->sctp_association)
gst_sctp_association_reset_stream (self->sctp_association, stream_id);
gst_element_remove_pad (element, pad);
}
static void
gst_sctp_enc_srcpad_loop (GstPad * pad)
{
GstSctpEnc *self = GST_SCTP_ENC (GST_PAD_PARENT (pad));
GstFlowReturn flow_ret;
GstDataQueueItem *item;
if (self->need_stream_start_caps) {
gchar s_id[32];
GstCaps *caps;
g_snprintf (s_id, sizeof (s_id), "sctpenc-%08x", g_random_int ());
gst_pad_push_event (self->src_pad, gst_event_new_stream_start (s_id));
caps = gst_caps_new_empty_simple ("application/x-sctp");
gst_pad_set_caps (self->src_pad, caps);
gst_caps_unref (caps);
self->need_stream_start_caps = FALSE;
}
if (self->need_segment) {
GstSegment segment;
gst_segment_init (&segment, GST_FORMAT_BYTES);
gst_pad_push_event (self->src_pad, gst_event_new_segment (&segment));
self->need_segment = FALSE;
}
if (gst_data_queue_pop (self->outbound_sctp_packet_queue, &item)) {
flow_ret = gst_pad_push (self->src_pad, GST_BUFFER (item->object));
item->object = NULL;
if (G_UNLIKELY (flow_ret == GST_FLOW_FLUSHING
|| flow_ret == GST_FLOW_NOT_LINKED)) {
GST_DEBUG_OBJECT (pad, "Push failed on packet source pad. Error: %s",
gst_flow_get_name (flow_ret));
} else if (G_UNLIKELY (flow_ret != GST_FLOW_OK)) {
GST_ERROR_OBJECT (pad, "Push failed on packet source pad. Error: %s",
gst_flow_get_name (flow_ret));
}
if (G_UNLIKELY (flow_ret != GST_FLOW_OK)) {
GST_DEBUG_OBJECT (pad, "Pausing task because of an error");
gst_data_queue_set_flushing (self->outbound_sctp_packet_queue, TRUE);
gst_data_queue_flush (self->outbound_sctp_packet_queue);
gst_pad_pause_task (pad);
}
item->destroy (item);
} else {
GST_DEBUG_OBJECT (pad, "Pausing task because we're flushing");
gst_pad_pause_task (pad);
}
}
static GstFlowReturn
gst_sctp_enc_sink_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{
GstSctpEnc *self = GST_SCTP_ENC (parent);
GstSctpEncPad *sctpenc_pad = GST_SCTP_ENC_PAD (pad);
GstMapInfo map;
guint32 ppid;
gboolean ordered;
GstSctpAssociationPartialReliability pr;
guint32 pr_param;
gpointer state = NULL;
GstMeta *meta;
const GstMetaInfo *meta_info = GST_SCTP_SEND_META_INFO;
GstFlowReturn flow_ret = GST_FLOW_ERROR;
ppid = sctpenc_pad->ppid;
ordered = sctpenc_pad->ordered;
pr = sctpenc_pad->reliability;
pr_param = sctpenc_pad->reliability_param;
while ((meta = gst_buffer_iterate_meta (buffer, &state))) {
if (meta->info->api == meta_info->api) {
GstSctpSendMeta *sctp_send_meta = (GstSctpSendMeta *) meta;
ppid = sctp_send_meta->ppid;
ordered = sctp_send_meta->ordered;
pr_param = sctp_send_meta->pr_param;
switch (sctp_send_meta->pr) {
case GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE:
pr = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_NONE;
break;
case GST_SCTP_SEND_META_PARTIAL_RELIABILITY_RTX:
pr = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_RTX;
break;
case GST_SCTP_SEND_META_PARTIAL_RELIABILITY_BUF:
pr = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_BUF;
break;
case GST_SCTP_SEND_META_PARTIAL_RELIABILITY_TTL:
pr = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_TTL;
break;
}
break;
}
}
if (!gst_buffer_map (buffer, &map, GST_MAP_READ)) {
g_warning ("Could not map GstBuffer");
goto error;
}
g_mutex_lock (&sctpenc_pad->lock);
while (!sctpenc_pad->flushing) {
gboolean data_sent = FALSE;
g_mutex_unlock (&sctpenc_pad->lock);
data_sent =
gst_sctp_association_send_data (self->sctp_association, map.data,
map.size, sctpenc_pad->stream_id, ppid, ordered, pr, pr_param);
g_mutex_lock (&sctpenc_pad->lock);
if (data_sent) {
sctpenc_pad->bytes_sent += map.size;
break;
} else if (!sctpenc_pad->flushing) {
gint64 end_time = g_get_monotonic_time () + BUFFER_FULL_SLEEP_TIME;
/* The buffer was probably full. Retry in a while */
GST_OBJECT_LOCK (self);
g_queue_push_tail (&self->pending_pads, sctpenc_pad);
GST_OBJECT_UNLOCK (self);
g_cond_wait_until (&sctpenc_pad->cond, &sctpenc_pad->lock, end_time);
GST_OBJECT_LOCK (self);
g_queue_remove (&self->pending_pads, sctpenc_pad);
GST_OBJECT_UNLOCK (self);
}
}
flow_ret = sctpenc_pad->flushing ? GST_FLOW_FLUSHING : GST_FLOW_OK;
g_mutex_unlock (&sctpenc_pad->lock);
gst_buffer_unmap (buffer, &map);
error:
gst_buffer_unref (buffer);
return flow_ret;
}
static gboolean
gst_sctp_enc_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
GstSctpEncPad *sctpenc_pad = GST_SCTP_ENC_PAD (pad);
gboolean ret, is_new_ppid;
guint32 new_ppid;
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_CAPS:{
GstCaps *caps;
gst_event_parse_caps (event, &caps);
get_config_from_caps (caps, &sctpenc_pad->ordered,
&sctpenc_pad->reliability, &sctpenc_pad->reliability_param, &new_ppid,
&is_new_ppid);
if (is_new_ppid)
sctpenc_pad->ppid = new_ppid;
gst_event_unref (event);
ret = TRUE;
break;
}
case GST_EVENT_STREAM_START:
case GST_EVENT_SEGMENT:
/* Drop these, we create our own */
ret = TRUE;
gst_event_unref (event);
break;
case GST_EVENT_EOS:
/* Drop this, we're never EOS until shut down */
ret = TRUE;
gst_event_unref (event);
break;
case GST_EVENT_FLUSH_START:
g_mutex_lock (&sctpenc_pad->lock);
sctpenc_pad->flushing = TRUE;
g_cond_signal (&sctpenc_pad->cond);
g_mutex_unlock (&sctpenc_pad->lock);
ret = gst_pad_event_default (pad, parent, event);
break;
case GST_EVENT_FLUSH_STOP:
sctpenc_pad->flushing = FALSE;
ret = gst_pad_event_default (pad, parent, event);
break;
default:
ret = gst_pad_event_default (pad, parent, event);
break;
}
return ret;
}
static void
flush_sinkpad (const GValue * item, gpointer user_data)
{
GstSctpEncPad *sctpenc_pad = g_value_get_object (item);
gboolean flush = GPOINTER_TO_INT (user_data);
if (flush) {
g_mutex_lock (&sctpenc_pad->lock);
sctpenc_pad->flushing = TRUE;
g_cond_signal (&sctpenc_pad->cond);
g_mutex_unlock (&sctpenc_pad->lock);
} else {
sctpenc_pad->flushing = FALSE;
}
}
static gboolean
gst_sctp_enc_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
GstSctpEnc *self = GST_SCTP_ENC (parent);
gboolean ret;
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_FLUSH_START:{
GstIterator *it;
gst_data_queue_set_flushing (self->outbound_sctp_packet_queue, TRUE);
gst_data_queue_flush (self->outbound_sctp_packet_queue);
it = gst_element_iterate_sink_pads (GST_ELEMENT (self));
while (gst_iterator_foreach (it, flush_sinkpad,
GINT_TO_POINTER (TRUE)) == GST_ITERATOR_RESYNC)
gst_iterator_resync (it);
gst_iterator_free (it);
ret = gst_pad_event_default (pad, parent, event);
break;
}
case GST_EVENT_RECONFIGURE:
case GST_EVENT_FLUSH_STOP:{
GstIterator *it;
it = gst_element_iterate_sink_pads (GST_ELEMENT (self));
while (gst_iterator_foreach (it, flush_sinkpad,
GINT_TO_POINTER (FALSE)) == GST_ITERATOR_RESYNC)
gst_iterator_resync (it);
gst_iterator_free (it);
gst_data_queue_set_flushing (self->outbound_sctp_packet_queue, FALSE);
self->need_segment = TRUE;
gst_pad_start_task (self->src_pad,
(GstTaskFunction) gst_sctp_enc_srcpad_loop, self->src_pad, NULL);
ret = gst_pad_event_default (pad, parent, event);
break;
}
default:
ret = gst_pad_event_default (pad, parent, event);
break;
}
return ret;
}
static gboolean
configure_association (GstSctpEnc * self)
{
gint state;
self->sctp_association = gst_sctp_association_get (self->sctp_association_id);
g_object_get (self->sctp_association, "state", &state, NULL);
if (state != GST_SCTP_ASSOCIATION_STATE_NEW) {
GST_WARNING_OBJECT (self,
"Could not configure SCTP association. Association already in use!");
g_object_unref (self->sctp_association);
self->sctp_association = NULL;
goto error;
}
self->signal_handler_state_changed =
g_signal_connect_object (self->sctp_association, "notify::state",
G_CALLBACK (on_sctp_association_state_changed), self, 0);
g_object_bind_property (self, "remote-sctp-port", self->sctp_association,
"remote-port", G_BINDING_SYNC_CREATE);
g_object_bind_property (self, "use-sock-stream", self->sctp_association,
"use-sock-stream", G_BINDING_SYNC_CREATE);
gst_sctp_association_set_on_packet_out (self->sctp_association,
on_sctp_packet_out, self);
return TRUE;
error:
return FALSE;
}
static void
on_sctp_association_state_changed (GstSctpAssociation * sctp_association,
GParamSpec * pspec, GstSctpEnc * self)
{
gint state;
g_object_get (sctp_association, "state", &state, NULL);
switch (state) {
case GST_SCTP_ASSOCIATION_STATE_NEW:
break;
case GST_SCTP_ASSOCIATION_STATE_READY:
gst_sctp_association_start (sctp_association);
break;
case GST_SCTP_ASSOCIATION_STATE_CONNECTING:
break;
case GST_SCTP_ASSOCIATION_STATE_CONNECTED:
g_signal_emit_by_name (self, "sctp-association-established", TRUE);
break;
case GST_SCTP_ASSOCIATION_STATE_DISCONNECTING:
g_signal_emit (self, signals[SIGNAL_SCTP_ASSOCIATION_ESTABLISHED], 0,
FALSE);
break;
case GST_SCTP_ASSOCIATION_STATE_DISCONNECTED:
break;
case GST_SCTP_ASSOCIATION_STATE_ERROR:
break;
}
}
static void
data_queue_item_free (GstDataQueueItem * item)
{
if (item->object)
gst_mini_object_unref (item->object);
g_free (item);
}
static void
on_sctp_packet_out (GstSctpAssociation * _association, const guint8 * buf,
gsize length, gpointer user_data)
{
GstSctpEnc *self = user_data;
GstBuffer *gstbuf;
GstDataQueueItem *item;
GList *pending_pads, *l;
GstSctpEncPad *sctpenc_pad;
gstbuf = gst_buffer_new_wrapped (g_memdup (buf, length), length);
item = g_new0 (GstDataQueueItem, 1);
item->object = GST_MINI_OBJECT (gstbuf);
item->size = length;
item->visible = TRUE;
item->destroy = (GDestroyNotify) data_queue_item_free;
if (!gst_data_queue_push (self->outbound_sctp_packet_queue, item)) {
item->destroy (item);
GST_DEBUG_OBJECT (self, "Failed to push item because we're flushing");
}
/* Wake up pads in the order they waited, oldest pad first */
GST_OBJECT_LOCK (self);
pending_pads = NULL;
while ((sctpenc_pad = g_queue_pop_tail (&self->pending_pads))) {
pending_pads = g_list_prepend (pending_pads, sctpenc_pad);
}
GST_OBJECT_UNLOCK (self);
for (l = pending_pads; l; l = l->next) {
sctpenc_pad = l->data;
g_mutex_lock (&sctpenc_pad->lock);
g_cond_signal (&sctpenc_pad->cond);
g_mutex_unlock (&sctpenc_pad->lock);
}
g_list_free (pending_pads);
}
static void
stop_srcpad_task (GstPad * pad, GstSctpEnc * self)
{
gst_data_queue_set_flushing (self->outbound_sctp_packet_queue, TRUE);
gst_data_queue_flush (self->outbound_sctp_packet_queue);
gst_pad_stop_task (pad);
}
static void
remove_sinkpad (const GValue * item, gpointer user_data)
{
GstSctpEncPad *sctpenc_pad = g_value_get_object (item);
GstSctpEnc *self = user_data;
gst_sctp_enc_release_pad (GST_ELEMENT (self), GST_PAD (sctpenc_pad));
}
static void
sctpenc_cleanup (GstSctpEnc * self)
{
GstIterator *it;
gst_sctp_association_set_on_packet_out (self->sctp_association, NULL, NULL);
g_signal_handler_disconnect (self->sctp_association,
self->signal_handler_state_changed);
stop_srcpad_task (self->src_pad, self);
gst_sctp_association_force_close (self->sctp_association);
g_object_unref (self->sctp_association);
self->sctp_association = NULL;
it = gst_element_iterate_sink_pads (GST_ELEMENT (self));
while (gst_iterator_foreach (it, remove_sinkpad, self) == GST_ITERATOR_RESYNC)
gst_iterator_resync (it);
gst_iterator_free (it);
g_queue_clear (&self->pending_pads);
}
static void
get_config_from_caps (const GstCaps * caps, gboolean * ordered,
GstSctpAssociationPartialReliability * reliability,
guint32 * reliability_param, guint32 * ppid, gboolean * ppid_available)
{
GstStructure *s;
guint i, n;
*ordered = TRUE;
*reliability = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_NONE;
*reliability_param = 0;
*ppid_available = FALSE;
n = gst_caps_get_size (caps);
for (i = 0; i < n; i++) {
s = gst_caps_get_structure (caps, i);
if (gst_structure_has_field (s, "ordered")) {
const GValue *v = gst_structure_get_value (s, "ordered");
*ordered = g_value_get_boolean (v);
}
if (gst_structure_has_field (s, "partially-reliability")) {
const GValue *v = gst_structure_get_value (s, "partially-reliability");
const gchar *reliability_string = g_value_get_string (v);
if (!g_strcmp0 (reliability_string, "none"))
*reliability = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_NONE;
else if (!g_strcmp0 (reliability_string, "ttl"))
*reliability = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_TTL;
else if (!g_strcmp0 (reliability_string, "buf"))
*reliability = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_BUF;
else if (!g_strcmp0 (reliability_string, "rtx"))
*reliability = GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_RTX;
}
if (gst_structure_has_field (s, "reliability-parameter")) {
const GValue *v = gst_structure_get_value (s, "reliability-parameter");
*reliability_param = g_value_get_uint (v);
}
if (gst_structure_has_field (s, "ppid")) {
const GValue *v = gst_structure_get_value (s, "ppid");
*ppid = g_value_get_uint (v);
*ppid_available = TRUE;
}
}
}
static guint64
on_get_stream_bytes_sent (GstSctpEnc * self, guint stream_id)
{
gchar *pad_name;
GstPad *pad;
GstSctpEncPad *sctpenc_pad;
guint64 bytes_sent;
pad_name = g_strdup_printf ("sink_%u", stream_id);
pad = gst_element_get_static_pad (GST_ELEMENT (self), pad_name);
g_free (pad_name);
if (!pad) {
GST_DEBUG_OBJECT (self,
"Buffered amount requested on a stream that does not exist!");
return 0;
}
sctpenc_pad = GST_SCTP_ENC_PAD (pad);
g_mutex_lock (&sctpenc_pad->lock);
bytes_sent = sctpenc_pad->bytes_sent;
g_mutex_unlock (&sctpenc_pad->lock);
gst_object_unref (sctpenc_pad);
return bytes_sent;
}

77
ext/sctp/gstsctpenc.h Normal file
View file

@ -0,0 +1,77 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifndef __GST_SCTP_ENC_H__
#define __GST_SCTP_ENC_H__
#include <gst/gst.h>
#include <gst/base/base.h>
#include "sctpassociation.h"
G_BEGIN_DECLS
#define GST_TYPE_SCTP_ENC (gst_sctp_enc_get_type())
#define GST_SCTP_ENC(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj), GST_TYPE_SCTP_ENC, GstSctpEnc))
#define GST_SCTP_ENC_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass), GST_TYPE_SCTP_ENC, GstSctpEncClass))
#define GST_IS_SCTP_ENC(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj), GST_TYPE_SCTP_ENC))
#define GST_IS_SCTP_ENC_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass), GST_TYPE_SCTP_ENC))
typedef struct _GstSctpEnc GstSctpEnc;
typedef struct _GstSctpEncClass GstSctpEncClass;
typedef struct _GstSctpEncPrivate GstSctpEncPrivate;
struct _GstSctpEnc
{
GstElement element;
GstPad *src_pad;
gboolean need_stream_start_caps, need_segment;
guint32 sctp_association_id;
guint16 remote_sctp_port;
gboolean use_sock_stream;
GstSctpAssociation *sctp_association;
GstDataQueue *outbound_sctp_packet_queue;
GQueue pending_pads;
gulong signal_handler_state_changed;
};
struct _GstSctpEncClass
{
GstElementClass parent_class;
void (*on_sctp_association_is_established) (GstSctpEnc * sctp_enc,
gboolean established);
guint64 (*on_get_stream_bytes_sent) (GstSctpEnc * sctp_enc,
guint stream_id);
};
GType gst_sctp_enc_get_type (void);
G_END_DECLS
#endif /* __GST_SCTP_ENC_H__ */

53
ext/sctp/gstsctpplugin.c Normal file
View file

@ -0,0 +1,53 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "gstsctpdec.h"
#include "gstsctpenc.h"
#include <gst/gst.h>
static gboolean
plugin_init (GstPlugin * plugin)
{
return gst_element_register (plugin, "sctpenc", GST_RANK_NONE,
GST_TYPE_SCTP_ENC)
&& gst_element_register (plugin, "sctpdec", GST_RANK_NONE,
GST_TYPE_SCTP_DEC);
}
#ifndef PACKAGE
#define PACKAGE "sctp"
#endif
GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
GST_VERSION_MINOR,
sctp,
"SCTP encoder/decoder plugin",
plugin_init, PACKAGE_VERSION, "BSD", GST_PACKAGE_NAME, GST_PACKAGE_ORIGIN)

836
ext/sctp/sctpassociation.c Normal file
View file

@ -0,0 +1,836 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include "sctpassociation.h"
#include <string.h>
#include <errno.h>
#include <stdlib.h>
#define GST_SCTP_ASSOCIATION_STATE_TYPE (gst_sctp_association_state_get_type())
static GType
gst_sctp_association_state_get_type (void)
{
static const GEnumValue values[] = {
{GST_SCTP_ASSOCIATION_STATE_NEW, "state-new", "state-new"},
{GST_SCTP_ASSOCIATION_STATE_READY, "state-ready", "state-ready"},
{GST_SCTP_ASSOCIATION_STATE_CONNECTING, "state-connecting",
"state-connecting"},
{GST_SCTP_ASSOCIATION_STATE_CONNECTED, "state-connected",
"state-connected"},
{GST_SCTP_ASSOCIATION_STATE_DISCONNECTING, "state-disconnecting",
"state-disconnecting"},
{GST_SCTP_ASSOCIATION_STATE_DISCONNECTED, "state-disconnected",
"state-disconnected"},
{GST_SCTP_ASSOCIATION_STATE_ERROR, "state-error", "state-error"},
{0, NULL, NULL}
};
static volatile GType id = 0;
if (g_once_init_enter ((gsize *) & id)) {
GType _id;
_id = g_enum_register_static ("GstSctpAssociationState", values);
g_once_init_leave ((gsize *) & id, _id);
}
return id;
}
G_DEFINE_TYPE (GstSctpAssociation, gst_sctp_association, G_TYPE_OBJECT);
enum
{
SIGNAL_STREAM_RESET,
LAST_SIGNAL
};
enum
{
PROP_0,
PROP_ASSOCIATION_ID,
PROP_LOCAL_PORT,
PROP_REMOTE_PORT,
PROP_STATE,
PROP_USE_SOCK_STREAM,
NUM_PROPERTIES
};
static guint signals[LAST_SIGNAL] = { 0 };
static GParamSpec *properties[NUM_PROPERTIES];
#define DEFAULT_NUMBER_OF_SCTP_STREAMS 10
#define DEFAULT_LOCAL_SCTP_PORT 0
#define DEFAULT_REMOTE_SCTP_PORT 0
static GHashTable *associations = NULL;
G_LOCK_DEFINE_STATIC (associations_lock);
static guint32 number_of_associations = 0;
/* Interface implementations */
static void gst_sctp_association_finalize (GObject * object);
static void gst_sctp_association_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec);
static void gst_sctp_association_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
static struct socket *create_sctp_socket (GstSctpAssociation *
gst_sctp_association);
static struct sockaddr_conn get_sctp_socket_address (GstSctpAssociation *
gst_sctp_association, guint16 port);
static gpointer connection_thread_func (GstSctpAssociation * self);
static gboolean client_role_connect (GstSctpAssociation * self);
static int sctp_packet_out (void *addr, void *buffer, size_t length, guint8 tos,
guint8 set_df);
static int receive_cb (struct socket *sock, union sctp_sockstore addr,
void *data, size_t datalen, struct sctp_rcvinfo rcv_info, gint flags,
void *ulp_info);
static void handle_notification (GstSctpAssociation * self,
const union sctp_notification *notification, size_t length);
static void handle_association_changed (GstSctpAssociation * self,
const struct sctp_assoc_change *sac);
static void handle_stream_reset_event (GstSctpAssociation * self,
const struct sctp_stream_reset_event *ssr);
static void handle_message (GstSctpAssociation * self, guint8 * data,
guint32 datalen, guint16 stream_id, guint32 ppid);
static void maybe_set_state_to_ready (GstSctpAssociation * self);
static void gst_sctp_association_change_state (GstSctpAssociation * self,
GstSctpAssociationState new_state, gboolean notify);
static void
gst_sctp_association_class_init (GstSctpAssociationClass * klass)
{
GObjectClass *gobject_class;
gobject_class = (GObjectClass *) klass;
gobject_class->finalize = gst_sctp_association_finalize;
gobject_class->set_property = gst_sctp_association_set_property;
gobject_class->get_property = gst_sctp_association_get_property;
signals[SIGNAL_STREAM_RESET] =
g_signal_new ("stream-reset", G_OBJECT_CLASS_TYPE (klass),
G_SIGNAL_RUN_FIRST, G_STRUCT_OFFSET (GstSctpAssociationClass,
on_sctp_stream_reset), NULL, NULL, g_cclosure_marshal_generic,
G_TYPE_NONE, 1, G_TYPE_UINT);
properties[PROP_ASSOCIATION_ID] = g_param_spec_uint ("association-id",
"The SCTP association-id", "The SCTP association-id.", 0, G_MAXUSHORT,
DEFAULT_LOCAL_SCTP_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
properties[PROP_LOCAL_PORT] = g_param_spec_uint ("local-port", "Local SCTP",
"The local SCTP port for this association", 0, G_MAXUSHORT,
DEFAULT_LOCAL_SCTP_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
properties[PROP_REMOTE_PORT] =
g_param_spec_uint ("remote-port", "Remote SCTP",
"The remote SCTP port for this association", 0, G_MAXUSHORT,
DEFAULT_LOCAL_SCTP_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
properties[PROP_STATE] = g_param_spec_enum ("state", "SCTP Association state",
"The state of the SCTP association", GST_SCTP_ASSOCIATION_STATE_TYPE,
GST_SCTP_ASSOCIATION_STATE_NEW,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
properties[PROP_USE_SOCK_STREAM] =
g_param_spec_boolean ("use-sock-stream", "Use sock-stream",
"When set to TRUE, a sequenced, reliable, connection-based connection is used."
"When TRUE the partial reliability parameters of the channel is ignored.",
FALSE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS);
g_object_class_install_properties (gobject_class, NUM_PROPERTIES, properties);
}
static void
gst_sctp_association_init (GstSctpAssociation * self)
{
/* No need to lock mutex here as long as the function is only called from gst_sctp_association_get */
if (number_of_associations == 0) {
usrsctp_init (0, sctp_packet_out, g_print);
/* Explicit Congestion Notification */
usrsctp_sysctl_set_sctp_ecn_enable (0);
usrsctp_sysctl_set_sctp_nr_outgoing_streams_default
(DEFAULT_NUMBER_OF_SCTP_STREAMS);
}
number_of_associations++;
self->local_port = DEFAULT_LOCAL_SCTP_PORT;
self->remote_port = DEFAULT_REMOTE_SCTP_PORT;
self->sctp_ass_sock = NULL;
self->connection_thread = NULL;
g_mutex_init (&self->association_mutex);
self->state = GST_SCTP_ASSOCIATION_STATE_NEW;
self->use_sock_stream = FALSE;
usrsctp_register_address ((void *) self);
}
static void
gst_sctp_association_finalize (GObject * object)
{
GstSctpAssociation *self = GST_SCTP_ASSOCIATION (object);
G_LOCK (associations_lock);
g_hash_table_remove (associations, GUINT_TO_POINTER (self->association_id));
usrsctp_deregister_address ((void *) self);
number_of_associations--;
if (number_of_associations == 0) {
usrsctp_finish ();
}
G_UNLOCK (associations_lock);
g_thread_join (self->connection_thread);
G_OBJECT_CLASS (gst_sctp_association_parent_class)->finalize (object);
}
static void
gst_sctp_association_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec)
{
GstSctpAssociation *self = GST_SCTP_ASSOCIATION (object);
g_mutex_lock (&self->association_mutex);
if (self->state != GST_SCTP_ASSOCIATION_STATE_NEW) {
switch (prop_id) {
case PROP_LOCAL_PORT:
case PROP_REMOTE_PORT:
g_warning ("These properties cannot be set in this state");
goto error;
}
}
switch (prop_id) {
case PROP_ASSOCIATION_ID:
self->association_id = g_value_get_uint (value);
break;
case PROP_LOCAL_PORT:
self->local_port = g_value_get_uint (value);
break;
case PROP_REMOTE_PORT:
self->remote_port = g_value_get_uint (value);
break;
case PROP_STATE:
self->state = g_value_get_enum (value);
break;
case PROP_USE_SOCK_STREAM:
self->use_sock_stream = g_value_get_boolean (value);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (self, prop_id, pspec);
break;
}
g_mutex_unlock (&self->association_mutex);
if (prop_id == PROP_LOCAL_PORT || prop_id == PROP_REMOTE_PORT)
maybe_set_state_to_ready (self);
return;
error:
g_mutex_unlock (&self->association_mutex);
}
static void
maybe_set_state_to_ready (GstSctpAssociation * self)
{
gboolean signal_ready_state = FALSE;
g_mutex_lock (&self->association_mutex);
if ((self->state == GST_SCTP_ASSOCIATION_STATE_NEW) &&
(self->local_port != 0 && self->remote_port != 0)
&& (self->packet_out_cb != NULL) && (self->packet_received_cb != NULL)) {
signal_ready_state = TRUE;
gst_sctp_association_change_state (self, GST_SCTP_ASSOCIATION_STATE_READY,
FALSE);
}
g_mutex_unlock (&self->association_mutex);
/* The reason the state is changed twice is that we do not want to change state with
* notification while the association_mutex is locked. If someone listens
* on property change and call this object a deadlock might occur.*/
if (signal_ready_state)
gst_sctp_association_change_state (self, GST_SCTP_ASSOCIATION_STATE_READY,
TRUE);
}
static void
gst_sctp_association_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec)
{
GstSctpAssociation *self = GST_SCTP_ASSOCIATION (object);
switch (prop_id) {
case PROP_ASSOCIATION_ID:
g_value_set_uint (value, self->association_id);
break;
case PROP_LOCAL_PORT:
g_value_set_uint (value, self->local_port);
break;
case PROP_REMOTE_PORT:
g_value_set_uint (value, self->remote_port);
break;
case PROP_STATE:
g_value_set_enum (value, self->state);
break;
case PROP_USE_SOCK_STREAM:
g_value_set_boolean (value, self->use_sock_stream);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (self, prop_id, pspec);
break;
}
}
/* Public functions */
GstSctpAssociation *
gst_sctp_association_get (guint32 association_id)
{
GstSctpAssociation *association;
G_LOCK (associations_lock);
if (!associations) {
associations =
g_hash_table_new_full (g_direct_hash, g_direct_equal, NULL, NULL);
}
association =
g_hash_table_lookup (associations, GUINT_TO_POINTER (association_id));
if (!association) {
association =
g_object_new (GST_SCTP_TYPE_ASSOCIATION, "association-id",
association_id, NULL);
g_hash_table_insert (associations, GUINT_TO_POINTER (association_id),
association);
} else {
g_object_ref (association);
}
G_UNLOCK (associations_lock);
return association;
}
gboolean
gst_sctp_association_start (GstSctpAssociation * self)
{
gchar *thread_name;
g_mutex_lock (&self->association_mutex);
if (self->state != GST_SCTP_ASSOCIATION_STATE_READY) {
g_warning ("SCTP association is in wrong state and cannot be started");
goto configure_required;
}
if ((self->sctp_ass_sock = create_sctp_socket (self)) == NULL)
goto error;
gst_sctp_association_change_state (self,
GST_SCTP_ASSOCIATION_STATE_CONNECTING, FALSE);
g_mutex_unlock (&self->association_mutex);
/* The reason the state is changed twice is that we do not want to change state with
* notification while the association_mutex is locked. If someone listens
* on property change and call this object a deadlock might occur.*/
gst_sctp_association_change_state (self,
GST_SCTP_ASSOCIATION_STATE_CONNECTING, TRUE);
thread_name = g_strdup_printf ("connection_thread_%u", self->association_id);
self->connection_thread = g_thread_new (thread_name,
(GThreadFunc) connection_thread_func, self);
g_free (thread_name);
return TRUE;
error:
g_mutex_unlock (&self->association_mutex);
gst_sctp_association_change_state (self, GST_SCTP_ASSOCIATION_STATE_ERROR,
TRUE);
configure_required:
g_mutex_unlock (&self->association_mutex);
return FALSE;
}
void
gst_sctp_association_set_on_packet_out (GstSctpAssociation * self,
GstSctpAssociationPacketOutCb packet_out_cb, gpointer user_data)
{
g_return_if_fail (GST_SCTP_IS_ASSOCIATION (self));
g_mutex_lock (&self->association_mutex);
if (self->state == GST_SCTP_ASSOCIATION_STATE_NEW) {
self->packet_out_cb = packet_out_cb;
self->packet_out_user_data = user_data;
} else {
/* This is to be thread safe. The Association might try to write to the closure already */
g_warning ("It is not possible to change packet callback in this state");
}
g_mutex_unlock (&self->association_mutex);
maybe_set_state_to_ready (self);
}
void
gst_sctp_association_set_on_packet_received (GstSctpAssociation * self,
GstSctpAssociationPacketReceivedCb packet_received_cb, gpointer user_data)
{
g_return_if_fail (GST_SCTP_IS_ASSOCIATION (self));
g_mutex_lock (&self->association_mutex);
if (self->state == GST_SCTP_ASSOCIATION_STATE_NEW) {
self->packet_received_cb = packet_received_cb;
self->packet_received_user_data = user_data;
} else {
/* This is to be thread safe. The Association might try to write to the closure already */
g_warning ("It is not possible to change receive callback in this state");
}
g_mutex_unlock (&self->association_mutex);
maybe_set_state_to_ready (self);
}
void
gst_sctp_association_incoming_packet (GstSctpAssociation * self, guint8 * buf,
guint32 length)
{
usrsctp_conninput ((void *) self, (const void *) buf, (size_t) length, 0);
}
gboolean
gst_sctp_association_send_data (GstSctpAssociation * self, guint8 * buf,
guint32 length, guint16 stream_id, guint32 ppid, gboolean ordered,
GstSctpAssociationPartialReliability pr, guint32 reliability_param)
{
struct sctp_sendv_spa spa;
gint32 bytes_sent;
gboolean result = FALSE;
struct sockaddr_conn remote_addr;
g_mutex_lock (&self->association_mutex);
if (self->state != GST_SCTP_ASSOCIATION_STATE_CONNECTED)
goto end;
memset (&spa, 0, sizeof (spa));
spa.sendv_sndinfo.snd_ppid = g_htonl (ppid);
spa.sendv_sndinfo.snd_sid = stream_id;
spa.sendv_sndinfo.snd_flags = ordered ? 0 : SCTP_UNORDERED;
spa.sendv_sndinfo.snd_context = 0;
spa.sendv_sndinfo.snd_assoc_id = 0;
spa.sendv_flags = SCTP_SEND_SNDINFO_VALID;
if (pr != GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_NONE) {
spa.sendv_flags |= SCTP_SEND_PRINFO_VALID;
spa.sendv_prinfo.pr_value = g_htonl (reliability_param);
if (pr == GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_TTL)
spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_TTL;
else if (pr == GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_RTX)
spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_RTX;
else if (pr == GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_BUF)
spa.sendv_prinfo.pr_policy = SCTP_PR_SCTP_BUF;
}
remote_addr = get_sctp_socket_address (self, self->remote_port);
bytes_sent =
usrsctp_sendv (self->sctp_ass_sock, buf, length,
(struct sockaddr *) &remote_addr, 1, (void *) &spa,
(socklen_t) sizeof (struct sctp_sendv_spa), SCTP_SENDV_SPA, 0);
if (bytes_sent < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
/* Resending this buffer is taken care of by the gstsctpenc */
goto end;
} else {
g_warning ("Error sending data on stream %u: (%u) %s", stream_id, errno,
strerror (errno));
goto end;
}
}
result = TRUE;
end:
g_mutex_unlock (&self->association_mutex);
return result;
}
void
gst_sctp_association_reset_stream (GstSctpAssociation * self, guint16 stream_id)
{
struct sctp_reset_streams *srs;
socklen_t length;
length = (socklen_t) (sizeof (struct sctp_reset_streams) + sizeof (guint16));
srs = (struct sctp_reset_streams *) g_malloc0 (length);
srs->srs_flags = SCTP_STREAM_RESET_OUTGOING;
srs->srs_number_streams = 1;
srs->srs_stream_list[0] = stream_id;
g_mutex_lock (&self->association_mutex);
usrsctp_setsockopt (self->sctp_ass_sock, IPPROTO_SCTP, SCTP_RESET_STREAMS,
srs, length);
g_mutex_unlock (&self->association_mutex);
g_free (srs);
}
void
gst_sctp_association_force_close (GstSctpAssociation * self)
{
g_mutex_lock (&self->association_mutex);
if (self->sctp_ass_sock) {
usrsctp_close (self->sctp_ass_sock);
self->sctp_ass_sock = NULL;
}
g_mutex_unlock (&self->association_mutex);
}
static struct socket *
create_sctp_socket (GstSctpAssociation * self)
{
struct socket *sock;
struct linger l;
struct sctp_event event;
struct sctp_assoc_value stream_reset;
int value = 1;
guint16 event_types[] = {
SCTP_ASSOC_CHANGE,
SCTP_PEER_ADDR_CHANGE,
SCTP_REMOTE_ERROR,
SCTP_SEND_FAILED,
SCTP_SHUTDOWN_EVENT,
SCTP_ADAPTATION_INDICATION,
/*SCTP_PARTIAL_DELIVERY_EVENT, */
/*SCTP_AUTHENTICATION_EVENT, */
SCTP_STREAM_RESET_EVENT,
/*SCTP_SENDER_DRY_EVENT, */
/*SCTP_NOTIFICATIONS_STOPPED_EVENT, */
/*SCTP_ASSOC_RESET_EVENT, */
SCTP_STREAM_CHANGE_EVENT
};
guint32 i;
guint sock_type = self->use_sock_stream ? SOCK_STREAM : SOCK_SEQPACKET;
if ((sock =
usrsctp_socket (AF_CONN, sock_type, IPPROTO_SCTP, receive_cb, NULL, 0,
(void *) self)) == NULL)
goto error;
if (usrsctp_set_non_blocking (sock, 1) < 0) {
g_warning ("Could not set non-blocking mode on SCTP socket");
goto error;
}
memset (&l, 0, sizeof (l));
l.l_onoff = 1;
l.l_linger = 0;
if (usrsctp_setsockopt (sock, SOL_SOCKET, SO_LINGER, (const void *) &l,
(socklen_t) sizeof (struct linger)) < 0) {
g_warning ("Could not set SO_LINGER on SCTP socket");
goto error;
}
if (usrsctp_setsockopt (sock, IPPROTO_SCTP, SCTP_NODELAY, &value,
sizeof (int))) {
g_warning ("Could not set SCTP_NODELAY");
goto error;
}
memset (&stream_reset, 0, sizeof (stream_reset));
stream_reset.assoc_id = SCTP_ALL_ASSOC;
stream_reset.assoc_value = 1;
if (usrsctp_setsockopt (sock, IPPROTO_SCTP, SCTP_ENABLE_STREAM_RESET,
&stream_reset, sizeof (stream_reset))) {
g_warning ("Could not set SCTP_ENABLE_STREAM_RESET");
goto error;
}
memset (&event, 0, sizeof (event));
event.se_assoc_id = SCTP_ALL_ASSOC;
event.se_on = 1;
for (i = 0; i < sizeof (event_types) / sizeof (event_types[0]); i++) {
event.se_type = event_types[i];
if (usrsctp_setsockopt (sock, IPPROTO_SCTP, SCTP_EVENT,
&event, sizeof (event)) < 0) {
g_warning ("Failed to register event %u", event_types[i]);
}
}
return sock;
error:
if (sock) {
usrsctp_close (sock);
g_warning ("Could not create socket. Error: (%u) %s", errno,
strerror (errno));
errno = 0;
sock = NULL;
}
return NULL;
}
static struct sockaddr_conn
get_sctp_socket_address (GstSctpAssociation * gst_sctp_association,
guint16 port)
{
struct sockaddr_conn addr;
memset ((void *) &addr, 0, sizeof (struct sockaddr_conn));
#ifdef __APPLE__
addr.sconn_len = sizeof (struct sockaddr_conn);
#endif
addr.sconn_family = AF_CONN;
addr.sconn_port = g_htons (port);
addr.sconn_addr = (void *) gst_sctp_association;
return addr;
}
static gpointer
connection_thread_func (GstSctpAssociation * self)
{
/* TODO: Support both server and client role */
client_role_connect (self);
return NULL;
}
static gboolean
client_role_connect (GstSctpAssociation * self)
{
struct sockaddr_conn addr;
gint ret;
g_mutex_lock (&self->association_mutex);
addr = get_sctp_socket_address (self, self->local_port);
ret =
usrsctp_bind (self->sctp_ass_sock, (struct sockaddr *) &addr,
sizeof (struct sockaddr_conn));
if (ret < 0) {
g_warning ("usrsctp_bind() error: (%u) %s", errno, strerror (errno));
goto error;
}
addr = get_sctp_socket_address (self, self->remote_port);
ret =
usrsctp_connect (self->sctp_ass_sock, (struct sockaddr *) &addr,
sizeof (struct sockaddr_conn));
if (ret < 0 && errno != EINPROGRESS) {
g_warning ("usrsctp_connect() error: (%u) %s", errno, strerror (errno));
goto error;
}
g_mutex_unlock (&self->association_mutex);
return TRUE;
error:
g_mutex_unlock (&self->association_mutex);
return FALSE;
}
static int
sctp_packet_out (void *addr, void *buffer, size_t length, guint8 tos,
guint8 set_df)
{
GstSctpAssociation *self = GST_SCTP_ASSOCIATION (addr);
if (self->packet_out_cb) {
self->packet_out_cb (self, buffer, length, self->packet_out_user_data);
}
return 0;
}
static int
receive_cb (struct socket *sock, union sctp_sockstore addr, void *data,
size_t datalen, struct sctp_rcvinfo rcv_info, gint flags, void *ulp_info)
{
GstSctpAssociation *self = GST_SCTP_ASSOCIATION (ulp_info);
if (!data) {
/* Not sure if this can happend. */
g_warning ("Received empty data buffer");
} else {
if (flags & MSG_NOTIFICATION) {
handle_notification (self, (const union sctp_notification *) data,
datalen);
free (data);
} else {
handle_message (self, data, datalen, rcv_info.rcv_sid,
ntohl (rcv_info.rcv_ppid));
}
}
return 1;
}
static void
handle_notification (GstSctpAssociation * self,
const union sctp_notification *notification, size_t length)
{
g_assert (notification->sn_header.sn_length == length);
switch (notification->sn_header.sn_type) {
case SCTP_ASSOC_CHANGE:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_ASSOC_CHANGE");
handle_association_changed (self, &notification->sn_assoc_change);
break;
case SCTP_PEER_ADDR_CHANGE:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_PEER_ADDR_CHANGE");
break;
case SCTP_REMOTE_ERROR:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_REMOTE_ERROR");
break;
case SCTP_SEND_FAILED:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_SEND_FAILED");
break;
case SCTP_SHUTDOWN_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_SHUTDOWN_EVENT");
break;
case SCTP_ADAPTATION_INDICATION:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO,
"Event: SCTP_ADAPTATION_INDICATION");
break;
case SCTP_PARTIAL_DELIVERY_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO,
"Event: SCTP_PARTIAL_DELIVERY_EVENT");
break;
case SCTP_AUTHENTICATION_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO,
"Event: SCTP_AUTHENTICATION_EVENT");
break;
case SCTP_STREAM_RESET_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_STREAM_RESET_EVENT");
handle_stream_reset_event (self, &notification->sn_strreset_event);
break;
case SCTP_SENDER_DRY_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_SENDER_DRY_EVENT");
break;
case SCTP_NOTIFICATIONS_STOPPED_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO,
"Event: SCTP_NOTIFICATIONS_STOPPED_EVENT");
break;
case SCTP_ASSOC_RESET_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_ASSOC_RESET_EVENT");
break;
case SCTP_STREAM_CHANGE_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_STREAM_CHANGE_EVENT");
break;
case SCTP_SEND_FAILED_EVENT:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "Event: SCTP_SEND_FAILED_EVENT");
break;
default:
break;
}
}
static void
handle_association_changed (GstSctpAssociation * self,
const struct sctp_assoc_change *sac)
{
gboolean change_state = FALSE;
GstSctpAssociationState new_state;
switch (sac->sac_state) {
case SCTP_COMM_UP:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "SCTP_COMM_UP()");
g_mutex_lock (&self->association_mutex);
if (self->state == GST_SCTP_ASSOCIATION_STATE_CONNECTING) {
change_state = TRUE;
new_state = GST_SCTP_ASSOCIATION_STATE_CONNECTED;
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO, "SCTP association connected!");
} else if (self->state == GST_SCTP_ASSOCIATION_STATE_CONNECTED) {
g_warning ("SCTP association already open");
} else {
g_warning ("SCTP association in unexpected state");
}
g_mutex_unlock (&self->association_mutex);
break;
case SCTP_COMM_LOST:
g_warning ("SCTP event SCTP_COMM_LOST received");
/* TODO: Tear down association and signal that this has happend */
break;
case SCTP_RESTART:
g_log (G_LOG_DOMAIN, G_LOG_LEVEL_INFO,
"SCTP event SCTP_RESTART received");
break;
case SCTP_SHUTDOWN_COMP:
g_warning ("SCTP event SCTP_SHUTDOWN_COMP received");
/* TODO: Tear down association and signal that this has happend */
break;
case SCTP_CANT_STR_ASSOC:
g_warning ("SCTP event SCTP_CANT_STR_ASSOC received");
break;
}
if (change_state)
gst_sctp_association_change_state (self, new_state, TRUE);
}
static void
handle_stream_reset_event (GstSctpAssociation * self,
const struct sctp_stream_reset_event *sr)
{
guint32 i, n;
if (!(sr->strreset_flags & SCTP_STREAM_RESET_DENIED) &&
!(sr->strreset_flags & SCTP_STREAM_RESET_DENIED)) {
n = (sr->strreset_length -
sizeof (struct sctp_stream_reset_event)) / sizeof (uint16_t);
for (i = 0; i < n; i++) {
if (sr->strreset_flags & SCTP_STREAM_RESET_INCOMING_SSN) {
g_signal_emit (self, signals[SIGNAL_STREAM_RESET], 0,
sr->strreset_stream_list[i]);
}
}
}
}
static void
handle_message (GstSctpAssociation * self, guint8 * data, guint32 datalen,
guint16 stream_id, guint32 ppid)
{
if (self->packet_received_cb) {
self->packet_received_cb (self, data, datalen, stream_id, ppid,
self->packet_received_user_data);
}
}
static void
gst_sctp_association_change_state (GstSctpAssociation * self,
GstSctpAssociationState new_state, gboolean notify)
{
self->state = new_state;
if (notify)
g_object_notify_by_pspec (G_OBJECT (self), properties[PROP_STATE]);
}

123
ext/sctp/sctpassociation.h Normal file
View file

@ -0,0 +1,123 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifndef __GST_SCTP_ASSOCIATION_H__
#define __GST_SCTP_ASSOCIATION_H__
#include <glib-object.h>
#define INET
#define INET6
#include <usrsctp.h>
G_BEGIN_DECLS
#define GST_SCTP_TYPE_ASSOCIATION (gst_sctp_association_get_type ())
#define GST_SCTP_ASSOCIATION(obj) (G_TYPE_CHECK_INSTANCE_CAST ((obj), GST_SCTP_TYPE_ASSOCIATION, GstSctpAssociation))
#define GST_SCTP_IS_ASSOCIATION(obj) (G_TYPE_CHECK_INSTANCE_TYPE ((obj), GST_SCTP_TYPE_ASSOCIATION))
#define GST_SCTP_ASSOCIATION_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST ((klass), GST_SCTP_TYPE_ASSOCIATION, GstSctpAssociationClass))
#define GST_SCTP_IS_ASSOCIATION_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE ((klass), GST_SCTP_TYPE_ASSOCIATION))
#define GST_SCTP_ASSOCIATION_GET_CLASS(obj) (G_TYPE_INSTANCE_GET_CLASS ((obj), GST_SCTP_TYPE_ASSOCIATION, GstSctpAssociationClass))
typedef struct _GstSctpAssociation GstSctpAssociation;
typedef struct _GstSctpAssociationClass GstSctpAssociationClass;
typedef enum
{
GST_SCTP_ASSOCIATION_STATE_NEW,
GST_SCTP_ASSOCIATION_STATE_READY,
GST_SCTP_ASSOCIATION_STATE_CONNECTING,
GST_SCTP_ASSOCIATION_STATE_CONNECTED,
GST_SCTP_ASSOCIATION_STATE_DISCONNECTING,
GST_SCTP_ASSOCIATION_STATE_DISCONNECTED,
GST_SCTP_ASSOCIATION_STATE_ERROR
} GstSctpAssociationState;
typedef enum
{
GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_NONE = 0x0000,
GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_TTL = 0x0001,
GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_BUF = 0x0002,
GST_SCTP_ASSOCIATION_PARTIAL_RELIABILITY_RTX = 0x0003
} GstSctpAssociationPartialReliability;
typedef void (*GstSctpAssociationPacketReceivedCb) (GstSctpAssociation *
sctp_association, guint8 * data, gsize length, guint16 stream_id,
guint ppid, gpointer user_data);
typedef void (*GstSctpAssociationPacketOutCb) (GstSctpAssociation *
sctp_association, const guint8 * data, gsize length, gpointer user_data);
struct _GstSctpAssociation
{
GObject parent_instance;
guint32 association_id;
guint16 local_port;
guint16 remote_port;
gboolean use_sock_stream;
struct socket *sctp_ass_sock;
GMutex association_mutex;
GstSctpAssociationState state;
GThread *connection_thread;
GstSctpAssociationPacketReceivedCb packet_received_cb;
gpointer packet_received_user_data;
GstSctpAssociationPacketOutCb packet_out_cb;
gpointer packet_out_user_data;
};
struct _GstSctpAssociationClass
{
GObjectClass parent_class;
void (*on_sctp_stream_reset) (GstSctpAssociation * sctp_association,
guint16 stream_id);
};
GType gst_sctp_association_get_type (void);
GstSctpAssociation *gst_sctp_association_get (guint32 association_id);
gboolean gst_sctp_association_start (GstSctpAssociation * self);
void gst_sctp_association_set_on_packet_out (GstSctpAssociation * self,
GstSctpAssociationPacketOutCb packet_out_cb, gpointer user_data);
void gst_sctp_association_set_on_packet_received (GstSctpAssociation * self,
GstSctpAssociationPacketReceivedCb packet_received_cb, gpointer user_data);
void gst_sctp_association_incoming_packet (GstSctpAssociation * self,
guint8 * buf, guint32 length);
gboolean gst_sctp_association_send_data (GstSctpAssociation * self,
guint8 * buf, guint32 length, guint16 stream_id, guint32 ppid,
gboolean ordered, GstSctpAssociationPartialReliability pr,
guint32 reliability_param);
void gst_sctp_association_reset_stream (GstSctpAssociation * self,
guint16 stream_id);
void gst_sctp_association_force_close (GstSctpAssociation * self);
G_END_DECLS
#endif /* __GST_SCTP_ASSOCIATION_H__ */

8
gst-libs/gst/Makefile.am
View file

@ -6,9 +6,13 @@ if USE_OPENCV
OPENCV_DIR=opencv
endif
if USE_SCTP
SCTP_DIR=sctp
endif
SUBDIRS = uridownloader adaptivedemux interfaces basecamerabinsrc codecparsers \
insertbin mpegts video audio player isoff webrtc $(WAYLAND_DIR) \
$(OPENCV_DIR)
$(OPENCV_DIR) $(SCTP_DIR)
noinst_HEADERS = gst-i18n-plugin.h gettext.h glib-compat-private.h
DIST_SUBDIRS = uridownloader adaptivedemux interfaces basecamerabinsrc \
@ -18,7 +22,7 @@ adaptivedemux: uridownloader
INDEPENDENT_SUBDIRS = \
interfaces basecamerabinsrc codecparsers insertbin uridownloader \
mpegts player isoff $(WAYLAND_DIR) $(OPENCV_DIR)
mpegts player isoff $(WAYLAND_DIR) $(OPENCV_DIR) $(SCTP_DIR)
.PHONY: independent-subdirs $(INDEPENDENT_SUBDIRS)

17
gst-libs/gst/sctp/Makefile.am Normal file
View file

@ -0,0 +1,17 @@
lib_LTLIBRARIES = libgstsctp-1.0.la
libgstsctp_1_0_la_SOURCES = \
sctpsendmeta.c \
sctpreceivemeta.c
libgstsctp_1_0_la_CFLAGS = \
$(GST_PLUGINS_BASE_CFLAGS) \
$(GST_BASE_CFLAGS) \
$(GST_CFLAGS)
libgstsctp_1_0_la_LIBADD = $(GST_LIBS) $(GST_BASE_LIBS)
libgstsctp_1_0_includedir = $(includedir)/gstreamer-1.0/gst/sctp
libgstsctp_1_0_include_HEADERS = \
sctpsendmeta.h \
sctpreceivemeta.h

90
gst-libs/gst/sctp/sctpreceivemeta.c Normal file
View file

@ -0,0 +1,90 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#include "sctpreceivemeta.h"
static gboolean gst_sctp_receive_meta_init (GstMeta * meta, gpointer params,
GstBuffer * buffer);
static gboolean gst_sctp_receive_meta_transform (GstBuffer * transbuf,
GstMeta * meta, GstBuffer * buffer, GQuark type, gpointer data);
GType
gst_sctp_receive_meta_api_get_type (void)
{
static const gchar *tags[] = { NULL };
static volatile GType type;
if (g_once_init_enter (&type)) {
GType _type = gst_meta_api_type_register ("GstSctpReceiveMetaAPI", tags);
g_once_init_leave (&type, _type);
}
return type;
}
const GstMetaInfo *
gst_sctp_receive_meta_get_info (void)
{
static const GstMetaInfo *gst_sctp_receive_meta_info = NULL;
if (g_once_init_enter (&gst_sctp_receive_meta_info)) {
const GstMetaInfo *meta = gst_meta_register (GST_SCTP_RECEIVE_META_API_TYPE,
"GstSctpReceiveMeta",
sizeof (GstSctpReceiveMeta),
gst_sctp_receive_meta_init,
(GstMetaFreeFunction) NULL,
gst_sctp_receive_meta_transform);
g_once_init_leave (&gst_sctp_receive_meta_info, meta);
}
return gst_sctp_receive_meta_info;
}
static gboolean
gst_sctp_receive_meta_init (GstMeta * meta, gpointer params, GstBuffer * buffer)
{
GstSctpReceiveMeta *gst_sctp_receive_meta = (GstSctpReceiveMeta *) meta;
gst_sctp_receive_meta->ppid = 0;
return TRUE;
}
static gboolean
gst_sctp_receive_meta_transform (GstBuffer * transbuf, GstMeta * meta,
GstBuffer * buffer, GQuark type, gpointer data)
{
GstSctpReceiveMeta *gst_sctp_receive_meta = (GstSctpReceiveMeta *) meta;
gst_sctp_buffer_add_receive_meta (transbuf, gst_sctp_receive_meta->ppid);
return TRUE;
}
GstSctpReceiveMeta *
gst_sctp_buffer_add_receive_meta (GstBuffer * buffer, guint32 ppid)
{
GstSctpReceiveMeta *gst_sctp_receive_meta = NULL;
g_return_val_if_fail (GST_IS_BUFFER (buffer), NULL);
gst_sctp_receive_meta =
(GstSctpReceiveMeta *) gst_buffer_add_meta (buffer,
GST_SCTP_RECEIVE_META_INFO, NULL);
gst_sctp_receive_meta->ppid = ppid;
return gst_sctp_receive_meta;
}

53
gst-libs/gst/sctp/sctpreceivemeta.h Normal file
View file

@ -0,0 +1,53 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifndef __GST_SCTP_RECEIVE_META_H__
#define __GST_SCTP_RECEIVE_META_H__
#include <gst/gst.h>
G_BEGIN_DECLS
#define GST_SCTP_RECEIVE_META_API_TYPE (gst_sctp_receive_meta_api_get_type())
#define GST_SCTP_RECEIVE_META_INFO (gst_sctp_receive_meta_get_info())
typedef struct _GstSctpReceiveMeta GstSctpReceiveMeta;
struct _GstSctpReceiveMeta
{
GstMeta meta;
guint32 ppid;
};
GType gst_sctp_receive_meta_api_get_type (void);
const GstMetaInfo *gst_sctp_receive_meta_get_info (void);
GstSctpReceiveMeta *gst_sctp_buffer_add_receive_meta (GstBuffer * buffer,
guint32 ppid);
#define gst_sctp_buffer_get_receive_meta(b) ((GstSctpReceiveMeta *)gst_buffer_get_meta((b), GST_SCTP_RECEIVE_META_TYPE))
G_END_DECLS
#endif /* __GST_SCTP_RECEIVE_META_H__ */

99
gst-libs/gst/sctp/sctpsendmeta.c Normal file
View file

@ -0,0 +1,99 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#include "sctpsendmeta.h"
static gboolean gst_sctp_send_meta_init (GstMeta * meta, gpointer params,
GstBuffer * buffer);
static gboolean gst_sctp_send_meta_transform (GstBuffer * transbuf,
GstMeta * meta, GstBuffer * buffer, GQuark type, gpointer data);
GType
gst_sctp_send_meta_api_get_type (void)
{
static const gchar *tags[] = { NULL };
static volatile GType type;
if (g_once_init_enter (&type)) {
GType _type = gst_meta_api_type_register ("GstSctpSendMetaAPI", tags);
g_once_init_leave (&type, _type);
}
return type;
}
const GstMetaInfo *
gst_sctp_send_meta_get_info (void)
{
static const GstMetaInfo *gst_sctp_send_meta_info = NULL;
if (g_once_init_enter (&gst_sctp_send_meta_info)) {
const GstMetaInfo *meta = gst_meta_register (GST_SCTP_SEND_META_API_TYPE,
"GstSctpSendMeta",
sizeof (GstSctpSendMeta),
gst_sctp_send_meta_init,
(GstMetaFreeFunction) NULL,
gst_sctp_send_meta_transform);
g_once_init_leave (&gst_sctp_send_meta_info, meta);
}
return gst_sctp_send_meta_info;
}
static gboolean
gst_sctp_send_meta_init (GstMeta * meta, gpointer params, GstBuffer * buffer)
{
GstSctpSendMeta *gst_sctp_send_meta = (GstSctpSendMeta *) meta;
gst_sctp_send_meta->ppid = 0;
gst_sctp_send_meta->ordered = TRUE;
gst_sctp_send_meta->pr = GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE;
gst_sctp_send_meta->pr_param = 0;
return TRUE;
}
static gboolean
gst_sctp_send_meta_transform (GstBuffer * transbuf, GstMeta * meta,
GstBuffer * buffer, GQuark type, gpointer data)
{
GstSctpSendMeta *gst_sctp_send_meta = (GstSctpSendMeta *) meta;
gst_sctp_buffer_add_send_meta (transbuf, gst_sctp_send_meta->ppid,
gst_sctp_send_meta->ordered, gst_sctp_send_meta->pr,
gst_sctp_send_meta->pr_param);
return TRUE;
}
GstSctpSendMeta *
gst_sctp_buffer_add_send_meta (GstBuffer * buffer, guint32 ppid,
gboolean ordered, GstSctpSendMetaPartiallyReliability pr, guint32 pr_param)
{
GstSctpSendMeta *gst_sctp_send_meta = NULL;
g_return_val_if_fail (GST_IS_BUFFER (buffer), NULL);
gst_sctp_send_meta =
(GstSctpSendMeta *) gst_buffer_add_meta (buffer, GST_SCTP_SEND_META_INFO,
NULL);
gst_sctp_send_meta->ppid = ppid;
gst_sctp_send_meta->ordered = ordered;
gst_sctp_send_meta->pr = pr;
gst_sctp_send_meta->pr_param = pr_param;
return gst_sctp_send_meta;
}

65
gst-libs/gst/sctp/sctpsendmeta.h Normal file
View file

@ -0,0 +1,65 @@
/*
* Copyright (c) 2015, Collabora Ltd.
*
* Redistribution and use in source and binary forms, with or without modification,
* are permitted provided that the following conditions are met:
*
* 1. Redistributions of source code must retain the above copyright notice, this
* list of conditions and the following disclaimer.
*
* 2. Redistributions in binary form must reproduce the above copyright notice, this
* list of conditions and the following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
* ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
* IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
* NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
* PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY
* OF SUCH DAMAGE.
*/
#ifndef __GST_SCTP_SEND_META_H__
#define __GST_SCTP_SEND_META_H__
#include <gst/gst.h>
G_BEGIN_DECLS
typedef enum
{
GST_SCTP_SEND_META_PARTIAL_RELIABILITY_NONE,
GST_SCTP_SEND_META_PARTIAL_RELIABILITY_TTL,
GST_SCTP_SEND_META_PARTIAL_RELIABILITY_BUF,
GST_SCTP_SEND_META_PARTIAL_RELIABILITY_RTX
} GstSctpSendMetaPartiallyReliability;
#define GST_SCTP_SEND_META_API_TYPE (gst_sctp_send_meta_api_get_type())
#define GST_SCTP_SEND_META_INFO (gst_sctp_send_meta_get_info())
typedef struct _GstSctpSendMeta GstSctpSendMeta;
struct _GstSctpSendMeta
{
GstMeta meta;
guint32 ppid;
gboolean ordered;
GstSctpSendMetaPartiallyReliability pr;
guint32 pr_param;
};
GType gst_sctp_send_meta_api_get_type (void);
const GstMetaInfo *gst_sctp_send_meta_get_info (void);
GstSctpSendMeta *gst_sctp_buffer_add_send_meta (GstBuffer * buffer,
guint32 ppid, gboolean ordered, GstSctpSendMetaPartiallyReliability pr,
guint32 pr_param);
#define gst_sctp_buffer_get_send_meta(b) ((GstSctpSendMeta *)gst_buffer_get_meta((b), GST_SCTP_SEND_META_TYPE))
G_END_DECLS
#endif /* __GST_SCTP_SEND_META_H__ */

8
pkgconfig/Makefile.am
View file

@ -20,6 +20,11 @@ pcverfiles_uninstalled = \
gstreamer-bad-audio-@GST_API_VERSION@-uninstalled.pc \
gstreamer-bad-video-@GST_API_VERSION@-uninstalled.pc
if USE_SCTP
pcverfiles += gstreamer-sctp-@GST_API_VERSION@.pc
pcverfiles_uninstalled += gstreamer-sctp-@GST_API_VERSION@-uninstalled.pc
endif
all-local: $(pcverfiles) $(pcverfiles_uninstalled)
cp_verbose = $(cp_verbose_$(V))
@ -57,7 +62,8 @@ pcinfiles = \
gstreamer-player.pc.in gstreamer-player-uninstalled.pc.in \
gstreamer-webrtc.pc.in gstreamer-webrtc-uninstalled.pc.in \
gstreamer-bad-audio.pc.in gstreamer-bad-audio-uninstalled.pc.in \
gstreamer-bad-video.pc.in gstreamer-bad-video-uninstalled.pc.in
gstreamer-bad-video.pc.in gstreamer-bad-video-uninstalled.pc.in \
gstreamer-sctp.pc.in gstreamer-sctp-uninstalled.pc.in
DISTCLEANFILES = $(pcinfiles:.in=)
EXTRA_DIST = $(pcinfiles)

12
pkgconfig/gstreamer-sctp-uninstalled.pc.in Normal file
View file

@ -0,0 +1,12 @@
prefix=
exec_prefix=
libdir=${pcfiledir}/../gst-libs/gst/sctp
includedir=${pcfiledir}/../gst-libs
Name: GStreamer SCTP Library
Description: SCTP helper functions, uninstalled
Requires: gstreamer-@GST_API_VERSION@
Version: @VERSION@
Libs: -L${libdir} ${libdir}/libgstsctp-@GST_API_VERSION@.la
Cflags: -I${includedir}

12
pkgconfig/gstreamer-sctp.pc.in Normal file
View file

@ -0,0 +1,12 @@
prefix=@prefix@
exec_prefix=@exec_prefix@
libdir=@libdir@
includedir=@includedir@/gstreamer-@GST_API_VERSION@
Name: GStreamer SCTP Library
Description: SCTP helper functions
Requires: gstreamer-@GST_API_VERSION@
Version: @VERSION@
Libs: -L${libdir} -lgstsctp-@GST_API_VERSION@
Cflags: -I${includedir}