<commit_before>/*
 * Copyright (C) 2013 Kurento
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 3
 * as published by the Free Software Foundation.
 *
 * This program 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <libsoup/soup.h>
#include <uuid/uuid.h>
#include <string.h>

#include "KmsHttpEPServer.h"
#include "kms-enumtypes.h"
#include "kms-marshal.h"

#define OBJECT_NAME "HttpEPServer"

/* 36-byte string (plus tailing '\0') */
#define UUID_STR_SIZE 37

#define KEY_HTTP_EP_SERVER "kms-http-ep-server"
#define KEY_NEW_SAMPLE_HANDLER_ID "kms-new-sample-handler-id"
#define KEY_GOT_CHUNK_HANDLER_ID "kms-got-chunk-handler-id"
#define KEY_FINISHED_HANDLER_ID "kms-finish-handler-id"
#define KEY_EOS_HANDLER_ID "kms-eos-handler-id"
#define KEY_FINISHED "kms-finish"
#define KEY_BOUNDARY "kms-boundary"
#define KEY_MESSAGE "kms-message"

#define GST_CAT_DEFAULT kms_http_ep_server_debug_category
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);

#define KMS_HTTP_EP_SERVER_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE ((obj), KMS_TYPE_HTTP_EP_SERVER, KmsHttpEPServerPrivate))
struct _KmsHttpEPServerPrivate {
  GHashTable *handlers;
  SoupServer *server;
  gchar *iface;
  gint port;
};

static GType http_t = G_TYPE_INVALID;

#define KMS_IS_EXPECTED_TYPE(obj, objtype) (G_TYPE_CHECK_INSTANCE_TYPE((obj),(objtype)))

/* class initialization */

G_DEFINE_TYPE_WITH_CODE (KmsHttpEPServer, kms_http_ep_server,
    G_TYPE_OBJECT,
    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, OBJECT_NAME,
        0, "debug category for " OBJECT_NAME " element") )

/* properties */
enum {
  PROP_0,

  PROP_KMS_HTTP_EP_SERVER_PORT,
  PROP_KMS_HTTP_EP_SERVER_INTERFACE,

  N_PROPERTIES
};

#define KMS_HTTP_EP_SERVER_DEFAULT_PORT 0
#define KMS_HTTP_EP_SERVER_DEFAULT_INTERFACE NULL

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

/* signals */
enum {
  ACTION_REQUESTED,
  URL_REMOVED,
  LAST_SIGNAL
};

static guint obj_signals[LAST_SIGNAL] = { 0 };

struct resolv_data {
  KmsHttpEPServerStartCallback cb;
  KmsHttpEPServer *server;
};

struct sample_data {
  KmsHttpEPServer *httpepserver;
  GstSample *sample;
  SoupMessage *msg;
};

static gboolean
msg_has_finished (SoupMessage *msg)
{
  gboolean *finished;

  finished = (gboolean *) g_object_get_data (G_OBJECT (msg), KEY_FINISHED);

  return *finished;
}

static gboolean
send_buffer_cb (gpointer data)
{
  struct sample_data *sdata = (struct sample_data *) data;
  GstBuffer *buffer;
  GstMapInfo info;

  if (msg_has_finished (sdata->msg) ) {
    GST_WARNING ("Client has closed underlaying HTTP connection. "
        "Buffer won't be sent");
    return FALSE;
  }

  buffer = gst_sample_get_buffer (sdata->sample);

  if (buffer == NULL)
    return FALSE;

  if (!gst_buffer_map (buffer, &info, GST_MAP_READ) ) {
    GST_WARNING ("Could not get buffer map");
    return FALSE;
  }

  soup_message_body_append (sdata->msg->response_body, SOUP_MEMORY_COPY,
      info.data, info.size);
  soup_server_unpause_message (sdata->httpepserver->priv->server, sdata->msg);

  gst_buffer_unmap (buffer, &info);
  return FALSE;
}

static void
destroy_sample_data (gpointer data)
{
  struct sample_data *sdata = (struct sample_data *) data;

  if (sdata->sample != NULL)
    gst_sample_unref (sdata->sample);

  if (sdata->httpepserver != NULL)
    g_object_unref (sdata->httpepserver);

  if (sdata->msg != NULL)
    g_object_unref (sdata->msg);

  g_slice_free (struct sample_data, sdata);
}

static GstFlowReturn
new_sample_handler (GstElement *httpep, gpointer data)
{
  SoupMessage *msg = (SoupMessage *) g_object_get_data (G_OBJECT (httpep), KEY_MESSAGE);
  GstSample *sample = NULL;
  struct sample_data *sdata;

  GST_DEBUG ("New-sample for message %P", msg);

  g_signal_emit_by_name (httpep, "pull-sample", &sample);

  if (sample == NULL)
    return GST_FLOW_ERROR;

  sdata = g_slice_new (struct sample_data);
  sdata->sample = gst_sample_ref (sample);
  sdata->msg = (SoupMessage *) g_object_ref (G_OBJECT (msg) );
  sdata->httpepserver = KMS_HTTP_EP_SERVER (g_object_ref (
      g_object_get_data (G_OBJECT (msg), KEY_HTTP_EP_SERVER) ) );

  /* Write buffer in the main context thread */
  g_idle_add_full (G_PRIORITY_HIGH_IDLE, send_buffer_cb, sdata,
      destroy_sample_data);

  gst_sample_unref (sample);
  return GST_FLOW_OK;
}

static void
get_recv_eos (GstElement *httep, gpointer data)
{
  SoupMessage *msg = (SoupMessage *) data;

  GST_DEBUG ("EOS received on HttpEndPoint %s", GST_ELEMENT_NAME (httep) );
  soup_message_body_complete (msg->response_body);
}

static void
msg_finished (SoupMessage *msg)
{
  gboolean *finished;

  finished = (gboolean *) g_object_get_data (G_OBJECT (msg), KEY_FINISHED);
  *finished = TRUE;
}

static void
disconnect_eos_new_sample_signals (SoupMessage *msg)
{
  KmsHttpEPServer *serv = KMS_HTTP_EP_SERVER (
      g_object_get_data (G_OBJECT (msg), KEY_HTTP_EP_SERVER) );
  SoupURI *uri = soup_message_get_uri (msg);
  const char *path = soup_uri_get_path (uri);
  GstElement *httpep;
  gulong *h1, *h2;

  if (!g_hash_table_contains (serv->priv->handlers, path) ) {
    GST_WARNING ("Message %P was bounded to an unregistered HttpEndPoint", msg);
    return;
  }

  httpep = (GstElement *) g_hash_table_lookup (serv->priv->handlers, path);
  GST_DEBUG ("Message %P is bounded to %s", msg, GST_ELEMENT_NAME (httpep) );

  /* Disconnect signals */
  h1 = (gulong *) g_object_get_data (G_OBJECT (msg), KEY_NEW_SAMPLE_HANDLER_ID);
  h2 = (gulong *) g_object_get_data (G_OBJECT (msg), KEY_EOS_HANDLER_ID);
  g_signal_handler_disconnect (httpep, *h1);
  g_signal_handler_disconnect (httpep, *h2);
}

static void
finished_get_processing (SoupMessage *msg, gpointer data)
{
  GstElement *httpep = GST_ELEMENT (data);
  gpointer param;

  GST_DEBUG ("Message finished %P", msg);<commit_msg>Use only one gulong variable to retrieve handler ids<commit_after>/*
 * Copyright (C) 2013 Kurento
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 3
 * as published by the Free Software Foundation.
 *
 * This program 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <libsoup/soup.h>
#include <uuid/uuid.h>
#include <string.h>

#include "KmsHttpEPServer.h"
#include "kms-enumtypes.h"
#include "kms-marshal.h"

#define OBJECT_NAME "HttpEPServer"

/* 36-byte string (plus tailing '\0') */
#define UUID_STR_SIZE 37

#define KEY_HTTP_EP_SERVER "kms-http-ep-server"
#define KEY_NEW_SAMPLE_HANDLER_ID "kms-new-sample-handler-id"
#define KEY_GOT_CHUNK_HANDLER_ID "kms-got-chunk-handler-id"
#define KEY_FINISHED_HANDLER_ID "kms-finish-handler-id"
#define KEY_EOS_HANDLER_ID "kms-eos-handler-id"
#define KEY_FINISHED "kms-finish"
#define KEY_BOUNDARY "kms-boundary"
#define KEY_MESSAGE "kms-message"

#define GST_CAT_DEFAULT kms_http_ep_server_debug_category
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);

#define KMS_HTTP_EP_SERVER_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE ((obj), KMS_TYPE_HTTP_EP_SERVER, KmsHttpEPServerPrivate))
struct _KmsHttpEPServerPrivate {
  GHashTable *handlers;
  SoupServer *server;
  gchar *iface;
  gint port;
};

static GType http_t = G_TYPE_INVALID;

#define KMS_IS_EXPECTED_TYPE(obj, objtype) (G_TYPE_CHECK_INSTANCE_TYPE((obj),(objtype)))

/* class initialization */

G_DEFINE_TYPE_WITH_CODE (KmsHttpEPServer, kms_http_ep_server,
    G_TYPE_OBJECT,
    GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, OBJECT_NAME,
        0, "debug category for " OBJECT_NAME " element") )

/* properties */
enum {
  PROP_0,

  PROP_KMS_HTTP_EP_SERVER_PORT,
  PROP_KMS_HTTP_EP_SERVER_INTERFACE,

  N_PROPERTIES
};

#define KMS_HTTP_EP_SERVER_DEFAULT_PORT 0
#define KMS_HTTP_EP_SERVER_DEFAULT_INTERFACE NULL

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

/* signals */
enum {
  ACTION_REQUESTED,
  URL_REMOVED,
  LAST_SIGNAL
};

static guint obj_signals[LAST_SIGNAL] = { 0 };

struct resolv_data {
  KmsHttpEPServerStartCallback cb;
  KmsHttpEPServer *server;
};

struct sample_data {
  KmsHttpEPServer *httpepserver;
  GstSample *sample;
  SoupMessage *msg;
};

static gboolean
msg_has_finished (SoupMessage *msg)
{
  gboolean *finished;

  finished = (gboolean *) g_object_get_data (G_OBJECT (msg), KEY_FINISHED);

  return *finished;
}

static gboolean
send_buffer_cb (gpointer data)
{
  struct sample_data *sdata = (struct sample_data *) data;
  GstBuffer *buffer;
  GstMapInfo info;

  if (msg_has_finished (sdata->msg) ) {
    GST_WARNING ("Client has closed underlaying HTTP connection. "
        "Buffer won't be sent");
    return FALSE;
  }

  buffer = gst_sample_get_buffer (sdata->sample);

  if (buffer == NULL)
    return FALSE;

  if (!gst_buffer_map (buffer, &info, GST_MAP_READ) ) {
    GST_WARNING ("Could not get buffer map");
    return FALSE;
  }

  soup_message_body_append (sdata->msg->response_body, SOUP_MEMORY_COPY,
      info.data, info.size);
  soup_server_unpause_message (sdata->httpepserver->priv->server, sdata->msg);

  gst_buffer_unmap (buffer, &info);
  return FALSE;
}

static void
destroy_sample_data (gpointer data)
{
  struct sample_data *sdata = (struct sample_data *) data;

  if (sdata->sample != NULL)
    gst_sample_unref (sdata->sample);

  if (sdata->httpepserver != NULL)
    g_object_unref (sdata->httpepserver);

  if (sdata->msg != NULL)
    g_object_unref (sdata->msg);

  g_slice_free (struct sample_data, sdata);
}

static GstFlowReturn
new_sample_handler (GstElement *httpep, gpointer data)
{
  SoupMessage *msg = (SoupMessage *) g_object_get_data (G_OBJECT (httpep), KEY_MESSAGE);
  GstSample *sample = NULL;
  struct sample_data *sdata;

  GST_DEBUG ("New-sample for message %P", msg);

  g_signal_emit_by_name (httpep, "pull-sample", &sample);

  if (sample == NULL)
    return GST_FLOW_ERROR;

  sdata = g_slice_new (struct sample_data);
  sdata->sample = gst_sample_ref (sample);
  sdata->msg = (SoupMessage *) g_object_ref (G_OBJECT (msg) );
  sdata->httpepserver = KMS_HTTP_EP_SERVER (g_object_ref (
      g_object_get_data (G_OBJECT (msg), KEY_HTTP_EP_SERVER) ) );

  /* Write buffer in the main context thread */
  g_idle_add_full (G_PRIORITY_HIGH_IDLE, send_buffer_cb, sdata,
      destroy_sample_data);

  gst_sample_unref (sample);
  return GST_FLOW_OK;
}

static void
get_recv_eos (GstElement *httep, gpointer data)
{
  SoupMessage *msg = (SoupMessage *) data;

  GST_DEBUG ("EOS received on HttpEndPoint %s", GST_ELEMENT_NAME (httep) );
  soup_message_body_complete (msg->response_body);
}

static void
msg_finished (SoupMessage *msg)
{
  gboolean *finished;

  finished = (gboolean *) g_object_get_data (G_OBJECT (msg), KEY_FINISHED);
  *finished = TRUE;
}

static void
disconnect_eos_new_sample_signals (SoupMessage *msg)
{
  KmsHttpEPServer *serv = KMS_HTTP_EP_SERVER (
      g_object_get_data (G_OBJECT (msg), KEY_HTTP_EP_SERVER) );
  SoupURI *uri = soup_message_get_uri (msg);
  const char *path = soup_uri_get_path (uri);
  GstElement *httpep;
  gulong *handler;

  if (!g_hash_table_contains (serv->priv->handlers, path) ) {
    GST_WARNING ("Message %P was bounded to an unregistered HttpEndPoint", msg);
    return;
  }

  httpep = (GstElement *) g_hash_table_lookup (serv->priv->handlers, path);
  GST_DEBUG ("Message %P is bounded to %s", msg, GST_ELEMENT_NAME (httpep) );

  /* Disconnect signals */
  handler = (gulong *) g_object_get_data (G_OBJECT (msg),
      KEY_NEW_SAMPLE_HANDLER_ID);
  g_signal_handler_disconnect (httpep, *handler);

  handler = (gulong *) g_object_get_data (G_OBJECT (msg), KEY_EOS_HANDLER_ID);
  g_signal_handler_disconnect (httpep, *handler);
}

static void
finished_get_processing (SoupMessage *msg, gpointer data)
{
  GstElement *httpep = GST_ELEMENT (data);
  gpointer param;

  GST_DEBUG ("Message finished %P", msg);
{'fraction_non_alphanumeric': 0.1043700976468705, 'fraction_numerical': 0.0027213062269889547, 'mean_word_length': 4.969899665551839, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 55, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4717062', 'n_tokens_mistral': 4673, 'n_tokens_neox': 4578, 'n_words': 1072}