mirrors/pipewire
1
0
Fork 0
mirror of https://gitlab.freedesktop.org/pipewire/pipewire.git synced 2025-11-02 09:01:50 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

pinossocketsink: add new socket sink

Browse source 
Add a new sink that replaces the pinospay ! multisocketsink element
pair. This would allow us to track the fds per client more closely.
This commit is contained in:
Wim Taymans 2016-04-27 12:05:45 +02:00
parent d5e333ac4b
commit c8f34750e2
9 changed files with 2588 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

5
pinos/Makefile.am
View file

@ -226,7 +226,9 @@ plugindir = $(libdir)/gstreamer-1.0
plugin_LTLIBRARIES = libgstpinos.la
libgstpinos_la_SOURCES = \
gst/gstburstcache.c \
gst/gstpinos.c \
gst/gstpinossocketsink.c \
gst/gstpinospay.c \
gst/gstpinosdepay.c \
gst/gstpinosdeviceprovider.c \
@ -239,7 +241,8 @@ libgstpinos_la_LIBADD = $(GST_BASE_LIBS) $(GST_LIBS) $(GLIB_LIBS) $(LIBM) -lgst
libpinos-@PINOS_MAJORMINOR@.la libpinoscore-@PINOS_MAJORMINOR@.la
libgstpinos_la_LIBTOOLFLAGS = $(GST_PLUGIN_LIBTOOLFLAGS)
noinst_HEADERS = gst/gstpinossrc.h gst/gstpinossink.h gst/gstpinospay.h \
noinst_HEADERS = gst/gstburstcache.h gst/gstpinossrc.h \
gst/gstpinossocketsink.h gst/gstpinossink.h gst/gstpinospay.h \
gst/gstpinosdepay.h gst/gstpinosdeviceprovider.h
###################################

1402
pinos/gst/gstburstcache.c Normal file
View file

File diff suppressed because it is too large Load diff

301
pinos/gst/gstburstcache.h Normal file
View file

@ -0,0 +1,301 @@
/* GStreamer
* Copyright (C) <2012> Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __GST_BURST_CACHE_H__
#define __GST_BURST_CACHE_H__
#include <gst/gst.h>
G_BEGIN_DECLS
#define GST_TYPE_BURST_CACHE \
(gst_burst_cache_get_type())
#define GST_BURST_CACHE(obj) \
(G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_BURST_CACHE,GstBurstCache))
#define GST_BURST_CACHE_CLASS(klass) \
(G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_BURST_CACHE,GstBurstCacheClass))
#define GST_IS_BURST_CACHE(obj) \
(G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_BURST_CACHE))
#define GST_IS_BURST_CACHE_CLASS(klass) \
(G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_BURST_CACHE))
#define GST_BURST_CACHE_GET_CLASS(klass) \
(G_TYPE_INSTANCE_GET_CLASS ((klass), GST_TYPE_BURST_CACHE, GstBurstCacheClass))
typedef struct _GstBurstCache GstBurstCache;
typedef struct _GstBurstCacheClass GstBurstCacheClass;
typedef struct _GstBurstCacheReader GstBurstCacheReader;
/**
* GstBurstCacheRecover:
* @GST_BURST_CACHE_RECOVER_NONE : no recovering is done
* @GST_BURST_CACHE_RECOVER_RESYNC_LATEST : reader is moved to last buffer
* @GST_BURST_CACHE_RECOVER_RESYNC_SOFT_LIMIT: reader is moved to the soft limit
* @GST_BURST_CACHE_RECOVER_RESYNC_KEYFRAME : reader is moved to latest keyframe
*
* Possible values for the recovery procedure to use when a reader consumes
* data too slowly and has a backlog of more that soft-limit buffers.
*/
typedef enum
{
GST_BURST_CACHE_RECOVER_NONE,
GST_BURST_CACHE_RECOVER_RESYNC_LATEST,
GST_BURST_CACHE_RECOVER_RESYNC_SOFT_LIMIT,
GST_BURST_CACHE_RECOVER_RESYNC_KEYFRAME
} GstBurstCacheRecover;
/**
* GstBurstCacheStart:
* @GST_BURST_CACHE_START_LATEST : reader receives most recent buffer
* @GST_BURST_CACHE_START_NEXT_KEYFRAME : reader receives next keyframe
* @GST_BURST_CACHE_START_LATEST_KEYFRAME : reader receives latest keyframe (burst)
* @GST_BURST_CACHE_START_BURST : reader receives specific amount of data
* @GST_BURST_CACHE_START_BURST_KEYFRAME : reader receives specific amount of data
* starting from latest keyframe
* @GST_BURST_CACHE_START_BURST_WITH_KEYFRAME : reader receives specific amount of data from
* a keyframe, or if there is not enough data after
* the keyframe, starting before the keyframe
*
* This enum defines the selection of the first buffer that is sent
* to a new reader.
*/
typedef enum
{
GST_BURST_CACHE_START_LATEST,
GST_BURST_CACHE_START_NEXT_KEYFRAME,
GST_BURST_CACHE_START_LATEST_KEYFRAME,
GST_BURST_CACHE_START_BURST,
GST_BURST_CACHE_START_BURST_KEYFRAME,
GST_BURST_CACHE_START_BURST_WITH_KEYFRAME
} GstBurstCacheStart;
/**
* GstBurstCacheError:
* @GST_BURST_CACHE_ERROR_NONE : No error
* @GST_BURST_CACHE_ERROR_SLOW : reader is too slow
* @GST_BURST_CACHE_ERROR_ERROR : reader is in error
* @GST_BURST_CACHE_ERROR_DUPLICATE: same reader added twice
*
* Error codes used in the reason GError.
*/
typedef enum
{
GST_BURST_CACHE_ERROR_NONE = 0,
GST_BURST_CACHE_ERROR_SLOW = 1,
GST_BURST_CACHE_ERROR_ERROR = 2,
GST_BURST_CACHE_ERROR_DUPLICATE = 3,
} GstBurstCacheError;
/**
* GstBurstCacheResult:
* @GST_BURST_CACHE_RESULT_ERROR : An error occured
* @GST_BURST_CACHE_RESULT_OK : No error
* @GST_BURST_CACHE_RESULT_WAIT : Wait for more buffers
* @GST_BURST_CACHE_RESULT_EOS : No more buffers
*
* Error codes used in the reason GError.
*/
typedef enum
{
GST_BURST_CACHE_RESULT_ERROR = 0,
GST_BURST_CACHE_RESULT_OK = 1,
GST_BURST_CACHE_RESULT_WAIT = 2,
GST_BURST_CACHE_RESULT_EOS = 3,
} GstBurstCacheResult;
/**
* GstBurstCacheReaderCallback:
* @cache: a #GstBurstCache
* @reader: a #GstBurstCacheReader
* @user_data: user data given when creating @reader
*
* Called when @reader in @cache has data. You can get the new data with
* gst_burst_cache_get_buffer() from this callback or any other thread.
*/
typedef void (*GstBurstCacheReaderCallback) (GstBurstCache *cache,
GstBurstCacheReader *reader,
gpointer user_data);
/**
* GstBurstCacheReader:
* @object: parent miniobject
* @bufpos: position of this reader in the global queue
* @draincount: the remaining number of buffers to drain or -1 if the
* reader is not draining.
* @new_reader: this is a new reader
* @discont: is the next buffer was discont
* @reason: the reason why the reader is being removed
*
* structure for a reader
*/
struct _GstBurstCacheReader {
GHook hook;
gint bufpos;
gint draincount;
GstBurstCacheReaderCallback callback;
gpointer user_data;
GDestroyNotify notify;
gboolean new_reader;
gboolean discont;
GError *reason;
/* method to sync reader when connecting */
GstBurstCacheStart start_method;
GstFormat min_format;
guint64 min_value;
GstFormat max_format;
guint64 max_value;
/* stats */
guint64 bytes_sent;
guint64 dropped_buffers;
guint64 avg_queue_size;
guint64 first_buffer_ts;
guint64 last_buffer_ts;
guint64 add_time;
guint64 remove_time;
guint64 last_activity_time;
guint64 timeout;
gchar debug[30]; /* a debug string used in debug calls to
identify the reader */
};
/**
* GstBurstCache:
* @parent: parent GObject
* @lock: lock to protect @readers
* @bufqueue: global queue of buffers
* @readers: list of readers we are serving
* @readers_cookie: Cookie to detect changes to @readers
* @limit_format: the format of @limit_max and @@limit_soft_max
* @limit_max: max units to queue for a reader
* @limit_soft_max: max units a reader can lag before recovery starts
* @recover: how to recover a lagging reader
* @bytes_min: min number of bytes to queue
* @time_min: min time to queue
* @buffers_min: min number of buffers to queue
* @bytes_to_serve: how much bytes we must serve
* @bytes_served: how much bytes have we served
* @bytes_queued: number of queued bytes
* @time_queued: amount of queued time
* @buffers_queued: number of queued buffers
*/
struct _GstBurstCache {
GObject parent;
/*< private >*/
GRecMutex lock;
GPtrArray *bufqueue;
/* the readers */
GHookList readers;
guint readers_cookie;
/* these values are used to check if a reader is reading fast
* enough and to control recovery */
GstFormat limit_format;
gint64 limit_max;
gint64 limit_soft_max;
GstBurstCacheRecover recover;
/* these values are used to control the amount of data
* kept in the queues. It allows readers to perform a burst
* on connect. */
gint bytes_min;
gint64 time_min;
gint buffers_min;
/* stats */
gint bytes_queued;
gint64 time_queued;
gint buffers_queued;
};
/**
* GstBurstCacheClass:
* @parent_class: parent GObjectClass
* @reader_ready: called when a reader has a new buffer available
*
* The GstBurstCache class structure.
*/
struct _GstBurstCacheClass {
GObjectClass parent_class;
};
GType gst_burst_cache_get_type (void);
GType gst_burst_cache_reader_get_type (void);
GstBurstCache * gst_burst_cache_new (guint reader_size);
void gst_burst_cache_set_min_amount (GstBurstCache *cache,
gint bytes_min,
gint64 time_min,
gint buffers_min);
void gst_burst_cache_get_min_amount (GstBurstCache *cache,
gint *bytes_min,
gint64 *time_min,
gint *buffers_min);
void gst_burst_cache_set_limits (GstBurstCache *cache,
GstFormat format,
gint64 max,
gint64 soft_max,
GstBurstCacheRecover recover);
void gst_burst_cache_get_limits (GstBurstCache *cache,
GstFormat *format,
gint64 *max,
gint64 *soft_max,
GstBurstCacheRecover *recover);
void gst_burst_cache_queue_buffer (GstBurstCache *cache,
GstBuffer *buffer);
GstBurstCacheReader * gst_burst_cache_reader_new (GstBurstCache *cache,
GstBurstCacheReaderCallback callback,
gpointer user_data,
GDestroyNotify notify);
gboolean gst_burst_cache_reader_set_burst (GstBurstCacheReader *reader,
GstBurstCacheStart start_method,
GstFormat min_format, guint64 min_value,
GstFormat max_format, guint64 max_value);
void gst_burst_cache_reader_destroy (GstBurstCacheReader *reader);
gboolean gst_burst_cache_add_reader (GstBurstCache *cache,
GstBurstCacheReader *reader);
gboolean gst_burst_cache_remove_reader (GstBurstCache *cache,
GstBurstCacheReader *reader,
gboolean drain);
gboolean gst_burst_cache_error_reader (GstBurstCache *cache,
GstBurstCacheReader *reader,
GError *error);
void gst_burst_cache_clear_readers (GstBurstCache * cache);
GstBurstCacheResult gst_burst_cache_get_buffer (GstBurstCache *cache,
GstBurstCacheReader *reader,
GstBuffer **buffer);
G_END_DECLS
#endif /* __GST_BURST_CACHE_H__ */

3
pinos/gst/gstpinos.c
View file

@ -32,6 +32,7 @@
#include "config.h"
#endif
#include "gstpinossocketsink.h"
#include "gstpinossrc.h"
#include "gstpinossink.h"
#include "gstpinosdeviceprovider.h"
@ -51,6 +52,8 @@ plugin_init (GstPlugin * plugin)
GST_TYPE_PINOS_SRC);
gst_element_register (plugin, "pinossink", GST_RANK_NONE,
GST_TYPE_PINOS_SINK);
gst_element_register (plugin, "pinossocketsink", GST_RANK_NONE,
GST_TYPE_PINOS_SOCKET_SINK);
if (!gst_device_provider_register (plugin, "pinosdeviceprovider",
GST_RANK_PRIMARY + 1, GST_TYPE_PINOS_DEVICE_PROVIDER))

7
pinos/gst/gstpinossink.c
View file

@ -152,7 +152,6 @@ gst_pinos_sink_class_init (GstPinosSinkClass * klass)
gst_element_class_add_pad_template (gstelement_class,
gst_static_pad_template_get (&gst_pinos_sink_template));
gstbasesink_class->get_caps = gst_pinos_sink_getcaps;
gstbasesink_class->set_caps = gst_pinos_sink_setcaps;
gstbasesink_class->fixate = gst_pinos_sink_sink_fixate;
gstbasesink_class->propose_allocation = gst_pinos_sink_propose_allocation;
@ -338,12 +337,6 @@ on_stream_notify (GObject *gobject,
pinos_main_loop_signal (pinossink->loop, FALSE);
}
static GstCaps *
gst_pinos_sink_getcaps (GstBaseSink * bsink, GstCaps * filter)
{
return GST_BASE_SINK_CLASS (parent_class)->get_caps (bsink, filter);
}
static gboolean
gst_pinos_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
{

798
pinos/gst/gstpinossocketsink.c Normal file
View file

@ -0,0 +1,798 @@
/* GStreamer
* Copyright (C) <2016> Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
/**
* SECTION:element-pinossocketsink
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <string.h>
#include <stdlib.h>
#include <fcntl.h>
#include <sys/socket.h>
#include <unistd.h>
#include <gio/gunixfdmessage.h>
#include <gst/allocators/gstfdmemory.h>
#include <gst/net/gstnetcontrolmessagemeta.h>
#include "gstpinossocketsink.h"
#include "gsttmpfileallocator.h"
typedef struct _MyReader MyReader;
typedef struct _MySource MySource;
struct _MyReader {
GstBurstCacheReader reader;
GSocket *socket;
MySource *source;
guint id;
};
struct _MySource {
GSource source;
GIOCondition condition;
gpointer tag;
MyReader *reader;
};
typedef gboolean (*MyReaderSourceFunc) (MyReader *reader, GIOCondition condition, gpointer user_data);
static gboolean
mysource_dispatch (GSource *source,
GSourceFunc callback,
gpointer user_data)
{
MyReaderSourceFunc func = (MyReaderSourceFunc)callback;
MySource *mysource = (MySource *)source;
MyReader *myreader = mysource->reader;
guint events;
gboolean ret;
events = g_source_query_unix_fd (source, mysource->tag);
ret = (*func) (myreader, events, user_data);
return ret;
}
static GSourceFuncs mysource_funcs =
{
NULL, NULL, /* check, prepare */
mysource_dispatch,
NULL, /* finalize */
NULL,
NULL,
};
static GQuark fdids_quark;
static GQuark orig_buffer_quark;
GST_DEBUG_CATEGORY_STATIC (pinos_socket_sink_debug);
#define GST_CAT_DEFAULT pinos_socket_sink_debug
static GstStaticPadTemplate gst_pinos_socket_sink_template =
GST_STATIC_PAD_TEMPLATE ("sink",
GST_PAD_SINK,
GST_PAD_ALWAYS,
GST_STATIC_CAPS_ANY
);
enum
{
PROP_0,
PROP_NUM_HANDLES,
};
/* PinosSocketSink signals and args */
enum
{
/* methods */
SIGNAL_ADD,
SIGNAL_REMOVE,
LAST_SIGNAL
};
static guint gst_pinos_socket_sink_signals[LAST_SIGNAL] = { 0 };
#define gst_pinos_socket_sink_parent_class parent_class
G_DEFINE_TYPE (GstPinosSocketSink, gst_pinos_socket_sink, GST_TYPE_BASE_SINK);
static gboolean
gst_pinos_socket_sink_propose_allocation (GstBaseSink * bsink, GstQuery * query)
{
GstPinosSocketSink *this = GST_PINOS_SOCKET_SINK (bsink);
gst_query_add_allocation_param (query, this->allocator, NULL);
gst_query_add_allocation_meta (query, GST_NET_CONTROL_MESSAGE_META_API_TYPE,
NULL);
return TRUE;
}
static void
gst_pinos_socket_sink_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec)
{
switch (prop_id) {
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static void
gst_pinos_socket_sink_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec)
{
GstPinosSocketSink *this = GST_PINOS_SOCKET_SINK (object);
switch (prop_id) {
case PROP_NUM_HANDLES:
g_value_set_uint (value, g_hash_table_size (this->hash));
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static gboolean
gst_pinos_socket_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
{
GstPinosSocketSink *this = GST_PINOS_SOCKET_SINK (bsink);
GstStructure *str;
str = gst_caps_get_structure (caps, 0);
this->pinos_input = gst_structure_has_name (str, "application/x-pinos");
return GST_BASE_SINK_CLASS (parent_class)->set_caps (bsink, caps);
}
static void
release_fds (GstPinosSocketSink *this, GstBuffer *buffer)
{
GArray *fdids;
guint i;
PinosBufferBuilder b;
PinosPacketReleaseFDPayload r;
PinosBuffer pbuf;
gsize size;
gpointer data;
GstBuffer *outbuf;
GstEvent *ev;
fdids = gst_mini_object_steal_qdata (GST_MINI_OBJECT_CAST (buffer),
fdids_quark);
if (fdids == NULL)
return;
pinos_buffer_builder_init (&b);
for (i = 0; i < fdids->len; i++) {
r.id = g_array_index (fdids, guint32, i);
GST_LOG ("release fd index %d", r.id);
pinos_buffer_builder_add_release_fd_payload (&b, &r);
}
pinos_buffer_builder_end (&b, &pbuf);
g_array_unref (fdids);
data = pinos_buffer_steal (&pbuf, &size, NULL);
outbuf = gst_buffer_new_wrapped (data, size);
ev = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
gst_structure_new ("GstNetworkMessage",
"object", G_TYPE_OBJECT, this,
"buffer", GST_TYPE_BUFFER, outbuf, NULL));
gst_buffer_unref (outbuf);
gst_pad_push_event (GST_BASE_SINK_PAD (this), ev);
g_object_unref (this);
}
static GstFlowReturn
gst_pinos_socket_sink_render_pinos (GstPinosSocketSink * this, GstBuffer * buffer)
{
GstMapInfo info;
PinosBuffer pbuf;
PinosBufferIter it;
GArray *fdids = NULL;
gst_buffer_map (buffer, &info, GST_MAP_READ);
pinos_buffer_init_data (&pbuf, info.data, info.size, NULL);
pinos_buffer_iter_init (&it, &pbuf);
while (pinos_buffer_iter_next (&it)) {
switch (pinos_buffer_iter_get_type (&it)) {
case PINOS_PACKET_TYPE_FD_PAYLOAD:
{
PinosPacketFDPayload p;
if (!pinos_buffer_iter_parse_fd_payload (&it, &p))
continue;
if (fdids == NULL)
fdids = g_array_new (FALSE, FALSE, sizeof (guint32));
GST_LOG ("track fd index %d", p.id);
g_array_append_val (fdids, p.id);
break;
}
case PINOS_PACKET_TYPE_FORMAT_CHANGE:
{
PinosPacketFormatChange p;
GstCaps * caps;
if (!pinos_buffer_iter_parse_format_change (&it, &p))
continue;
caps = gst_caps_from_string (p.format);
gst_element_post_message (GST_ELEMENT (this),
gst_message_new_element (GST_OBJECT (this),
gst_structure_new ("PinosPayloaderFormatChange",
"format", GST_TYPE_CAPS, caps, NULL)));
gst_caps_unref (caps);
break;
}
default:
break;
}
}
gst_buffer_unmap (buffer, &info);
pinos_buffer_clear (&pbuf);
if (fdids != NULL) {
gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (buffer),
fdids_quark, fdids, NULL);
gst_mini_object_weak_ref (GST_MINI_OBJECT_CAST (buffer),
(GstMiniObjectNotify) release_fds, g_object_ref (this));
}
gst_burst_cache_queue_buffer (this->cache, gst_buffer_ref (buffer));
return GST_FLOW_OK;
}
static GstMemory *
gst_pinos_socket_sink_get_fd_memory (GstPinosSocketSink * this, GstBuffer * buffer, gboolean *tmpfile)
{
GstMemory *mem = NULL;
if (gst_buffer_n_memory (buffer) == 1
&& gst_is_fd_memory (gst_buffer_peek_memory (buffer, 0))) {
mem = gst_buffer_get_memory (buffer, 0);
*tmpfile = gst_is_tmpfile_memory (mem);
} else {
GstMapInfo info;
GstAllocationParams params = {0, 0, 0, 0, { NULL, }};
gsize size = gst_buffer_get_size (buffer);
GST_INFO_OBJECT (this, "Buffer cannot be sent without copying");
mem = gst_allocator_alloc (this->allocator, size, &params);
if (!gst_memory_map (mem, &info, GST_MAP_WRITE))
return NULL;
gst_buffer_extract (buffer, 0, info.data, size);
gst_memory_unmap (mem, &info);
*tmpfile = TRUE;
}
return mem;
}
static GstFlowReturn
gst_pinos_socket_sink_render_other (GstPinosSocketSink * this, GstBuffer * buffer)
{
GstMemory *fdmem = NULL;
GError *err = NULL;
GstBuffer *outbuf;
PinosBuffer pbuf;
PinosBufferBuilder builder;
PinosPacketHeader hdr;
PinosPacketFDPayload p;
gsize size;
gpointer data;
GSocketControlMessage *msg;
gboolean tmpfile = TRUE;
hdr.flags = 0;
hdr.seq = GST_BUFFER_OFFSET (buffer);
hdr.pts = GST_BUFFER_PTS (buffer) + GST_ELEMENT_CAST (this)->base_time;
hdr.dts_offset = 0;
pinos_buffer_builder_init (&builder);
pinos_buffer_builder_add_header (&builder, &hdr);
fdmem = gst_pinos_socket_sink_get_fd_memory (this, buffer, &tmpfile);
p.fd_index = pinos_buffer_builder_add_fd (&builder, gst_fd_memory_get_fd (fdmem), &err);
if (p.fd_index == -1)
goto add_fd_failed;
p.id = pinos_fd_manager_get_id (this->fdmanager);
p.offset = fdmem->offset;
p.size = fdmem->size;
pinos_buffer_builder_add_fd_payload (&builder, &p);
pinos_buffer_builder_end (&builder, &pbuf);
gst_memory_unref(fdmem);
fdmem = NULL;
data = pinos_buffer_steal (&pbuf, &size, &msg);
outbuf = gst_buffer_new_wrapped (data, size);
GST_BUFFER_PTS (outbuf) = GST_BUFFER_PTS (buffer);
GST_BUFFER_DTS (outbuf) = GST_BUFFER_DTS (buffer);
GST_BUFFER_DURATION (outbuf) = GST_BUFFER_DURATION (buffer);
GST_BUFFER_OFFSET (outbuf) = GST_BUFFER_OFFSET (buffer);
GST_BUFFER_OFFSET_END (outbuf) = GST_BUFFER_OFFSET_END (buffer);
if (!tmpfile) {
GArray *fdids;
/* we are using the original buffer fd in the control message, we need
* to make sure it is not reused before everyone is finished with it.
* We tag the output buffer with the array of fds in it and the original
* buffer (to keep it alive). All clients that receive the fd will
* increment outbuf refcount, all clients that do release-fd on the fd
* will decrease the refcount again. */
fdids = g_array_new (FALSE, FALSE, sizeof (guint32));
g_array_append_val (fdids, p.id);
gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (outbuf),
fdids_quark, fdids, (GDestroyNotify) g_array_unref);
gst_mini_object_set_qdata (GST_MINI_OBJECT_CAST (outbuf),
orig_buffer_quark, gst_buffer_ref (buffer), (GDestroyNotify) gst_buffer_unref);
}
gst_buffer_add_net_control_message_meta (outbuf, msg);
g_object_unref (msg);
gst_burst_cache_queue_buffer (this->cache, outbuf);
return GST_FLOW_OK;
/* ERRORS */
add_fd_failed:
{
GST_WARNING_OBJECT (this, "Adding fd failed: %s", err->message);
gst_memory_unref(fdmem);
g_clear_error (&err);
return GST_FLOW_ERROR;
}
}
static GstFlowReturn
gst_pinos_socket_sink_render (GstBaseSink * bsink, GstBuffer * buffer)
{
GstPinosSocketSink *this = GST_PINOS_SOCKET_SINK (bsink);
if (this->pinos_input)
return gst_pinos_socket_sink_render_pinos (this, buffer);
else
return gst_pinos_socket_sink_render_other (this, buffer);
}
static gboolean
gst_pinos_socket_sink_start (GstBaseSink * basesink)
{
return TRUE;
}
static gboolean
gst_pinos_socket_sink_stop (GstBaseSink * basesink)
{
return TRUE;
}
static gpointer
socketsink_loop (GstPinosSocketSink * this)
{
g_main_loop_run (this->loop);
return NULL;
}
static gboolean
gst_pinos_socket_sink_open (GstPinosSocketSink * this)
{
GError *error = NULL;
this->cache = gst_burst_cache_new (sizeof (MyReader));
this->context = g_main_context_new ();
this->loop = g_main_loop_new (this->context, TRUE);
GST_DEBUG ("context %p, loop %p", this->context, this->loop);
this->thread = g_thread_try_new ("PinosSocketSink",
(GThreadFunc) socketsink_loop,
this,
&error);
if (this->thread == NULL)
goto thread_error;
return TRUE;
/* ERRORS */
thread_error:
{
GST_ELEMENT_ERROR (this, RESOURCE, FAILED,
("Failed to start mainloop thread: %s", error->message), (NULL));
g_clear_error (&error);
g_clear_pointer (&this->loop, g_main_loop_unref);
g_clear_pointer (&this->context, g_main_context_unref);
return FALSE;
}
}
static gboolean
gst_pinos_socket_sink_close (GstPinosSocketSink * this)
{
GST_DEBUG ("context %p, loop %p", this->context, this->loop);
g_main_loop_quit (this->loop);
g_thread_join (this->thread);
this->thread = NULL;
g_clear_pointer (&this->loop, g_main_loop_unref);
g_clear_pointer (&this->context, g_main_context_unref);
g_hash_table_remove_all (this->hash);
g_clear_pointer (&this->cache, g_object_unref);
return TRUE;
}
static GstStateChangeReturn
gst_pinos_socket_sink_change_state (GstElement * element, GstStateChange transition)
{
GstStateChangeReturn ret;
GstPinosSocketSink *this = GST_PINOS_SOCKET_SINK_CAST (element);
switch (transition) {
case GST_STATE_CHANGE_NULL_TO_READY:
if (!gst_pinos_socket_sink_open (this))
goto open_failed;
break;
case GST_STATE_CHANGE_READY_TO_PAUSED:
break;
case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
break;
case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
break;
default:
break;
}
ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
switch (transition) {
case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
break;
case GST_STATE_CHANGE_PAUSED_TO_READY:
break;
case GST_STATE_CHANGE_READY_TO_NULL:
gst_pinos_socket_sink_close (this);
break;
default:
break;
}
return ret;
/* ERRORS */
open_failed:
{
return GST_STATE_CHANGE_FAILURE;
}
}
static void
myreader_callback (GstBurstCache *cache,
GstBurstCacheReader *reader,
gpointer user_data)
{
MyReader *myreader = (MyReader *) reader;
MySource *mysource = myreader->source;
GST_LOG ("%p: callback", reader);
mysource->condition |= G_IO_OUT;
g_source_modify_unix_fd ((GSource *)mysource, mysource->tag, mysource->condition);
}
static int
map_n_memory_output_vector (GstBuffer * buf, GOutputVector * vectors,
GstMapInfo * mapinfo, guint num_vectors)
{
guint mem_len;
guint i;
mem_len = gst_buffer_n_memory (buf);
for (i = 0; i < mem_len && i < num_vectors; i++) {
GstMapInfo map = { 0 };
GstMemory *mem = gst_buffer_peek_memory (buf, i);
if (!gst_memory_map (mem, &map, GST_MAP_READ))
g_error ("Unable to map memory %p. This should never happen.", mem);
vectors[i].buffer = map.data;
vectors[i].size = map.size;
mapinfo[i] = map;
}
return i;
}
static void
unmap_n_memorys (GstMapInfo * mapinfo, int num_mappings)
{
gint i;
for (i = 0; i < num_mappings; i++)
gst_memory_unmap (mapinfo[i].memory, &mapinfo[i]);
}
static gsize
gst_buffer_get_cmsg_list (GstBuffer * buf, GSocketControlMessage ** msgs,
gsize msg_space)
{
gpointer iter_state = NULL;
GstMeta *meta;
gsize msg_count = 0;
while ((meta = gst_buffer_iterate_meta (buf, &iter_state)) != NULL
&& msg_count < msg_space) {
if (meta->info->api == GST_NET_CONTROL_MESSAGE_META_API_TYPE)
msgs[msg_count++] = ((GstNetControlMessageMeta *) meta)->message;
}
return msg_count;
}
#define CMSG_MAX 255
static gboolean
myreader_source_func (GstBurstCacheReader *reader, GIOCondition condition, gpointer user_data)
{
GstPinosSocketSink *this = user_data;
MyReader *myreader = (MyReader *) reader;
MySource *mysource = myreader->source;
GST_LOG ("%p: io condition %d", reader, condition);
if (condition & (G_IO_HUP | G_IO_ERR)) {
GST_DEBUG ("client error");
return FALSE;
}
if (condition & G_IO_IN) {
gssize navail, nread, maxmem;
GstBuffer *buf;
GstEvent *ev;
gchar *mem;
navail = g_socket_get_available_bytes (myreader->socket);
maxmem = MAX (navail, 1);
mem = g_malloc (maxmem);
nread = g_socket_receive (myreader->socket, mem, maxmem, NULL, NULL);
if (nread > 0) {
buf = gst_buffer_new_wrapped (mem, navail);
ev = gst_event_new_custom (GST_EVENT_CUSTOM_UPSTREAM,
gst_structure_new ("GstNetworkMessage",
"object", G_TYPE_OBJECT, myreader->socket,
"buffer", GST_TYPE_BUFFER, buf, NULL));
gst_buffer_unref (buf);
gst_pad_push_event (GST_BASE_SINK_PAD (this), ev);
} else {
GST_DEBUG ("client closed");
mysource->condition &= ~G_IO_IN;
g_source_modify_unix_fd ((GSource *)mysource, mysource->tag, mysource->condition);
g_free (mem);
}
}
if (condition & G_IO_OUT) {
GstBuffer *buf = NULL;
GstBurstCacheResult res;
res = gst_burst_cache_get_buffer (this->cache, reader, &buf);
switch (res) {
case GST_BURST_CACHE_RESULT_ERROR:
break;
case GST_BURST_CACHE_RESULT_OK:
break;
case GST_BURST_CACHE_RESULT_WAIT:
mysource->condition &= ~G_IO_OUT;
g_source_modify_unix_fd ((GSource *)mysource, mysource->tag, mysource->condition);
break;
case GST_BURST_CACHE_RESULT_EOS:
gst_burst_cache_remove_reader (this->cache, reader, FALSE);
break;
}
if (buf) {
GstMapInfo maps[8];
GOutputVector vec[8];
guint mems_mapped;
GSocketControlMessage *cmsgs[CMSG_MAX];
gsize msg_count;
mems_mapped = map_n_memory_output_vector (buf, vec, maps, 8);
msg_count = gst_buffer_get_cmsg_list (buf, cmsgs, CMSG_MAX);
g_socket_send_message (myreader->socket, NULL, vec, mems_mapped, cmsgs, msg_count, 0,
NULL, NULL);
unmap_n_memorys (maps, mems_mapped);
gst_buffer_unref (buf);
}
}
return TRUE;
}
static void
myreader_destroy (MyReader *myreader)
{
gst_burst_cache_reader_destroy ((GstBurstCacheReader *)myreader);
g_clear_object (&myreader->socket);
g_source_destroy ((GSource*) myreader->source);
myreader->id = 0;
}
static void
gst_pinos_socket_sink_add (GstPinosSocketSink * this, GSocket *socket)
{
GstBurstCacheReader *reader;
MyReader *myreader;
MySource *mysource;
int fd;
fd = g_socket_get_fd (socket);
if (g_hash_table_lookup (this->hash, GINT_TO_POINTER (fd)))
return;
reader = gst_burst_cache_reader_new (this->cache,
(GstBurstCacheReaderCallback) myreader_callback,
this,
NULL);
reader->hook.destroy = (GDestroyNotify) myreader_destroy;
myreader = (MyReader *)reader;
myreader->socket = g_object_ref (socket);
mysource = (MySource*) g_source_new (&mysource_funcs, sizeof (MySource));
mysource->reader = myreader;
mysource->condition = G_IO_IN;
mysource->tag = g_source_add_unix_fd ((GSource*)mysource, fd, mysource->condition);
myreader->source = mysource;
g_source_set_callback ((GSource*)mysource,
(GSourceFunc) myreader_source_func,
this, NULL);
myreader->id = g_source_attach ((GSource*)mysource, this->context);
g_hash_table_insert (this->hash, GINT_TO_POINTER (fd), reader);
gst_burst_cache_add_reader (this->cache, reader);
}
static void
gst_pinos_socket_sink_remove (GstPinosSocketSink * this, GSocket *socket, gboolean drain)
{
GstBurstCacheReader *reader;
MyReader *myreader;
int fd;
fd = g_socket_get_fd (socket);
myreader = g_hash_table_lookup (this->hash, GINT_TO_POINTER (fd));
if (myreader == NULL)
return;
g_hash_table_remove (this->hash, GINT_TO_POINTER (fd));
reader = (GstBurstCacheReader *) myreader;
gst_burst_cache_remove_reader (this->cache, reader, drain);
}
static void
gst_pinos_socket_sink_finalize (GObject * object)
{
GstPinosSocketSink *this = GST_PINOS_SOCKET_SINK (object);
g_clear_pointer (&this->hash, g_hash_table_unref);
G_OBJECT_CLASS (parent_class)->finalize (object);
}
static void
gst_pinos_socket_sink_class_init (GstPinosSocketSinkClass * klass)
{
GObjectClass *gobject_class;
GstElementClass *gstelement_class;
GstBaseSinkClass *gstbasesink_class;
gobject_class = (GObjectClass *) klass;
gstelement_class = (GstElementClass *) klass;
gstbasesink_class = (GstBaseSinkClass *) klass;
gobject_class->finalize = gst_pinos_socket_sink_finalize;
gobject_class->set_property = gst_pinos_socket_sink_set_property;
gobject_class->get_property = gst_pinos_socket_sink_get_property;
g_object_class_install_property (gobject_class, PROP_NUM_HANDLES,
g_param_spec_uint ("num-handles", "Number of handles",
"The current number of client handles",
0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
gstelement_class->change_state = gst_pinos_socket_sink_change_state;
/**
* GstPinosSocketSink::add:
* @gstpinossocketsink: the pinossocketsink element to emit this signal on
* @socket: the socket to add to pinossocketsink
*
* Hand the given open file descriptor to pinossocketsink to write to.
*/
gst_pinos_socket_sink_signals[SIGNAL_ADD] =
g_signal_new ("add", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstPinosSocketSinkClass, add), NULL, NULL,
g_cclosure_marshal_generic, G_TYPE_NONE, 1, G_TYPE_SOCKET);
/**
* GstPinosSocketSink::remove:
* @gstpinossocketsink: the pinossocketsink element to emit this signal on
* @socket: the socket to remove from pinossocketsink
* @drain: if pending data should be written first.
*
* Remove the given open file descriptor from pinossocketsink.
*/
gst_pinos_socket_sink_signals[SIGNAL_REMOVE] =
g_signal_new ("remove", G_TYPE_FROM_CLASS (klass),
G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
G_STRUCT_OFFSET (GstPinosSocketSinkClass, remove), NULL, NULL,
g_cclosure_marshal_generic, G_TYPE_NONE, 2, G_TYPE_SOCKET, G_TYPE_BOOLEAN);
gst_element_class_set_static_metadata (gstelement_class,
"Pinos FD sink", "Sink/Video",
"Send data to pinos clients", "Wim Taymans <wim.taymans@gmail.com>");
gst_element_class_add_pad_template (gstelement_class,
gst_static_pad_template_get (&gst_pinos_socket_sink_template));
gstbasesink_class->set_caps = gst_pinos_socket_sink_setcaps;
gstbasesink_class->propose_allocation = gst_pinos_socket_sink_propose_allocation;
gstbasesink_class->start = gst_pinos_socket_sink_start;
gstbasesink_class->stop = gst_pinos_socket_sink_stop;
gstbasesink_class->render = gst_pinos_socket_sink_render;
klass->add = GST_DEBUG_FUNCPTR (gst_pinos_socket_sink_add);
klass->remove = GST_DEBUG_FUNCPTR (gst_pinos_socket_sink_remove);
fdids_quark = g_quark_from_static_string ("GstPinosSocketSinkFDIds");
orig_buffer_quark = g_quark_from_static_string ("GstPinosSocketSinkOrigBuffer");
GST_DEBUG_CATEGORY_INIT (pinos_socket_sink_debug, "pinossocketsink", 0,
"Pinos Socket Sink");
}
static void
gst_pinos_socket_sink_init (GstPinosSocketSink * this)
{
this->hash = g_hash_table_new (g_direct_hash, g_direct_equal);
this->allocator = gst_tmpfile_allocator_new ();
this->fdmanager = pinos_fd_manager_get (PINOS_FD_MANAGER_DEFAULT);
}

74
pinos/gst/gstpinossocketsink.h Normal file
View file

@ -0,0 +1,74 @@
/* GStreamer
* Copyright (C) <2016> Wim Taymans <wim.taymans@gmail.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef __GST_PINOS_SOCKET_SINK_H__
#define __GST_PINOS_SOCKET_SINK_H__
#include <gio/gio.h>
#include <client/pinos.h>
#include <gst/gst.h>
#include <gst/base/gstbasesink.h>
#include <gst/gstburstcache.h>
G_BEGIN_DECLS
#define GST_TYPE_PINOS_SOCKET_SINK (gst_pinos_socket_sink_get_type())
#define GST_PINOS_SOCKET_SINK(obj) (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_PINOS_SOCKET_SINK,GstPinosSocketSink))
#define GST_PINOS_SOCKET_SINK_CLASS(klass) (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_PINOS_SOCKET_SINK,GstPinosSocketSinkClass))
#define GST_IS_PINOS_SOCKET_SINK(obj) (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_PINOS_SOCKET_SINK))
#define GST_IS_PINOS_SOCKET_SINK_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_PINOS_SOCKET_SINK))
#define GST_PINOS_SOCKET_SINK_CAST(obj) ((GstPinosSocketSink *) (obj))
typedef struct _GstPinosSocketSink GstPinosSocketSink;
typedef struct _GstPinosSocketSinkClass GstPinosSocketSinkClass;
/**
* GstPinosSocketSink:
*
* Opaque data structure.
*/
struct _GstPinosSocketSink {
GstBaseSink element;
gboolean pinos_input;
GstAllocator *allocator;
GstBurstCache *cache;
GHashTable *hash;
GThread *thread;
GMainLoop *loop;
GMainContext *context;
PinosFdManager *fdmanager;
};
struct _GstPinosSocketSinkClass {
GstBaseSinkClass parent_class;
void (*add) (GstPinosSocketSink *sink, GSocket *socket);
void (*remove) (GstPinosSocketSink *sink, GSocket *socket, gboolean drain);
};
GType gst_pinos_socket_sink_get_type (void);
G_END_DECLS
#endif /* __GST_PINOS_SOCKET_SINK_H__ */

18
pinos/modules/gst/gst-source.c
View file

@ -116,7 +116,6 @@ setup_pipeline (PinosGstSource *source, GError **error)
{
PinosGstSourcePrivate *priv = source->priv;
GstBus *bus;
GstElement *elem;
priv->pipeline = gst_pipeline_new (NULL);
@ -128,23 +127,13 @@ setup_pipeline (PinosGstSource *source, GError **error)
gst_bin_add (GST_BIN (priv->pipeline), priv->filter);
gst_element_link (priv->element, priv->filter);
elem = gst_element_factory_make ("pinospay", NULL);
gst_bin_add (GST_BIN (priv->pipeline), elem);
gst_element_link (priv->filter, elem);
priv->sink = gst_element_factory_make ("multisocketsink", NULL);
g_object_set (priv->sink, "buffers-max", 2,
"buffers-soft-max", 1,
"recover-policy", 1, /* latest */
"sync-method", 0, /* latest */
"sync", TRUE,
priv->sink = gst_element_factory_make ("pinossocketsink", NULL);
g_object_set (priv->sink, "sync", TRUE,
"enable-last-sample", FALSE,
"send-dispatched", TRUE,
"send-messages", TRUE,
NULL);
gst_bin_add (GST_BIN (priv->pipeline), priv->sink);
gst_element_link (elem, priv->sink);
gst_element_link (priv->filter, priv->sink);
bus = gst_pipeline_get_bus (GST_PIPELINE (priv->pipeline));
gst_bus_add_watch (bus, bus_handler, source);
@ -218,6 +207,7 @@ stop_pipeline (PinosGstSource *source)
g_debug ("gst-source %p: stopping pipeline", source);
gst_element_set_state (priv->pipeline, GST_STATE_NULL);
g_clear_object (&priv->provider);
}
static void

12
pinos/server/client-source.c
View file

@ -151,17 +151,9 @@ setup_pipeline (PinosClientSource *source)
"name=src "
"caps=application/x-pinos "
"send-messages=true ! "
"pinospay ! "
"multisocketsink "
"buffers-max=2 "
"buffers-soft-max=1 "
"recover-policy=latest "
"sync-method=latest "
"pinossocketsink "
"name=sink "
"sync=true "
"enable-last-sample=false "
"send-messages=true "
"send-dispatched=true",
"enable-last-sample=false ",
NULL);
priv->sink = gst_bin_get_by_name (GST_BIN (priv->pipeline), "sink");
priv->src = gst_bin_get_by_name (GST_BIN (priv->pipeline), "src");