path: root/src/encode.c

/*
 * vim: set sts=2 sw=2 et :
 *
 * License: LGPL-2.1+
 * Copyright (c) 2014 Nirbheek Chauhan <nirbheek@centricular.com>
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Library General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Library General Public License for more details.
 *
 * You should have received a copy of the GNU Library General Public
 * License along with this library; if not, write to the
 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
 * Boston, MA 02110-1301, USA.
 */

#include "encode.h"

#include <gst/pbutils/encoding-profile.h>

static gboolean
on_autoplug_continue (GstElement *decodebin,
                      GstPad     *srcpad,
                      GstCaps    *caps,
                      GstElement *encodebin)
{
  gboolean ret;
  char *name;
  GstPad *sinkpad;

  name = gst_caps_to_string (caps);
  g_signal_emit_by_name (encodebin, "request-pad", caps, &sinkpad);

  if (sinkpad != NULL)
    ret = FALSE, g_debug ("encodebin can passthrough %s", name);
  else
    ret = TRUE, g_debug ("encodebin cannot passthrough %s", name);

  if (sinkpad)
    gst_object_unref (sinkpad);
  g_free (name);
  return ret;
}

static void
on_decodebin_pad_added (GstElement *decodebin,
                        GstPad     *srcpad,
                        GstElement *encodebin)
{
  GstCaps *caps;
  GstPad *sinkpad = NULL;
  char *name;

  caps = gst_pad_query_caps (srcpad, NULL);
  name = gst_caps_to_string (caps);

  /* BUG: We need to ignore subtitle streams. Trying to request
   * a pad for this from encodebin causes a pipeline stall. */
  if (g_str_has_prefix (name, "text/x-raw"))
    goto out;

  /* BUG: We don't try to fetch a compatible pad for raw audio because
   * that somehow always fails to link. Transmageddon does the same. */
  if (!g_str_has_prefix (name, "audio/x-raw"))
    /* If we successfully requested a compatible sink pad in
     * "autoplug-continue", we can fetch that here. */
    sinkpad = gst_element_get_compatible_pad (encodebin, srcpad, NULL);

  if (!sinkpad) {
    /* We request a sink pad for the decoded stream */
    g_signal_emit_by_name (encodebin, "request-pad", caps, &sinkpad);
    if (!sinkpad) {
      g_critical ("Failed to request a new sink pad for %s", name);
      goto out;
    }
    g_debug ("Requested a new sink pad for %s", name);
  } else {
    g_debug ("Found an existing sink pad for %s", name);
  }

  if (gst_pad_link (srcpad, sinkpad) != GST_PAD_LINK_OK)
    g_critical ("Couldn't link pads for %s", name);

out:
  g_free (name);
  gst_caps_unref (caps);
  if (sinkpad)
    gst_object_unref (sinkpad);
  return;
}

static GstPadProbeReturn
on_pad_probe_buffer (GstPad          *pad,
                     GstPadProbeInfo *info,
                     gpointer         user_data)
{
  GstBuffer *buffer;
  static int count;
  
  /* Due to a bug, the DISCONT flag is set on the first two buffers,
   * which matroskademux doesn't handle properly. So, we rewrite that flag. */
  buffer = gst_pad_probe_info_get_buffer (info);
  if (GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DISCONT))
    count++;
  GST_BUFFER_FLAG_UNSET (buffer, GST_BUFFER_FLAG_DISCONT);

  if (count != 2)
    return GST_PAD_PROBE_PASS;
  else
    return GST_PAD_PROBE_REMOVE;
}

static void
on_demuxer_pad_added (GstElement *demuxer,
                      GstPad     *srcpad,
                      GstElement *pipeline)
{
  char *name;
  GstCaps *caps;
  GstPad *sinkpad;
  GstElement *rtppay, *udpsink;

  caps = gst_pad_query_caps (srcpad, NULL);
  name = gst_caps_to_string (caps);

  g_debug ("Pad added: %s\n", name);
  if (!g_str_has_prefix (name, "video/x-vp8"))
    goto out;

  /* Broadcast VP8 RTP over UDP://localhost:5004 */
  rtppay = gst_element_factory_make ("rtpvp8pay", "rtpvp8pay");
  udpsink = gst_element_factory_make ("udpsink", "udpsink");
  gst_bin_add_many (GST_BIN (pipeline), rtppay, udpsink, NULL);
  gst_element_link (rtppay, udpsink);

  if (!gst_element_sync_state_with_parent (rtppay) ||
      !gst_element_sync_state_with_parent (udpsink)) {
    g_critical ("Unable to sync rtppay ! udpsink with parent pipeline\n");
    goto out;
  }

  sinkpad = gst_element_get_static_pad (rtppay, "sink");
  gst_pad_link (srcpad, sinkpad);
  stp_print_status ("Streaming a vp8 stream on UDP\n");
out:
  g_free (name);
  gst_object_unref (sinkpad);
}

static void
do_udp_rtp_broadcast (GstElement *decodebin,
                      GstElement *pipeline)
{
  GstPadTemplate *template;
  GstPad *srcpad, *sinkpad;
  GstElement *demuxer, *q, *tee;

  /* We demux the encode stream, and rtp all the vp8 streams we find.
   * Hacky, but quick for this POC. */
  q = gst_element_factory_make ("queue", "rtpq");
  /*demuxer = gst_element_factory_make ("filesink", "matroskademux");
  g_object_set (demuxer, "location", "filesink.webm", NULL);*/
  demuxer = gst_element_factory_make ("matroskademux", "matroskademux");

  gst_bin_add_many (GST_BIN (pipeline), q, demuxer, NULL);
  gst_element_link (q, demuxer);

  tee = gst_bin_get_by_name (GST_BIN (pipeline), "tee");
  template = gst_pad_template_new ("teesrc", GST_PAD_SRC,
                                   GST_PAD_REQUEST, GST_CAPS_ANY);
  srcpad = gst_element_request_pad (tee, template, NULL, GST_CAPS_ANY);
  sinkpad = gst_element_get_static_pad (q, "sink");

  if (!gst_element_sync_state_with_parent (q) ||
      !gst_element_sync_state_with_parent (demuxer)) {
    g_critical ("Unable to sync q ! demuxer with parent pipeline\n");
    goto out;
  }

  gst_pad_link (srcpad, sinkpad);

  /* Due to a bug in webmmux, DISCONT is sent repeatedly for no reason */
  gst_pad_add_probe (srcpad, GST_PAD_PROBE_TYPE_BUFFER,
                     (GstPadProbeCallback) on_pad_probe_buffer, NULL, NULL);

  g_signal_connect (demuxer, "pad-added",
                    G_CALLBACK (on_demuxer_pad_added), pipeline);

out:
  gst_object_unref (template);
  gst_object_unref (srcpad);
  gst_object_unref (tee);
}

static GstEncodingProfile *
create_webm_profile (void)
{
  GstEncodingProfile *t;
  GstEncodingContainerProfile *prof;
  GstCaps *caps;
  GstPreset *vp8preset;

  caps = gst_caps_from_string ("video/webm");
  prof = gst_encoding_container_profile_new ("WebM audio/video", "Standard WEBM/VP8/VORBIS", caps, NULL);
  gst_caps_unref (caps);

  vp8preset = GST_PRESET (gst_element_factory_make ("vp8enc", "vp8preset"));
  /* FIXME: This thing still doesn't encode fast enough in real time */
  g_object_set (vp8preset,
                "cpu-used", 5,
                "end-usage", 1,
                "max-quantizer", 56,
                "threads", 4,
                "undershoot", 95,
                NULL);
  gst_preset_save_preset (vp8preset, "stp_vp8preset");
  gst_object_unref (GST_OBJECT (vp8preset));

  caps = gst_caps_from_string ("video/x-vp8");
  t = (GstEncodingProfile*) gst_encoding_video_profile_new (caps, "stp_vp8preset",
                                                            NULL, 0);
  gst_encoding_container_profile_add_profile (prof, t);
  gst_caps_unref (caps);

  caps = gst_caps_from_string ("audio/x-vorbis");
  t = (GstEncodingProfile*) gst_encoding_audio_profile_new (caps, NULL, NULL, 0);
  gst_encoding_container_profile_add_profile (prof, t);
  gst_caps_unref (caps);

  /* What about application/x-ass? */

  return (GstEncodingProfile*) prof;
}

void
stp_encode_from_msg (TranscodeServerCtx *ctx)
{
#ifdef ENCODE_DEBUG
  char *tmp, *filename;
  GstPad *srcpad, *sinkpad;
  GstElement *filesink, *q2;
  GstPadTemplate *template;
#endif
  GstBus *bus;
  GstElement *src, *decodebin, *encodebin;
  GstElement *tee, *q1, *fakesink;
  GstEncodingProfile *profile;

  g_debug ("Constructing pipeline");

  src = gst_element_factory_make ("appsrc", "src");
  g_object_set (src, "is-live", TRUE,
                "emit-signals", FALSE,
                "stream-type", 0, NULL);
  ctx->appsrc = src;

  decodebin = gst_element_factory_make ("decodebin", "decodebin");

  gst_bin_add_many (GST_BIN (ctx->pipeline), src, decodebin, NULL);
  gst_element_link (src, decodebin);

  /* TODO: Allow setting of a scaling factor to
   * allow realtime encoding of all streams */
  profile = create_webm_profile ();
  encodebin = gst_element_factory_make ("encodebin", "encodebin");
  g_object_set (encodebin,
                "profile", profile,
                "avoid-reencoding", TRUE, NULL);

  tee = gst_element_factory_make ("tee", "tee");
  q1 = gst_element_factory_make ("queue", "q1");
  fakesink = gst_element_factory_make ("fakesink", "fakesink");

  /* Ensure that the stream is always realtime */
  g_object_set (fakesink,
                "sync", TRUE, NULL);

  gst_bin_add_many (GST_BIN (ctx->pipeline), encodebin, tee, q1, fakesink, NULL);
  gst_element_link_many (encodebin, tee, q1, fakesink, NULL);

#ifdef ENCODE_DEBUG
  q2 = gst_element_factory_make ("queue", "q2");
  filesink = gst_element_factory_make ("filesink", "filesink");

  tmp = g_uri_escape_string (ctx->path, NULL, TRUE);
  filename = g_strdup_printf ("debug-encode-%s.webm", tmp);
  g_object_set (filesink, "location", filename, NULL);

  gst_bin_add_many (GST_BIN (ctx->pipeline), q2, filesink, NULL);
  gst_element_link (q2, filesink);

  /* Link pads */
  template = gst_pad_template_new ("teesrc", GST_PAD_SRC,
                                   GST_PAD_REQUEST, GST_CAPS_ANY);
  srcpad = gst_element_request_pad (tee, template, NULL, GST_CAPS_ANY);
  sinkpad = gst_element_get_static_pad (q2, "sink");

  gst_pad_link (srcpad, sinkpad);

  gst_object_unref (template);
  gst_object_unref (srcpad);
  gst_object_unref (sinkpad);
  g_free (filename);
  g_free (tmp);
#endif

  /* The pads of decodebin and encodebin are dynamic,
   * so those will be linked when streams/pads are added */

  /* When decodebin finds a stream that can be decoded, we check
   * if we can pass that directly to encodebin instead of letting
   * decodebin find a decoding element automatically */
  g_signal_connect (decodebin, "autoplug-continue",
                    G_CALLBACK (on_autoplug_continue), encodebin);
  /* When decodebin exposes a source pad, we need to request a
   * corresponding sink pad on decodebin */
  g_signal_connect (decodebin, "pad-added",
                    G_CALLBACK (on_decodebin_pad_added), encodebin);
  /* When finished adding pads, try streaming it all over RTP/UDP */
  g_signal_connect (decodebin, "no-more-pads",
                    G_CALLBACK (do_udp_rtp_broadcast), ctx->pipeline);

  bus = gst_pipeline_get_bus (GST_PIPELINE (ctx->pipeline));
  gst_bus_add_signal_watch (bus);
  g_signal_connect (bus, "message", G_CALLBACK (stp_on_gst_bus_message), ctx);
  g_object_unref (bus);
  gst_encoding_profile_unref (profile);
}