<commit_before>/*
 * (C) Copyright 2014 Kurento (http://kurento.org/)
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser General Public License
 * (LGPL) version 2.1 which accompanies this distribution, and is available at
 * http://www.gnu.org/licenses/lgpl-2.1.html
 *
 * 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
 * Lesser General Public License for more details.
 *
 */

#include <gst/gst.h>
#include "RabbitMQConnection.hpp"
#include <amqp.h>
#include <amqp_tcp_socket.h>
#include <gst/gst.h>
#include <unistd.h>

#include <sys/types.h>
#include <sys/socket.h>

#define GST_CAT_DEFAULT kurento_rabbitmq_connection
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
#define GST_DEFAULT_NAME "KurentoRabbitMQConnection"

namespace kurento
{

static void
exception_on_error (amqp_rpc_reply_t x, const char *context)
{
  std::string ctx = context;

  switch (x.reply_type) {
  case AMQP_RESPONSE_NORMAL:
    return;

  case AMQP_RESPONSE_NONE:
    throw RabbitMQException (ctx + ": missing RPC reply type");

  case AMQP_RESPONSE_LIBRARY_EXCEPTION:
    if (AMQP_STATUS_TIMEOUT == x.library_error) {
      throw RabbitMQTimeoutException (ctx + ": Tiemout exception: " +
                                      amqp_error_string2 (x.library_error) );
    } else {
      throw RabbitMQException (ctx + ": Library exception: " +
                               amqp_error_string2 (x.library_error) );
    }

  case AMQP_RESPONSE_SERVER_EXCEPTION:
    switch (x.reply.id) {
    case AMQP_CONNECTION_CLOSE_METHOD: {
      amqp_connection_close_t *m = (amqp_connection_close_t *) x.reply.decoded;
      std::string message;
      gchar *error_message;

      error_message =
        g_strdup_printf ("%s: server connection error %d, message: %.*s\n", context,
                         m->reply_code, (int) m->reply_text.len, (char *) m->reply_text.bytes);

      message = error_message;
      g_free (error_message);

      throw RabbitMQException (message);
    }

    case AMQP_CHANNEL_CLOSE_METHOD: {
      amqp_channel_close_t *m = (amqp_channel_close_t *) x.reply.decoded;
      std::string message;
      gchar *error_message;

      error_message = g_strdup_printf ("%s: server channel error %d, message: %.*s\n",
                                       context, m->reply_code, (int) m->reply_text.len, (char *) m->reply_text.bytes);

      message = error_message;
      g_free (error_message);

      throw RabbitMQException (message);
      break;
    }

    default:
      throw RabbitMQException (ctx + ": Channel Unknown Error");
      break;
    }

    break;
  }
}

static void
makeSocketLinger (int fd)
{
  struct linger ling;

  ling.l_onoff = 1;
  ling.l_linger = 30;

  if (setsockopt (fd, SOL_SOCKET, SO_LINGER, &ling, sizeof (ling) ) < 0) {
    GST_WARNING ("Could not configure SO_LINGER option on RabbitMQ socket");
  }
}

RabbitMQConnection::RabbitMQConnection (const std::string &address, int port) :
  address (address), port (port)
{
  conn = amqp_new_connection();

  socket = amqp_tcp_socket_new (conn);

  if (!socket) {
    throw Glib::IOChannelError (Glib::IOChannelError::Code::FAILED,
                                "Cannot create TCP socket");
  }

  if (amqp_socket_open (socket, address.c_str(), port) ) {
    throw Glib::IOChannelError (Glib::IOChannelError::Code::FAILED,
                                "Cannot open TCP socket");
  }

  makeSocketLinger (amqp_socket_get_sockfd (socket) );

  exception_on_error (amqp_login (conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                                  "guest", "guest"), "Loging in");
  amqp_channel_open (conn, 1);
  exception_on_error (amqp_get_rpc_reply (conn), "Opening channel");
}

RabbitMQConnection::~RabbitMQConnection()
{
  if (conn == NULL) {
    GST_DEBUG ("service already stopped");
    return;
  }

  /* Errors are ignored during close */
  if (!closeOnRelease) {
    int fd = amqp_socket_get_sockfd (socket);

    /* inform remote side that we are done */
    shutdown (fd, SHUT_WR);

    /* close socket */
    close (fd);
  }

  amqp_channel_close (conn, 1, AMQP_REPLY_SUCCESS);
  amqp_connection_close (conn, AMQP_REPLY_SUCCESS);
  amqp_destroy_connection (conn);
  conn = NULL;
}

int
RabbitMQConnection::getFd()
{
  return amqp_socket_get_sockfd (socket);
}<commit_msg>RabbitMQ: Do not shutdown socket if closeOnRelease is not set<commit_after>/*
 * (C) Copyright 2014 Kurento (http://kurento.org/)
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser General Public License
 * (LGPL) version 2.1 which accompanies this distribution, and is available at
 * http://www.gnu.org/licenses/lgpl-2.1.html
 *
 * 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
 * Lesser General Public License for more details.
 *
 */

#include <gst/gst.h>
#include "RabbitMQConnection.hpp"
#include <amqp.h>
#include <amqp_tcp_socket.h>
#include <gst/gst.h>
#include <unistd.h>

#include <sys/types.h>
#include <sys/socket.h>

#define GST_CAT_DEFAULT kurento_rabbitmq_connection
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
#define GST_DEFAULT_NAME "KurentoRabbitMQConnection"

namespace kurento
{

static void
exception_on_error (amqp_rpc_reply_t x, const char *context)
{
  std::string ctx = context;

  switch (x.reply_type) {
  case AMQP_RESPONSE_NORMAL:
    return;

  case AMQP_RESPONSE_NONE:
    throw RabbitMQException (ctx + ": missing RPC reply type");

  case AMQP_RESPONSE_LIBRARY_EXCEPTION:
    if (AMQP_STATUS_TIMEOUT == x.library_error) {
      throw RabbitMQTimeoutException (ctx + ": Tiemout exception: " +
                                      amqp_error_string2 (x.library_error) );
    } else {
      throw RabbitMQException (ctx + ": Library exception: " +
                               amqp_error_string2 (x.library_error) );
    }

  case AMQP_RESPONSE_SERVER_EXCEPTION:
    switch (x.reply.id) {
    case AMQP_CONNECTION_CLOSE_METHOD: {
      amqp_connection_close_t *m = (amqp_connection_close_t *) x.reply.decoded;
      std::string message;
      gchar *error_message;

      error_message =
        g_strdup_printf ("%s: server connection error %d, message: %.*s\n", context,
                         m->reply_code, (int) m->reply_text.len, (char *) m->reply_text.bytes);

      message = error_message;
      g_free (error_message);

      throw RabbitMQException (message);
    }

    case AMQP_CHANNEL_CLOSE_METHOD: {
      amqp_channel_close_t *m = (amqp_channel_close_t *) x.reply.decoded;
      std::string message;
      gchar *error_message;

      error_message = g_strdup_printf ("%s: server channel error %d, message: %.*s\n",
                                       context, m->reply_code, (int) m->reply_text.len, (char *) m->reply_text.bytes);

      message = error_message;
      g_free (error_message);

      throw RabbitMQException (message);
      break;
    }

    default:
      throw RabbitMQException (ctx + ": Channel Unknown Error");
      break;
    }

    break;
  }
}

static void
makeSocketLinger (int fd)
{
  struct linger ling;

  ling.l_onoff = 1;
  ling.l_linger = 30;

  if (setsockopt (fd, SOL_SOCKET, SO_LINGER, &ling, sizeof (ling) ) < 0) {
    GST_WARNING ("Could not configure SO_LINGER option on RabbitMQ socket");
  }
}

RabbitMQConnection::RabbitMQConnection (const std::string &address, int port) :
  address (address), port (port)
{
  conn = amqp_new_connection();

  socket = amqp_tcp_socket_new (conn);

  if (!socket) {
    throw Glib::IOChannelError (Glib::IOChannelError::Code::FAILED,
                                "Cannot create TCP socket");
  }

  if (amqp_socket_open (socket, address.c_str(), port) ) {
    throw Glib::IOChannelError (Glib::IOChannelError::Code::FAILED,
                                "Cannot open TCP socket");
  }

  makeSocketLinger (amqp_socket_get_sockfd (socket) );

  exception_on_error (amqp_login (conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                                  "guest", "guest"), "Loging in");
  amqp_channel_open (conn, 1);
  exception_on_error (amqp_get_rpc_reply (conn), "Opening channel");
}

RabbitMQConnection::~RabbitMQConnection()
{
  int fd;

  if (conn == NULL) {
    GST_DEBUG ("service already stopped");
    return;
  }

  fd = amqp_socket_get_sockfd (socket);

  /* Errors are ignored during close */
  if (!closeOnRelease) {
    /* close socket */
    close (fd);
  }

  amqp_channel_close (conn, 1, AMQP_REPLY_SUCCESS);
  amqp_connection_close (conn, AMQP_REPLY_SUCCESS);
  amqp_destroy_connection (conn);

  if (closeOnRelease) {
    /* inform remote side that we are done */
    shutdown (fd, SHUT_WR);
  }

  conn = NULL;
}

int
RabbitMQConnection::getFd()
{
  return amqp_socket_get_sockfd (socket);
}
{'fraction_non_alphanumeric': 0.1078107810781078, 'fraction_numerical': 0.005280528052805281, 'mean_word_length': 3.175930179145613, 'pattern_counts': {'":': 8, '<': 19, '<?xml version=': 0, '>': 29, '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': '5310207', 'n_tokens_mistral': 3152, 'n_tokens_neox': 2963, 'n_words': 748}