<commit_before>/*
 * (C) Copyright 2014 Kurento (http://kurento.org/)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <gst/gst.h>
#include <commons/kmsagnosticcaps.h>
#include <commons/kms-core-marshal.h>

#include "kmshttppostendpoint.h"

#define PLUGIN_NAME "httppostendpoint"

#define APPSRC_DATA "appsrc-data"
G_DEFINE_QUARK (APPSRC_DATA, appsrc_data);

#define APPSINK_DATA "appsink-data"
G_DEFINE_QUARK (APPSINK_DATA, appsink_data);

#define BASE_TIME_DATA "base-time-data"
G_DEFINE_QUARK (BASE_TIME_DATA, base_time_data);

#define CAPS_DATA "caps-data"
G_DEFINE_QUARK (BASE_TIME_DATA, caps_data);

#define POST_PIPELINE "post-pipeline"

GST_DEBUG_CATEGORY_STATIC (kms_http_post_endpoint_debug_category);
#define GST_CAT_DEFAULT kms_http_post_endpoint_debug_category

#define KMS_HTTP_POST_ENDPOINT_GET_PRIVATE(obj) (  \
  G_TYPE_INSTANCE_GET_PRIVATE (                    \
    (obj),                                         \
    KMS_TYPE_HTTP_POST_ENDPOINT,                   \
    KmsHttpPostEndpointPrivate                     \
  )                                                \
)

struct _KmsHttpPostEndpointPrivate
{
  GstElement *appsrc;
  gboolean use_encoded_media;
  int handler_id;
  GstBus *bus;
};

/* Object properties */
enum
{
  PROP_0,
  PROP_USE_ENCODED_MEDIA,
  N_PROPERTIES
};

static GParamSpec *obj_properties[N_PROPERTIES] = { NULL, };

/* Object signals */
enum
{
  /* actions */
  SIGNAL_PUSH_BUFFER,
  SIGNAL_END_OF_STREAM,
  LAST_SIGNAL
};

static guint http_post_ep_signals[LAST_SIGNAL] = { 0 };

G_DEFINE_TYPE_WITH_CODE (KmsHttpPostEndpoint, kms_http_post_endpoint,
    KMS_TYPE_HTTP_ENDPOINT,
    GST_DEBUG_CATEGORY_INIT (kms_http_post_endpoint_debug_category, PLUGIN_NAME,
        0, "debug category for http post endpoint plugin"));

static void
bus_message (GstBus * bus, GstMessage * msg, KmsHttpEndpoint * self)
{
  if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_EOS)
    g_signal_emit_by_name (G_OBJECT (self), "eos", 0);
}

static void
release_gst_clock (gpointer data)
{
  g_slice_free (GstClockTime, data);
}

static GstFlowReturn
new_sample_post_handler (GstElement * appsink, gpointer user_data)
{
  GstElement *appsrc = GST_ELEMENT (user_data);
  GstSample *sample = NULL;
  GstBuffer *buffer;
  GstFlowReturn ret;
  GstClockTime *base_time;

  g_signal_emit_by_name (appsink, "pull-sample", &sample);
  if (sample == NULL)
    return GST_FLOW_OK;

  buffer = gst_sample_get_buffer (sample);
  if (buffer == NULL) {
    ret = GST_FLOW_OK;
    goto end;
  }

  gst_buffer_ref (buffer);
  buffer = gst_buffer_make_writable (buffer);

  BASE_TIME_LOCK (GST_OBJECT_PARENT (appsrc));

  base_time =
      g_object_get_qdata (G_OBJECT (GST_OBJECT_PARENT (appsrc)),
      base_time_data_quark ());

  if (base_time == NULL) {
    GstClock *clock;

    clock = gst_element_get_clock (appsrc);
    base_time = g_slice_new0 (GstClockTime);

    g_object_set_qdata_full (G_OBJECT (GST_OBJECT_PARENT (appsrc)),
        base_time_data_quark (), base_time, release_gst_clock);
    *base_time =
        gst_clock_get_time (clock) - gst_element_get_base_time (appsrc);
    g_object_unref (clock);
    GST_DEBUG ("Setting base time to: %" G_GUINT64_FORMAT, *base_time);
  }

  if (GST_BUFFER_PTS_IS_VALID (buffer))
    buffer->pts += *base_time;
  if (GST_BUFFER_DTS_IS_VALID (buffer))
    buffer->dts += *base_time;

  BASE_TIME_UNLOCK (GST_OBJECT_PARENT (appsrc));

  /* Pass the buffer through appsrc element which is */
  /* placed in a different pipeline */
  g_signal_emit_by_name (appsrc, "push-buffer", buffer, &ret);

  gst_buffer_unref (buffer);

  if (ret != GST_FLOW_OK) {
    /* something went wrong */
    GST_ERROR ("Could not send buffer to appsrc %s. Cause %s",
        GST_ELEMENT_NAME (appsrc), gst_flow_get_name (ret));
  }

end:
  if (sample != NULL)
    gst_sample_unref (sample);

  return ret;
}

static GstPadProbeReturn
set_appsrc_caps (GstPad * pad, GstPadProbeInfo * info, gpointer httpep)
{
  KmsHttpPostEndpoint *self = KMS_HTTP_POST_ENDPOINT (httpep);
  GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);
  GstCaps *audio_caps = NULL, *video_caps = NULL;
  GstElement *appsrc, *appsink, *agnosticbin;
  GstCaps *caps;
  gpointer data;

  if (GST_EVENT_TYPE (event) != GST_EVENT_CAPS) {
    return GST_PAD_PROBE_OK;
  }

  gst_event_parse_caps (event, &caps);
  if (caps == NULL) {
    GST_ERROR_OBJECT (pad, "Invalid caps received");
    return GST_PAD_PROBE_OK;
  }

  GST_TRACE ("caps are %" GST_PTR_FORMAT, caps);

  data = g_object_get_qdata (G_OBJECT (pad), appsrc_data_quark ());
  if (data != NULL) {
    g_object_set_qdata (G_OBJECT (data), caps_data_quark (), caps);
    goto end;
  }

  /* Get the proper agnosticbin */
  audio_caps = gst_caps_from_string (KMS_AGNOSTIC_AUDIO_CAPS);
  video_caps = gst_caps_from_string (KMS_AGNOSTIC_VIDEO_CAPS);

  if (gst_caps_can_intersect (audio_caps, caps))
    agnosticbin = kms_element_get_audio_agnosticbin (KMS_ELEMENT (self));
  else if (gst_caps_can_intersect (video_caps, caps))
    agnosticbin = kms_element_get_video_agnosticbin (KMS_ELEMENT (self));
  else {
    GST_ELEMENT_WARNING (self, CORE, CAPS,
        ("Unsupported media received: %" GST_PTR_FORMAT, caps),
        ("Unsupported media received: %" GST_PTR_FORMAT, caps));
    goto end;<commit_msg>httppostendpoint: do not set unused qdata<commit_after>/*
 * (C) Copyright 2014 Kurento (http://kurento.org/)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <gst/gst.h>
#include <commons/kmsagnosticcaps.h>
#include <commons/kms-core-marshal.h>

#include "kmshttppostendpoint.h"

#define PLUGIN_NAME "httppostendpoint"

#define APPSRC_DATA "appsrc-data"
G_DEFINE_QUARK (APPSRC_DATA, appsrc_data);

#define APPSINK_DATA "appsink-data"
G_DEFINE_QUARK (APPSINK_DATA, appsink_data);

#define BASE_TIME_DATA "base-time-data"
G_DEFINE_QUARK (BASE_TIME_DATA, base_time_data);

#define POST_PIPELINE "post-pipeline"

GST_DEBUG_CATEGORY_STATIC (kms_http_post_endpoint_debug_category);
#define GST_CAT_DEFAULT kms_http_post_endpoint_debug_category

#define KMS_HTTP_POST_ENDPOINT_GET_PRIVATE(obj) (  \
  G_TYPE_INSTANCE_GET_PRIVATE (                    \
    (obj),                                         \
    KMS_TYPE_HTTP_POST_ENDPOINT,                   \
    KmsHttpPostEndpointPrivate                     \
  )                                                \
)

struct _KmsHttpPostEndpointPrivate
{
  GstElement *appsrc;
  gboolean use_encoded_media;
  int handler_id;
  GstBus *bus;
};

/* Object properties */
enum
{
  PROP_0,
  PROP_USE_ENCODED_MEDIA,
  N_PROPERTIES
};

static GParamSpec *obj_properties[N_PROPERTIES] = { NULL, };

/* Object signals */
enum
{
  /* actions */
  SIGNAL_PUSH_BUFFER,
  SIGNAL_END_OF_STREAM,
  LAST_SIGNAL
};

static guint http_post_ep_signals[LAST_SIGNAL] = { 0 };

G_DEFINE_TYPE_WITH_CODE (KmsHttpPostEndpoint, kms_http_post_endpoint,
    KMS_TYPE_HTTP_ENDPOINT,
    GST_DEBUG_CATEGORY_INIT (kms_http_post_endpoint_debug_category, PLUGIN_NAME,
        0, "debug category for http post endpoint plugin"));

static void
bus_message (GstBus * bus, GstMessage * msg, KmsHttpEndpoint * self)
{
  if (GST_MESSAGE_TYPE (msg) == GST_MESSAGE_EOS)
    g_signal_emit_by_name (G_OBJECT (self), "eos", 0);
}

static void
release_gst_clock (gpointer data)
{
  g_slice_free (GstClockTime, data);
}

static GstFlowReturn
new_sample_post_handler (GstElement * appsink, gpointer user_data)
{
  GstElement *appsrc = GST_ELEMENT (user_data);
  GstSample *sample = NULL;
  GstBuffer *buffer;
  GstFlowReturn ret;
  GstClockTime *base_time;

  g_signal_emit_by_name (appsink, "pull-sample", &sample);
  if (sample == NULL)
    return GST_FLOW_OK;

  buffer = gst_sample_get_buffer (sample);
  if (buffer == NULL) {
    ret = GST_FLOW_OK;
    goto end;
  }

  gst_buffer_ref (buffer);
  buffer = gst_buffer_make_writable (buffer);

  BASE_TIME_LOCK (GST_OBJECT_PARENT (appsrc));

  base_time =
      g_object_get_qdata (G_OBJECT (GST_OBJECT_PARENT (appsrc)),
      base_time_data_quark ());

  if (base_time == NULL) {
    GstClock *clock;

    clock = gst_element_get_clock (appsrc);
    base_time = g_slice_new0 (GstClockTime);

    g_object_set_qdata_full (G_OBJECT (GST_OBJECT_PARENT (appsrc)),
        base_time_data_quark (), base_time, release_gst_clock);
    *base_time =
        gst_clock_get_time (clock) - gst_element_get_base_time (appsrc);
    g_object_unref (clock);
    GST_DEBUG ("Setting base time to: %" G_GUINT64_FORMAT, *base_time);
  }

  if (GST_BUFFER_PTS_IS_VALID (buffer))
    buffer->pts += *base_time;
  if (GST_BUFFER_DTS_IS_VALID (buffer))
    buffer->dts += *base_time;

  BASE_TIME_UNLOCK (GST_OBJECT_PARENT (appsrc));

  /* Pass the buffer through appsrc element which is */
  /* placed in a different pipeline */
  g_signal_emit_by_name (appsrc, "push-buffer", buffer, &ret);

  gst_buffer_unref (buffer);

  if (ret != GST_FLOW_OK) {
    /* something went wrong */
    GST_ERROR ("Could not send buffer to appsrc %s. Cause %s",
        GST_ELEMENT_NAME (appsrc), gst_flow_get_name (ret));
  }

end:
  if (sample != NULL)
    gst_sample_unref (sample);

  return ret;
}

static GstPadProbeReturn
set_appsrc_caps (GstPad * pad, GstPadProbeInfo * info, gpointer httpep)
{
  KmsHttpPostEndpoint *self = KMS_HTTP_POST_ENDPOINT (httpep);
  GstEvent *event = GST_PAD_PROBE_INFO_EVENT (info);
  GstCaps *audio_caps = NULL, *video_caps = NULL;
  GstElement *appsrc, *appsink, *agnosticbin;
  GstCaps *caps;
  gpointer data;

  if (GST_EVENT_TYPE (event) != GST_EVENT_CAPS) {
    return GST_PAD_PROBE_OK;
  }

  gst_event_parse_caps (event, &caps);
  if (caps == NULL) {
    GST_ERROR_OBJECT (pad, "Invalid caps received");
    return GST_PAD_PROBE_OK;
  }

  GST_TRACE ("caps are %" GST_PTR_FORMAT, caps);

  data = g_object_get_qdata (G_OBJECT (pad), appsrc_data_quark ());
  if (data != NULL) {
    goto end;
  }

  /* Get the proper agnosticbin */
  audio_caps = gst_caps_from_string (KMS_AGNOSTIC_AUDIO_CAPS);
  video_caps = gst_caps_from_string (KMS_AGNOSTIC_VIDEO_CAPS);

  if (gst_caps_can_intersect (audio_caps, caps))
    agnosticbin = kms_element_get_audio_agnosticbin (KMS_ELEMENT (self));
  else if (gst_caps_can_intersect (video_caps, caps))
    agnosticbin = kms_element_get_video_agnosticbin (KMS_ELEMENT (self));
  else {
    GST_ELEMENT_WARNING (self, CORE, CAPS,
        ("Unsupported media received: %" GST_PTR_FORMAT, caps),
        ("Unsupported media received: %" GST_PTR_FORMAT, caps));
    goto end;
{'fraction_non_alphanumeric': 0.09809170192556774, 'fraction_numerical': 0.0025904498747949226, 'mean_word_length': 4.048823016564952, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2633305', 'n_tokens_mistral': 4339, 'n_tokens_neox': 4205, 'n_words': 931}