<commit_before>#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

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

#include "kms-marshal.h"
#include "kmshttpendpoint.h"
#include "kmsagnosticcaps.h"

#define PLUGIN_NAME "httpendpoint"

#define AUDIO_APPSINK "audio_appsink"
#define AUDIO_APPSRC "audio_appsrc"
#define VIDEO_APPSINK "video_appsink"
#define VIDEO_APPSRC "video_appsrc"

#define APPSRC_DATA "appsrc_data"
#define APPSINK_DATA "appsink_data"

#define GET_PIPELINE "get-pipeline"
#define POST_PIPELINE "post-pipeline"

#define GST_CAT_DEFAULT kms_http_end_point_debug_category
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);

#define KMS_HTTP_END_POINT_GET_PRIVATE(obj) ( \
  G_TYPE_INSTANCE_GET_PRIVATE (               \
    (obj),                                    \
    KMS_TYPE_HTTP_END_POINT,                  \
    KmsHttpEndPointPrivate                    \
  )                                           \
)

typedef enum _HttpMethod HttpMethod;
enum _HttpMethod
{
  UNDEFINED_METHOD,
  GET_METHOD,
  POST_METHOD,
  UNSUPPORTED_METHOD,
};

struct _KmsHttpEndPointPrivate
{
  HttpMethod method;
  GstElement *pipeline;
  GstElement *postsrc;
  GstElement *get_encodebin;
  GstElement *get_appsink;
};

enum
{
  /* signals */
  SIGNAL_EOS,
  SIGNAL_NEW_SAMPLE,

  /* actions */
  SIGNAL_PULL_SAMPLE,
  SIGNAL_PUSH_BUFFER,
  SIGNAL_END_OF_STREAM,
  LAST_SIGNAL
};

static guint http_ep_signals[LAST_SIGNAL] = { 0 };

/* class initialization */

G_DEFINE_TYPE_WITH_CODE (KmsHttpEndPoint, kms_http_end_point,
    KMS_TYPE_ELEMENT,
    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, PLUGIN_NAME,
        0, "debug category for httpendpoint element"));

static KmsHttpEndPoint *
get_http_ep_from_elements (GstElement * appsink, GstElement * appsrc)
{
  GstElement *self;

  self = GST_ELEMENT (GST_OBJECT_PARENT (appsink));
  if (KMS_IS_HTTP_END_POINT (self))
    return KMS_HTTP_END_POINT (self);

  if (appsrc == NULL)
    goto fail;

  self = GST_ELEMENT (GST_OBJECT_PARENT (appsrc));
  if (KMS_IS_HTTP_END_POINT (self))
    return KMS_HTTP_END_POINT (self);

fail:
  GST_ERROR ("Cant get HttpEndPoint");
  return NULL;
}

static GstFlowReturn
new_sample_handler (GstElement * appsink, gpointer user_data)
{
  GstElement *element = GST_ELEMENT (user_data);
  KmsHttpEndPoint *self;
  GstFlowReturn ret;
  GstSample *sample = NULL;
  GstBuffer *buffer;
  GstCaps *caps;

  if (KMS_IS_HTTP_END_POINT (element)) {
    /* Data has been received in encodebin's source pad. */
    /* Raise new-sample signal so that application can */
    /* deal with this stuff. */
    g_signal_emit (G_OBJECT (element), http_ep_signals[SIGNAL_NEW_SAMPLE], 0,
        &ret);
    return ret;
  }

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

  /* element is an appsrc one */
  g_object_get (G_OBJECT (element), "caps", &caps, NULL);
  if (caps == NULL) {
    /* Appsrc has not yet caps defined */
    caps = gst_sample_get_caps (sample);
    if (caps != NULL) {
      GST_DEBUG ("Setting caps %" GST_PTR_FORMAT " to %s", caps,
          GST_ELEMENT_NAME (element));
      g_object_set (element, "caps", caps, NULL);
    } else
      GST_ERROR ("No caps found for %s", GST_ELEMENT_NAME (element));
  }

  buffer = gst_sample_get_buffer (sample);
  if (buffer == NULL)
    return GST_FLOW_ERROR;

  self = get_http_ep_from_elements (appsink, element);
  if (self && self->priv->method == GET_METHOD) {
    /* HttpEndPoint behaves as a recorder */
    buffer->pts = G_GUINT64_CONSTANT (0);
    buffer->dts = G_GUINT64_CONSTANT (0);
    buffer->offset = G_GUINT64_CONSTANT (0);
    buffer->offset_end = G_GUINT64_CONSTANT (0);
  }

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

  if (ret != GST_FLOW_OK) {
    // something wrong
    GST_ERROR ("Could not send buffer to appsrc  %s. Ret code %d", ret,
        GST_ELEMENT_NAME (element));
  }

  return ret;
}

static void
eos_handler (GstElement * appsink, gpointer user_data)
{
  GST_DEBUG ("TODO: Implement this");
}

static void
post_decodebin_pad_added_handler (GstElement * decodebin, GstPad * pad,
    KmsHttpEndPoint * self)
{
  GstElement *appsrc, *agnosticbin, *appsink;
  GstPad *sinkpad;
  GstCaps *audio_caps, *video_caps;
  GstCaps *src_caps;

  if (GST_PAD_IS_SINK (pad))
    return;

  GST_INFO ("pad %s added", gst_pad_get_name (pad));

  /* Create and link appsrc--agnosticbin with proper caps */
  audio_caps = gst_caps_from_string (KMS_AGNOSTIC_AUDIO_CAPS);
  video_caps = gst_caps_from_string (KMS_AGNOSTIC_VIDEO_CAPS);
  src_caps = gst_pad_query_caps (pad, NULL);
  GST_DEBUG ("caps are %" GST_PTR_FORMAT, src_caps);<commit_msg>Fix memory leak getting new sample in callbacks<commit_after>#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

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

#include "kms-marshal.h"
#include "kmshttpendpoint.h"
#include "kmsagnosticcaps.h"

#define PLUGIN_NAME "httpendpoint"

#define AUDIO_APPSINK "audio_appsink"
#define AUDIO_APPSRC "audio_appsrc"
#define VIDEO_APPSINK "video_appsink"
#define VIDEO_APPSRC "video_appsrc"

#define APPSRC_DATA "appsrc_data"
#define APPSINK_DATA "appsink_data"

#define GET_PIPELINE "get-pipeline"
#define POST_PIPELINE "post-pipeline"

#define GST_CAT_DEFAULT kms_http_end_point_debug_category
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);

#define KMS_HTTP_END_POINT_GET_PRIVATE(obj) ( \
  G_TYPE_INSTANCE_GET_PRIVATE (               \
    (obj),                                    \
    KMS_TYPE_HTTP_END_POINT,                  \
    KmsHttpEndPointPrivate                    \
  )                                           \
)

typedef enum _HttpMethod HttpMethod;
enum _HttpMethod
{
  UNDEFINED_METHOD,
  GET_METHOD,
  POST_METHOD,
  UNSUPPORTED_METHOD,
};

struct _KmsHttpEndPointPrivate
{
  HttpMethod method;
  GstElement *pipeline;
  GstElement *postsrc;
  GstElement *get_encodebin;
  GstElement *get_appsink;
};

enum
{
  /* signals */
  SIGNAL_EOS,
  SIGNAL_NEW_SAMPLE,

  /* actions */
  SIGNAL_PULL_SAMPLE,
  SIGNAL_PUSH_BUFFER,
  SIGNAL_END_OF_STREAM,
  LAST_SIGNAL
};

static guint http_ep_signals[LAST_SIGNAL] = { 0 };

/* class initialization */

G_DEFINE_TYPE_WITH_CODE (KmsHttpEndPoint, kms_http_end_point,
    KMS_TYPE_ELEMENT,
    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, PLUGIN_NAME,
        0, "debug category for httpendpoint element"));

static KmsHttpEndPoint *
get_http_ep_from_elements (GstElement * appsink, GstElement * appsrc)
{
  GstElement *self;

  self = GST_ELEMENT (GST_OBJECT_PARENT (appsink));
  if (KMS_IS_HTTP_END_POINT (self))
    return KMS_HTTP_END_POINT (self);

  if (appsrc == NULL)
    goto fail;

  self = GST_ELEMENT (GST_OBJECT_PARENT (appsrc));
  if (KMS_IS_HTTP_END_POINT (self))
    return KMS_HTTP_END_POINT (self);

fail:
  GST_ERROR ("Cant get HttpEndPoint");
  return NULL;
}

static GstFlowReturn
new_sample_handler (GstElement * appsink, gpointer user_data)
{
  GstElement *element = GST_ELEMENT (user_data);
  KmsHttpEndPoint *self;
  GstFlowReturn ret;
  GstSample *sample = NULL;
  GstBuffer *buffer;
  GstCaps *caps;

  if (KMS_IS_HTTP_END_POINT (element)) {
    /* Data has been received in encodebin's source pad. */
    /* Raise new-sample signal so that application can */
    /* deal with this stuff. */
    g_signal_emit (G_OBJECT (element), http_ep_signals[SIGNAL_NEW_SAMPLE], 0,
        &ret);
    return ret;
  }

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

  /* element is an appsrc one */
  g_object_get (G_OBJECT (element), "caps", &caps, NULL);
  if (caps == NULL) {
    /* Appsrc has not yet caps defined */
    caps = gst_sample_get_caps (sample);
    if (caps != NULL) {
      GST_DEBUG ("Setting caps %" GST_PTR_FORMAT " to %s", caps,
          GST_ELEMENT_NAME (element));
      g_object_set (element, "caps", caps, NULL);
    } else
      GST_ERROR ("No caps found for %s", GST_ELEMENT_NAME (element));
  }

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

  self = get_http_ep_from_elements (appsink, element);
  if (self && self->priv->method == GET_METHOD) {
    /* HttpEndPoint behaves as a recorder */
    buffer->pts = G_GUINT64_CONSTANT (0);
    buffer->dts = G_GUINT64_CONSTANT (0);
    buffer->offset = G_GUINT64_CONSTANT (0);
    buffer->offset_end = G_GUINT64_CONSTANT (0);
  }

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

  if (ret != GST_FLOW_OK) {
    // something wrong
    GST_ERROR ("Could not send buffer to appsrc  %s. Ret code %d", ret,
        GST_ELEMENT_NAME (element));
  }

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

  return ret;
}

static void
eos_handler (GstElement * appsink, gpointer user_data)
{
  GST_DEBUG ("TODO: Implement this");
}

static void
post_decodebin_pad_added_handler (GstElement * decodebin, GstPad * pad,
    KmsHttpEndPoint * self)
{
  GstElement *appsrc, *agnosticbin, *appsink;
  GstPad *sinkpad;
  GstCaps *audio_caps, *video_caps;
  GstCaps *src_caps;

  if (GST_PAD_IS_SINK (pad))
    return;

  GST_INFO ("pad %s added", gst_pad_get_name (pad));

  /* Create and link appsrc--agnosticbin with proper caps */
  audio_caps = gst_caps_from_string (KMS_AGNOSTIC_AUDIO_CAPS);
  video_caps = gst_caps_from_string (KMS_AGNOSTIC_VIDEO_CAPS);
  src_caps = gst_pad_query_caps (pad, NULL);
  GST_DEBUG ("caps are %" GST_PTR_FORMAT, src_caps);
{'fraction_non_alphanumeric': 0.10320174075225365, 'fraction_numerical': 0.003108486167236556, 'mean_word_length': 3.7711319822046465, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '2082096', 'n_tokens_mistral': 3669, 'n_tokens_neox': 3568, 'n_words': 792}