RTMP: add rtmpsink element for output to an RTMP server

docs/plugins/gst-plugins-bad-plugins-docs.sgml

@@ -89,7 +89,9 @@
     <xi:include href="xml/element-pyramidsegment.xml" />
     <xi:include href="xml/element-rsvgdec.xml" />
     <xi:include href="xml/element-rsvgoverlay.xml" />
-    <xi:include href="xml/element-rtpdtmfdepay.xml" />
+    <xi:include href="xml/element-rtmpsink.xml" />
+    <xi:include href="xml/element-rtmpsrc.xml" />
+    <xi:include href="xml/element-rtpmux.xml" />
     <xi:include href="xml/element-rtpdtmfsrc.xml" />
     <xi:include href="xml/element-rtpdtmfmux.xml" />
     <xi:include href="xml/element-rtpmux.xml" />
@@ -196,6 +198,7 @@
     <xi:include href="xml/plugin-real.xml" />
     <xi:include href="xml/plugin-rfbsrc.xml" />
     <xi:include href="xml/plugin-rsvg.xml" />
+    <xi:include href="xml/plugin-rtmp.xml" />
     <xi:include href="xml/plugin-rtpmux.xml" />
     <xi:include href="xml/plugin-scaletempo.xml" />
     <xi:include href="xml/plugin-sdl.xml" />

docs/plugins/gst-plugins-bad-plugins-sections.txt

@@ -1160,6 +1160,34 @@ GST_TYPE_RSVG_DEC
 gst_rsvg_dec_get_type
 </SECTION>
 
+<SECTION>
+<FILE>element-rtmpsink</FILE>
+<TITLE>rtmpsink</TITLE>
+GstRTMPSink
+<SUBSECTION Standard>
+GstRTMPSinkClass
+GST_RTMP_SINK
+GST_IS_RTMP_SINK
+GST_TYPE_RTMP_SINK
+gst_rtmp_sink_get_type
+GST_RTMP_SINK_CLASS
+GST_IS_RTMP_SINK_CLASS
+</SECTION>
+
+<SECTION>
+<FILE>element-rtmpsrc</FILE>
+<TITLE>rtmpsrc</TITLE>
+GstRTMPSrc
+<SUBSECTION Standard>
+GstRTMPSrcClass
+GST_RTMP_SRC
+GST_IS_RTMP_SRC
+GST_TYPE_RTMP_SRC
+gst_rtmp_src_get_type
+GST_RTMP_SRC_CLASS
+GST_IS_RTMP_SRC_CLASS
+</SECTION>
+
 <SECTION>
 <FILE>element-rtpdtmfdepay</FILE>
 <TITLE>rtpdtmfdepay</TITLE>

docs/plugins/inspect/plugin-rtmp.xml

@@ -1,6 +1,6 @@
 <plugin>
-  <name>rtmpsrc</name>
-  <description>RTMP source</description>
+  <name>rtmp</name>
+  <description>RTMP source and sink</description>
   <filename>../../ext/rtmp/.libs/libgstrtmp.so</filename>
   <basename>libgstrtmp.so</basename>
   <version>0.10.22.1</version>
@@ -9,6 +9,21 @@
   <package>GStreamer Bad Plug-ins git</package>
   <origin>Unknown package origin</origin>
   <elements>
+    <element>
+      <name>rtmpsink</name>
+      <longname>RTMP output sink</longname>
+      <class>Sink/Network</class>
+      <description>Sends FLV content to a server via RTMP</description>
+      <author>Jan Schmidt &lt;thaytan@noraisin.net&gt;</author>
+      <pads>
+        <caps>
+          <name>sink</name>
+          <direction>sink</direction>
+          <presence>always</presence>
+          <details>video/x-flv</details>
+        </caps>
+      </pads>
+    </element>
     <element>
       <name>rtmpsrc</name>
       <longname>RTMP Source</longname>

ext/rtmp/Makefile.am

@@ -1,8 +1,8 @@
 plugin_LTLIBRARIES = libgstrtmp.la
 
-libgstrtmp_la_SOURCES = gstrtmpsrc.c
+libgstrtmp_la_SOURCES = gstrtmpsrc.c gstrtmpsink.c gstrtmp.c
 
-noinst_HEADERS = gstrtmpsrc.h
+noinst_HEADERS = gstrtmpsrc.h gstrtmpsink.h
 libgstrtmp_la_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(GST_BASE_CFLAGS) $(GST_CFLAGS) $(RTMP_CFLAGS)
 libgstrtmp_la_LIBADD = $(GST_PLUGINS_BASE_LIBS) $(GST_BASE_LIBS) $(GST_LIBS) $(RTMP_LIBS)
 libgstrtmp_la_LDFLAGS = $(GST_PLUGIN_LDFLAGS)

ext/rtmp/gstrtmp.c

@@ -0,0 +1,54 @@
+/* GStreamer
+ * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
+ *                    2000 Wim Taymans <wtay@chello.be>
+ *                    2002 Kristian Rietveld <kris@gtk.org>
+ *                    2002,2003 Colin Walters <walters@gnu.org>
+ *                    2001,2010 Bastien Nocera <hadess@hadess.net>
+ *                    2010 Sebastian Dröge <sebastian.droege@collabora.co.uk>
+ *                    2010 Jan Schmidt <thaytan@noraisin.net>
+ *
+ * rtmpsrc.c:
+ *
+ * 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., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gst/gst.h>
+
+#include "gstrtmpsrc.h"
+#include "gstrtmpsink.h"
+
+static gboolean
+plugin_init (GstPlugin * plugin)
+{
+  gboolean ret;
+
+  ret = gst_element_register (plugin, "rtmpsrc", GST_RANK_PRIMARY,
+      GST_TYPE_RTMP_SRC);
+  ret &= gst_element_register (plugin, "rtmpsink", GST_RANK_PRIMARY,
+      GST_TYPE_RTMP_SINK);
+
+  return ret;
+}
+
+GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
+    GST_VERSION_MINOR,
+    "rtmp",
+    "RTMP source and sink",
+    plugin_init, VERSION, GST_LICENSE, GST_PACKAGE_NAME, GST_PACKAGE_ORIGIN);

ext/rtmp/gstrtmpsink.c

@@ -0,0 +1,347 @@
+/*
+ * GStreamer
+ * Copyright (C) 2010 Jan Schmidt <thaytan@noraisin.net>
+ *
+ * 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., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+/**
+ * SECTION:element-rtmpsink
+ *
+ * This element delivers data to a streaming server via RTMP. It uses
+ * librtmp, and supports any protocols/urls that librtmp supports.
+ * The URL/location can contain extra connection or session parameters
+ * for librtmp, such as 'flashver=version'. See the librtmp documentation
+ * for more detail
+ *
+ * <refsect2>
+ * <title>Example launch line</title>
+ * |[
+ * gst-launch -v videotestsrc ! ffenc_flv ! flvmux ! rtmpsink location='rtmp://localhost/path/to/stream live=1'
+ * ]| Encode a test video stream to FLV video format and stream it via RTMP.
+ * </refsect2>
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gst/gst.h>
+
+#include "gstrtmpsink.h"
+
+GST_DEBUG_CATEGORY_STATIC (gst_rtmp_sink_debug);
+#define GST_CAT_DEFAULT gst_rtmp_sink_debug
+
+/* Filter signals and args */
+enum
+{
+  /* FILL ME */
+  LAST_SIGNAL
+};
+
+enum
+{
+  PROP_0,
+  PROP_LOCATION
+};
+
+static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
+    GST_PAD_SINK,
+    GST_PAD_ALWAYS,
+    GST_STATIC_CAPS ("video/x-flv")
+    );
+
+static void gst_rtmp_sink_uri_handler_init (gpointer g_iface,
+    gpointer iface_data);
+static void gst_rtmp_sink_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec);
+static void gst_rtmp_sink_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec);
+static gboolean gst_rtmp_sink_stop (GstBaseSink * sink);
+static gboolean gst_rtmp_sink_start (GstBaseSink * sink);
+static GstFlowReturn gst_rtmp_sink_render (GstBaseSink * sink, GstBuffer * buf);
+
+static void
+_do_init (GType gtype)
+{
+  static const GInterfaceInfo urihandler_info = {
+    gst_rtmp_sink_uri_handler_init,
+    NULL,
+    NULL
+  };
+
+  g_type_add_interface_static (gtype, GST_TYPE_URI_HANDLER, &urihandler_info);
+
+  GST_DEBUG_CATEGORY_INIT (gst_rtmp_sink_debug, "rtmpsink", 0,
+      "RTMP server element");
+}
+
+GST_BOILERPLATE_FULL (GstRTMPSink, gst_rtmp_sink, GstBaseSink,
+    GST_TYPE_BASE_SINK, _do_init);
+
+
+static void
+gst_rtmp_sink_base_init (gpointer klass)
+{
+  GstElementClass *element_class = GST_ELEMENT_CLASS (klass);
+
+  gst_element_class_set_details_simple (element_class,
+      "RTMP output sink",
+      "Sink/Network", "Sends FLV content to a server via RTMP",
+      "Jan Schmidt <thaytan@noraisin.net>");
+
+  gst_element_class_add_pad_template (element_class,
+      gst_static_pad_template_get (&sink_template));
+}
+
+/* initialize the plugin's class */
+static void
+gst_rtmp_sink_class_init (GstRTMPSinkClass * klass)
+{
+  GObjectClass *gobject_class;
+  GstBaseSinkClass *gstbasesink_class = (GstBaseSinkClass *) klass;
+
+  gobject_class = (GObjectClass *) klass;
+  gobject_class->set_property = gst_rtmp_sink_set_property;
+  gobject_class->get_property = gst_rtmp_sink_get_property;
+
+  gstbasesink_class->start = GST_DEBUG_FUNCPTR (gst_rtmp_sink_start);
+  gstbasesink_class->stop = GST_DEBUG_FUNCPTR (gst_rtmp_sink_stop);
+  gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_rtmp_sink_render);
+
+  gst_element_class_install_std_props (GST_ELEMENT_CLASS (klass),
+      "location", PROP_LOCATION, G_PARAM_READWRITE, NULL);
+}
+
+/* initialize the new element
+ * initialize instance structure
+ */
+static void
+gst_rtmp_sink_init (GstRTMPSink * sink, GstRTMPSinkClass * klass)
+{
+}
+
+static gboolean
+gst_rtmp_sink_start (GstBaseSink * basesink)
+{
+  GstRTMPSink *sink = GST_RTMP_SINK (basesink);
+
+  if (!sink->uri) {
+    GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_WRITE,
+        ("Please set URI for RTMP output"), ("No URI set before starting"));
+    return FALSE;
+  }
+
+  sink->rtmp_uri = g_strdup (sink->uri);
+  sink->rtmp = RTMP_Alloc ();
+  RTMP_Init (sink->rtmp);
+  if (!RTMP_SetupURL (sink->rtmp, sink->rtmp_uri)) {
+    GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_WRITE, (NULL),
+        ("Failed to setup URL '%s'", sink->uri));
+    RTMP_Free (sink->rtmp);
+    sink->rtmp = NULL;
+    g_free (sink->rtmp_uri);
+    sink->rtmp_uri = NULL;
+    return FALSE;
+  }
+
+  GST_DEBUG_OBJECT (sink, "Created RTMP object");
+
+  /* Mark this as an output connection */
+  RTMP_EnableWrite (sink->rtmp);
+
+  /* open the connection */
+  if (!RTMP_IsConnected (sink->rtmp)) {
+    if (!RTMP_Connect (sink->rtmp, NULL) || !RTMP_ConnectStream (sink->rtmp, 0)) {
+      GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_WRITE, (NULL),
+          ("Could not connect to RTMP stream \"%s\" for writing", sink->uri));
+      RTMP_Free (sink->rtmp);
+      sink->rtmp = NULL;
+      g_free (sink->rtmp_uri);
+      sink->rtmp_uri = NULL;
+      return FALSE;
+    }
+    GST_DEBUG_OBJECT (sink, "Opened connection to %s", sink->rtmp_uri);
+  }
+
+  sink->first = TRUE;
+
+  return TRUE;
+}
+
+static gboolean
+gst_rtmp_sink_stop (GstBaseSink * basesink)
+{
+  GstRTMPSink *sink = GST_RTMP_SINK (basesink);
+
+  gst_buffer_replace (&sink->cache, NULL);
+
+  if (sink->rtmp) {
+    RTMP_Close (sink->rtmp);
+    RTMP_Free (sink->rtmp);
+    sink->rtmp = NULL;
+  }
+  if (sink->rtmp_uri) {
+    g_free (sink->rtmp_uri);
+    sink->rtmp_uri = NULL;
+  }
+
+  return TRUE;
+}
+
+static GstFlowReturn
+gst_rtmp_sink_render (GstBaseSink * bsink, GstBuffer * buf)
+{
+  GstRTMPSink *sink = GST_RTMP_SINK (bsink);
+  GstBuffer *reffed_buf = NULL;
+
+  if (sink->first) {
+    /* FIXME: Parse the first buffer and see if it contains a header plus a packet instead
+     * of just assuming it's only the header */
+    GST_LOG_OBJECT (sink, "Caching first buffer of size %d for concatenation",
+        GST_BUFFER_SIZE (buf));
+    gst_buffer_replace (&sink->cache, buf);
+    sink->first = FALSE;
+    return GST_FLOW_OK;
+  }
+
+  if (sink->cache) {
+    GST_LOG_OBJECT (sink, "Joining 2nd buffer of size %d to cached buf",
+        GST_BUFFER_SIZE (buf));
+    gst_buffer_ref (buf);
+    reffed_buf = buf = gst_buffer_join (sink->cache, buf);
+    sink->cache = NULL;
+  }
+
+  GST_LOG_OBJECT (sink, "Sending %d bytes to RTMP server",
+      GST_BUFFER_SIZE (buf));
+
+  if (!RTMP_Write (sink->rtmp,
+          (char *) GST_BUFFER_DATA (buf), GST_BUFFER_SIZE (buf))) {
+    GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL), ("Failed to write data"));
+    if (reffed_buf)
+      gst_buffer_unref (reffed_buf);
+    return GST_FLOW_ERROR;
+  }
+
+  if (reffed_buf)
+    gst_buffer_unref (reffed_buf);
+
+  return GST_FLOW_OK;
+}
+
+/*
+ * URI interface support.
+ */
+static GstURIType
+gst_rtmp_sink_uri_get_type (void)
+{
+  return GST_URI_SINK;
+}
+
+static gchar **
+gst_rtmp_sink_uri_get_protocols (void)
+{
+  static gchar *protocols[] =
+      { (char *) "rtmp", (char *) "rtmpt", (char *) "rtmps", (char *) "rtmpe",
+    (char *) "rtmfp", (char *) "rtmpte", (char *) "rtmpts", NULL
+  };
+  return protocols;
+}
+
+static const gchar *
+gst_rtmp_sink_uri_get_uri (GstURIHandler * handler)
+{
+  GstRTMPSink *sink = GST_RTMP_SINK (handler);
+
+  return sink->uri;
+}
+
+static gboolean
+gst_rtmp_sink_uri_set_uri (GstURIHandler * handler, const gchar * uri)
+{
+  GstRTMPSink *sink = GST_RTMP_SINK (handler);
+
+  if (GST_STATE (sink) >= GST_STATE_PAUSED)
+    return FALSE;
+
+  g_free (sink->uri);
+  sink->uri = NULL;
+
+  if (uri != NULL) {
+    int protocol;
+    AVal host;
+    unsigned int port;
+    AVal playpath, app;
+
+    if (!RTMP_ParseURL (uri, &protocol, &host, &port, &playpath, &app) ||
+        !host.av_len || !playpath.av_len) {
+      GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_WRITE,
+          ("Failed to parse URI %s", uri), (NULL));
+      return FALSE;
+    }
+    sink->uri = g_strdup (uri);
+  }
+
+  GST_DEBUG_OBJECT (sink, "Changed URI to %s", GST_STR_NULL (uri));
+
+  return TRUE;
+}
+
+static void
+gst_rtmp_sink_uri_handler_init (gpointer g_iface, gpointer iface_data)
+{
+  GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
+
+  iface->get_type = gst_rtmp_sink_uri_get_type;
+  iface->get_protocols = gst_rtmp_sink_uri_get_protocols;
+  iface->get_uri = gst_rtmp_sink_uri_get_uri;
+  iface->set_uri = gst_rtmp_sink_uri_set_uri;
+}
+
+static void
+gst_rtmp_sink_set_property (GObject * object, guint prop_id,
+    const GValue * value, GParamSpec * pspec)
+{
+  GstRTMPSink *sink = GST_RTMP_SINK (object);
+
+  switch (prop_id) {
+    case PROP_LOCATION:
+      gst_rtmp_sink_uri_set_uri (GST_URI_HANDLER (sink),
+          g_value_get_string (value));
+      break;
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}
+
+static void
+gst_rtmp_sink_get_property (GObject * object, guint prop_id,
+    GValue * value, GParamSpec * pspec)
+{
+  GstRTMPSink *sink = GST_RTMP_SINK (object);
+
+  switch (prop_id) {
+    case PROP_LOCATION:
+      g_value_set_string (value, sink->uri);
+      break;
+    default:
+      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
+      break;
+  }
+}

ext/rtmp/gstrtmpsink.h

@@ -0,0 +1,68 @@
+/* 
+ * GStreamer
+ * Copyright (C) 2010 Jan Schmidt <thaytan@noraisin.net>
+ *
+ * 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., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+ 
+#ifndef __GST_RTMP_SINK_H__
+#define __GST_RTMP_SINK_H__
+
+#include <gst/gst.h>
+#include <gst/base/gstbasesink.h>
+
+#include <librtmp/rtmp.h>
+#include <librtmp/log.h>
+#include <librtmp/amf.h>
+
+G_BEGIN_DECLS
+
+#define GST_TYPE_RTMP_SINK \
+  (gst_rtmp_sink_get_type())
+#define GST_RTMP_SINK(obj) \
+  (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_RTMP_SINK,GstRTMPSink))
+#define GST_RTMP_SINK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_RTMP_SINK,GstRTMPSinkClass))
+#define GST_IS_RTMP_SINK(obj) \
+  (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_RTMP_SINK))
+#define GST_IS_RTMP_SINK_CLASS(klass) \
+  (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_RTMP_SINK))
+
+typedef struct _GstRTMPSink      GstRTMPSink;
+typedef struct _GstRTMPSinkClass GstRTMPSinkClass;
+
+struct _GstRTMPSink {
+  GstBaseSink parent;
+
+  /* < private > */
+  gchar *uri;
+
+  RTMP *rtmp;
+  gchar *rtmp_uri; /* copy of url for librtmp */
+
+  GstBuffer *cache; /* Cached buffer */
+  gboolean first;
+};
+
+struct _GstRTMPSinkClass {
+  GstBaseSinkClass parent_class;
+};
+
+GType gst_rtmp_sink_get_type (void);
+
+G_END_DECLS
+
+#endif /* __GST_RTMP_SINK_H__ */

ext/rtmp/gstrtmpsrc.c

@@ -98,6 +98,8 @@ _do_init (GType gtype)
   };
 
   g_type_add_interface_static (gtype, GST_TYPE_URI_HANDLER, &urihandler_info);
+
+  GST_DEBUG_CATEGORY_INIT (rtmpsrc_debug, "rtmpsrc", 0, "RTMP Source");
 }
 
 GST_BOILERPLATE_FULL (GstRTMPSrc, gst_rtmp_src, GstPushSrc, GST_TYPE_PUSH_SRC,
@@ -581,18 +583,3 @@ gst_rtmp_src_stop (GstBaseSrc * basesrc)
 
   return TRUE;
 }
-
-static gboolean
-plugin_init (GstPlugin * plugin)
-{
-  GST_DEBUG_CATEGORY_INIT (rtmpsrc_debug, "rtmpsrc", 0, "RTMP Source");
-
-  return gst_element_register (plugin, "rtmpsrc", GST_RANK_PRIMARY,
-      GST_TYPE_RTMP_SRC);
-}
-
-GST_PLUGIN_DEFINE (GST_VERSION_MAJOR,
-    GST_VERSION_MINOR,
-    "rtmpsrc",
-    "RTMP source",
-    plugin_init, VERSION, GST_LICENSE, GST_PACKAGE_NAME, GST_PACKAGE_ORIGIN);