blob: 0cb9db0f33d08d5d5904e3d6cd1420b385184738 [file] [log] [blame]
/* GStreamer
* Copyright (C) <2004> Thomas Vander Stichele <thomas at apestaart dot org>
* Copyright (C) 2006 Andy Wingo <wingo@pobox.com>
* Copyright (C) 2008 Vincent Penquerc'h <ogg.k.ogg.k@googlemail.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-kateparse
* @short_description: parses kate streams
* @see_also: katedec, vorbisparse, oggdemux, theoraparse
*
* <refsect2>
* <para>
* The kateparse element will parse the header packets of the Kate
* stream and put them as the streamheader in the caps. This is used in the
* multifdsink case where you want to stream live kate streams to multiple
* clients, each client has to receive the streamheaders first before they can
* consume the kate packets.
* </para>
* <para>
* This element also makes sure that the buffers that it pushes out are properly
* timestamped and that their offset and offset_end are set. The buffers that
* kateparse outputs have all of the metadata that oggmux expects to receive,
* which allows you to (for example) remux an ogg/kate file.
* </para>
* <title>Example pipelines</title>
* <para>
* <programlisting>
* gst-launch-1.0 -v filesrc location=kate.ogg ! oggdemux ! kateparse ! fakesink
* </programlisting>
* This pipeline shows that the streamheader is set in the caps, and that each
* buffer has the timestamp, duration, offset, and offset_end set.
* </para>
* <para>
* <programlisting>
* gst-launch-1.0 filesrc location=kate.ogg ! oggdemux ! kateparse \
* ! oggmux ! filesink location=kate-remuxed.ogg
* </programlisting>
* This pipeline shows remuxing. kate-remuxed.ogg might not be exactly the same
* as kate.ogg, but they should produce exactly the same decoded data.
* </para>
* </refsect2>
*
*/
#ifdef HAVE_CONFIG_H
# include "config.h"
#endif
#include "gstkate.h"
#include "gstkateutil.h"
#include "gstkateparse.h"
GST_DEBUG_CATEGORY_EXTERN (gst_kateparse_debug);
#define GST_CAT_DEFAULT gst_kateparse_debug
static GstStaticPadTemplate gst_kate_parse_sink_factory =
GST_STATIC_PAD_TEMPLATE ("sink",
GST_PAD_SINK,
GST_PAD_ALWAYS,
GST_STATIC_CAPS ("subtitle/x-kate; application/x-kate")
);
static GstStaticPadTemplate gst_kate_parse_src_factory =
GST_STATIC_PAD_TEMPLATE ("src",
GST_PAD_SRC,
GST_PAD_ALWAYS,
GST_STATIC_CAPS ("subtitle/x-kate; application/x-kate")
);
#define gst_kate_parse_parent_class parent_class
G_DEFINE_TYPE (GstKateParse, gst_kate_parse, GST_TYPE_ELEMENT);
static GstFlowReturn gst_kate_parse_chain (GstPad * pad, GstObject * parent,
GstBuffer * buffer);
static GstStateChangeReturn gst_kate_parse_change_state (GstElement * element,
GstStateChange transition);
static gboolean gst_kate_parse_sink_event (GstPad * pad, GstObject * parent,
GstEvent * event);
static gboolean gst_kate_parse_src_query (GstPad * pad, GstObject * parent,
GstQuery * query);
#if 0
static gboolean gst_kate_parse_convert (GstPad * pad,
GstFormat src_format, gint64 src_value,
GstFormat * dest_format, gint64 * dest_value);
#endif
static GstFlowReturn gst_kate_parse_parse_packet (GstKateParse * parse,
GstBuffer * buf);
static void
gst_kate_parse_class_init (GstKateParseClass * klass)
{
GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
gstelement_class->change_state = gst_kate_parse_change_state;
gst_element_class_add_static_pad_template (gstelement_class,
&gst_kate_parse_src_factory);
gst_element_class_add_static_pad_template (gstelement_class,
&gst_kate_parse_sink_factory);
gst_element_class_set_static_metadata (gstelement_class, "Kate stream parser",
"Codec/Parser/Subtitle",
"parse raw kate streams",
"Vincent Penquerc'h <ogg.k.ogg.k at googlemail dot com>");
klass->parse_packet = GST_DEBUG_FUNCPTR (gst_kate_parse_parse_packet);
}
static void
gst_kate_parse_init (GstKateParse * parse)
{
parse->sinkpad =
gst_pad_new_from_static_template (&gst_kate_parse_sink_factory, "sink");
gst_pad_set_chain_function (parse->sinkpad,
GST_DEBUG_FUNCPTR (gst_kate_parse_chain));
gst_pad_set_event_function (parse->sinkpad,
GST_DEBUG_FUNCPTR (gst_kate_parse_sink_event));
gst_element_add_pad (GST_ELEMENT (parse), parse->sinkpad);
parse->srcpad =
gst_pad_new_from_static_template (&gst_kate_parse_src_factory, "src");
gst_pad_set_query_function (parse->srcpad,
GST_DEBUG_FUNCPTR (gst_kate_parse_src_query));
gst_element_add_pad (GST_ELEMENT (parse), parse->srcpad);
}
static void
gst_kate_parse_drain_event_queue (GstKateParse * parse)
{
while (parse->event_queue->length) {
GstEvent *event;
event = GST_EVENT_CAST (g_queue_pop_head (parse->event_queue));
gst_pad_event_default (parse->sinkpad, NULL, event);
}
}
static GstFlowReturn
gst_kate_parse_push_headers (GstKateParse * parse)
{
/* mark and put on caps */
GstCaps *caps;
GstBuffer *outbuf;
kate_packet packet;
GList *headers, *outbuf_list = NULL;
int ret;
gboolean res;
/* get the headers into the caps, passing them to kate as we go */
caps =
gst_kate_util_set_header_on_caps (&parse->element,
gst_pad_get_current_caps (parse->sinkpad), parse->streamheader);
if (G_UNLIKELY (!caps)) {
GST_ELEMENT_ERROR (parse, STREAM, DECODE, (NULL),
("Failed to set headers on caps"));
return GST_FLOW_ERROR;
}
GST_DEBUG_OBJECT (parse, "here are the caps: %" GST_PTR_FORMAT, caps);
res = gst_pad_set_caps (parse->srcpad, caps);
gst_caps_unref (caps);
if (G_UNLIKELY (!res)) {
GST_WARNING_OBJECT (parse->srcpad, "Failed to set caps on source pad");
return GST_FLOW_NOT_NEGOTIATED;
}
headers = parse->streamheader;
while (headers) {
GstMapInfo info;
outbuf = GST_BUFFER_CAST (headers->data);
if (!gst_buffer_map (outbuf, &info, GST_MAP_READ)) {
GST_WARNING_OBJECT (outbuf, "Failed to map buffer");
continue;
}
kate_packet_wrap (&packet, info.size, info.data);
ret = kate_decode_headerin (&parse->ki, &parse->kc, &packet);
if (G_UNLIKELY (ret < 0)) {
GST_WARNING_OBJECT (parse, "Failed to decode header: %s",
gst_kate_util_get_error_message (ret));
}
gst_buffer_unmap (outbuf, &info);
/* takes ownership of outbuf, which was previously in parse->streamheader */
GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_HEADER);
outbuf_list = g_list_append (outbuf_list, outbuf);
headers = headers->next;
}
/* first process queued events */
gst_kate_parse_drain_event_queue (parse);
/* push out buffers, ignoring return value... */
headers = outbuf_list;
while (headers) {
outbuf = GST_BUFFER_CAST (headers->data);
gst_pad_push (parse->srcpad, outbuf);
headers = headers->next;
}
g_list_free (outbuf_list);
g_list_free (parse->streamheader);
parse->streamheader = NULL;
parse->streamheader_sent = TRUE;
return GST_FLOW_OK;
}
static void
gst_kate_parse_clear_queue (GstKateParse * parse)
{
GST_DEBUG_OBJECT (parse, "Clearing queue");
while (parse->buffer_queue->length) {
GstBuffer *buf;
buf = GST_BUFFER_CAST (g_queue_pop_head (parse->buffer_queue));
gst_buffer_unref (buf);
}
while (parse->event_queue->length) {
GstEvent *event;
event = GST_EVENT_CAST (g_queue_pop_head (parse->event_queue));
gst_event_unref (event);
}
}
static GstFlowReturn
gst_kate_parse_push_buffer (GstKateParse * parse, GstBuffer * buf,
gint64 granulepos)
{
GST_LOG_OBJECT (parse, "granulepos %16" G_GINT64_MODIFIER "x", granulepos);
if (granulepos < 0) {
/* packets coming not from Ogg won't have a granpos in the offset end,
so we have to synthesize one here - only problem is we don't know
the backlink - pretend there's none for now */
GST_INFO_OBJECT (parse, "No granulepos on buffer, synthesizing one");
granulepos =
kate_duration_granule (&parse->ki,
GST_BUFFER_TIMESTAMP (buf) /
(double) GST_SECOND) << kate_granule_shift (&parse->ki);
}
GST_BUFFER_OFFSET (buf) =
kate_granule_time (&parse->ki, granulepos) * GST_SECOND;
GST_BUFFER_OFFSET_END (buf) = granulepos;
GST_BUFFER_TIMESTAMP (buf) = GST_BUFFER_OFFSET (buf);
return gst_pad_push (parse->srcpad, buf);
}
static GstFlowReturn
gst_kate_parse_drain_queue_prematurely (GstKateParse * parse)
{
GstFlowReturn ret = GST_FLOW_OK;
/* got an EOS event, make sure to push out any buffers that were in the queue
* -- won't normally be the case, but this catches the
* didn't-get-a-granulepos-on-the-last-packet case. Assuming a continuous
* stream. */
/* if we got EOS before any buffers came, go ahead and push the other events
* first */
gst_kate_parse_drain_event_queue (parse);
while (!g_queue_is_empty (parse->buffer_queue)) {
GstBuffer *buf;
gint64 granpos;
buf = GST_BUFFER_CAST (g_queue_pop_head (parse->buffer_queue));
granpos = GST_BUFFER_OFFSET_END (buf);
ret = gst_kate_parse_push_buffer (parse, buf, granpos);
if (ret != GST_FLOW_OK)
goto done;
}
g_assert (g_queue_is_empty (parse->buffer_queue));
done:
return ret;
}
static GstFlowReturn
gst_kate_parse_drain_queue (GstKateParse * parse, gint64 granulepos)
{
GstFlowReturn ret = GST_FLOW_OK;
if (!g_queue_is_empty (parse->buffer_queue)) {
GstBuffer *buf;
buf = GST_BUFFER_CAST (g_queue_pop_head (parse->buffer_queue));
ret = gst_kate_parse_push_buffer (parse, buf, granulepos);
if (ret != GST_FLOW_OK)
goto done;
}
g_assert (g_queue_is_empty (parse->buffer_queue));
done:
return ret;
}
static GstFlowReturn
gst_kate_parse_queue_buffer (GstKateParse * parse, GstBuffer * buf)
{
GstFlowReturn ret = GST_FLOW_OK;
gint64 granpos;
buf = gst_buffer_make_writable (buf);
/* oggdemux stores the granule pos in the offset end */
granpos = GST_BUFFER_OFFSET_END (buf);
GST_LOG_OBJECT (parse, "granpos %16" G_GINT64_MODIFIER "x", granpos);
g_queue_push_tail (parse->buffer_queue, buf);
#if 1
/* if getting buffers from matroska, we won't have a granpos here... */
//if (GST_BUFFER_OFFSET_END_IS_VALID (buf)) {
ret = gst_kate_parse_drain_queue (parse, granpos);
//}
#else
if (granpos >= 0) {
ret = gst_kate_parse_drain_queue (parse, granpos);
} else {
GST_ELEMENT_ERROR (parse, STREAM, DECODE, (NULL),
("Bad granulepos %" G_GINT64_FORMAT, granpos));
ret = GST_FLOW_ERROR;
}
#endif
return ret;
}
static GstFlowReturn
gst_kate_parse_parse_packet (GstKateParse * parse, GstBuffer * buf)
{
GstFlowReturn ret = GST_FLOW_OK;
guint8 header[1];
gsize size;
g_assert (parse);
parse->packetno++;
size = gst_buffer_extract (buf, 0, header, 1);
GST_LOG_OBJECT (parse, "Got packet %02x, %" G_GSIZE_FORMAT " bytes",
size ? header[0] : -1, gst_buffer_get_size (buf));
if (size > 0 && header[0] & 0x80) {
GST_DEBUG_OBJECT (parse, "Found header %02x", header[0]);
/* if 0x80 is set, it's streamheader,
* so put it on the streamheader list and return */
parse->streamheader = g_list_append (parse->streamheader, buf);
ret = GST_FLOW_OK;
} else {
if (!parse->streamheader_sent) {
GST_DEBUG_OBJECT (parse, "Found non header, pushing headers seen so far");
ret = gst_kate_parse_push_headers (parse);
}
if (ret == GST_FLOW_OK) {
ret = gst_kate_parse_queue_buffer (parse, buf);
}
}
return ret;
}
static GstFlowReturn
gst_kate_parse_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
{
GstKateParseClass *klass;
GstKateParse *parse;
parse = GST_KATE_PARSE (parent);
klass = GST_KATE_PARSE_CLASS (G_OBJECT_GET_CLASS (parse));
g_assert (klass->parse_packet != NULL);
if (G_UNLIKELY (!gst_pad_has_current_caps (pad)))
return GST_FLOW_NOT_NEGOTIATED;
return klass->parse_packet (parse, buffer);
}
static gboolean
gst_kate_parse_queue_event (GstKateParse * parse, GstEvent * event)
{
GstFlowReturn ret = TRUE;
g_queue_push_tail (parse->event_queue, event);
return ret;
}
static gboolean
gst_kate_parse_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
gboolean ret;
GstKateParse *parse;
parse = GST_KATE_PARSE (parent);
switch (GST_EVENT_TYPE (event)) {
case GST_EVENT_FLUSH_STOP:
gst_kate_parse_clear_queue (parse);
ret = gst_pad_event_default (pad, parent, event);
break;
case GST_EVENT_EOS:
if (!parse->streamheader_sent) {
GST_DEBUG_OBJECT (parse, "Got EOS, pushing headers seen so far");
ret = gst_kate_parse_push_headers (parse);
if (ret != GST_FLOW_OK)
break;
}
gst_kate_parse_drain_queue_prematurely (parse);
ret = gst_pad_event_default (pad, parent, event);
break;
default:
if (!parse->streamheader_sent && GST_EVENT_IS_SERIALIZED (event)
&& GST_EVENT_TYPE (event) > GST_EVENT_CAPS)
ret = gst_kate_parse_queue_event (parse, event);
else
ret = gst_pad_event_default (pad, parent, event);
break;
}
return ret;
}
#if 0
static gboolean
gst_kate_parse_convert (GstPad * pad,
GstFormat src_format, gint64 src_value,
GstFormat * dest_format, gint64 * dest_value)
{
gboolean res = TRUE;
GstKateParse *parse;
parse = GST_KATE_PARSE (GST_PAD_PARENT (pad));
/* fixme: assumes atomic access to lots of instance variables modified from
* the streaming thread, including 64-bit variables */
if (!parse->streamheader_sent)
return FALSE;
if (src_format == *dest_format) {
*dest_value = src_value;
return TRUE;
}
if (parse->sinkpad == pad &&
(src_format == GST_FORMAT_BYTES || *dest_format == GST_FORMAT_BYTES))
return FALSE;
switch (src_format) {
case GST_FORMAT_TIME:
switch (*dest_format) {
default:
res = FALSE;
}
break;
case GST_FORMAT_DEFAULT:
switch (*dest_format) {
case GST_FORMAT_TIME:
*dest_value = kate_granule_time (&parse->ki, src_value) * GST_SECOND;
break;
default:
res = FALSE;
}
break;
default:
res = FALSE;
}
return res;
}
#endif
static gboolean
gst_kate_parse_src_query (GstPad * pad, GstObject * parent, GstQuery * query)
{
#if 1
// TODO
GST_WARNING ("gst_kate_parse_src_query");
return FALSE;
#else
gint64 granulepos;
GstKateParse *parse;
gboolean res = FALSE;
parse = GST_KATE_PARSE (parent);
switch (GST_QUERY_TYPE (query)) {
case GST_QUERY_POSITION:
{
GstFormat format;
gint64 value;
granulepos = parse->prev_granulepos;
gst_query_parse_position (query, &format, NULL);
/* and convert to the final format */
if (!(res =
gst_kate_parse_convert (pad, GST_FORMAT_DEFAULT, granulepos,
&format, &value)))
goto error;
/* fixme: support segments
value = (value - parse->segment_start) + parse->segment_time;
*/
gst_query_set_position (query, format, value);
GST_LOG_OBJECT (parse, "query %p: peer returned granulepos: %"
G_GUINT64_FORMAT " - we return %" G_GUINT64_FORMAT " (format %u)",
query, granulepos, value, format);
break;
}
case GST_QUERY_DURATION:
{
/* fixme: not threadsafe */
/* query peer for total length */
if (!gst_pad_is_linked (parse->sinkpad)) {
GST_WARNING_OBJECT (parse, "sink pad %" GST_PTR_FORMAT " is not linked",
parse->sinkpad);
goto error;
}
if (!(res = gst_pad_query (GST_PAD_PEER (parse->sinkpad), query)))
goto error;
break;
}
case GST_QUERY_CONVERT:
{
GstFormat src_fmt, dest_fmt;
gint64 src_val, dest_val;
gst_query_parse_convert (query, &src_fmt, &src_val, &dest_fmt, &dest_val);
if (!(res =
gst_kate_parse_convert (pad, src_fmt, src_val, &dest_fmt,
&dest_val)))
goto error;
gst_query_set_convert (query, src_fmt, src_val, dest_fmt, dest_val);
break;
}
default:
res = gst_pad_query_default (pad, query);
break;
}
return res;
error:
{
GST_WARNING_OBJECT (parse, "error handling query");
return res;
}
#endif
}
static void
gst_kate_parse_free_stream_headers (GstKateParse * parse)
{
while (parse->streamheader != NULL) {
gst_buffer_unref (GST_BUFFER (parse->streamheader->data));
parse->streamheader = g_list_delete_link (parse->streamheader,
parse->streamheader);
}
}
static GstStateChangeReturn
gst_kate_parse_change_state (GstElement * element, GstStateChange transition)
{
GstKateParse *parse = GST_KATE_PARSE (element);
GstStateChangeReturn ret;
switch (transition) {
case GST_STATE_CHANGE_READY_TO_PAUSED:
kate_info_init (&parse->ki);
kate_comment_init (&parse->kc);
parse->packetno = 0;
parse->streamheader_sent = FALSE;
parse->streamheader = NULL;
parse->buffer_queue = g_queue_new ();
parse->event_queue = g_queue_new ();
break;
default:
break;
}
ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
switch (transition) {
case GST_STATE_CHANGE_PAUSED_TO_READY:
kate_info_clear (&parse->ki);
kate_comment_clear (&parse->kc);
gst_kate_parse_clear_queue (parse);
g_queue_free (parse->buffer_queue);
parse->buffer_queue = NULL;
g_queue_free (parse->event_queue);
parse->event_queue = NULL;
gst_kate_parse_free_stream_headers (parse);
break;
default:
break;
}
return ret;
}