<commit_before>/* * 
 *  This file is part of Feng
 * 
 * Copyright (C) 2008 by LScube team <<EMAIL>>
 * See AUTHORS for more details 
 *  
 * Feng is free software; you can redistribute it and/or 
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2 of
 * the License, or (at your option) any later version.
 * 
 * Feng 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 Feng; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 *  
 * */

/**
 * @file eventloop.c
 * Network main loop
 */

#include <stdio.h>
#include <sys/types.h>
#include <sys/select.h>
#include <unistd.h>

#include "eventloop.h"
#include "network/rtsp_utils.h"
#include <fenice/utils.h>

#include <fenice/schedule.h>

static GPtrArray *listen_socks;        //!< listen sockets
static fd_set rset;
static fd_set wset;
static int max_fd;
static GSList *clients; // of type RTSP_buffer

/**
 * Bind to the defined listening port
 */

int feng_bind_port(char *host, char *port, specific_config *s)
{
    int is_sctp = s->is_sctp;
    Sock *sock;

    if (is_sctp)
        sock = Sock_bind(host, port, NULL, SCTP, NULL);
    else
        sock = Sock_bind(host, port, NULL, TCP, NULL);
    if(!sock) {
        fnc_log(FNC_LOG_ERR,"Sock_bind() error for port %s.", port);
        fprintf(stderr,
                "[fatal] Sock_bind() error in main() for port %s.\n",
                port);
        return 1;
    }

    fnc_log(FNC_LOG_INFO, "Listening to port %s (%s) on %s",
            port,
            (is_sctp? "SCTP" : "TCP"),
            ((host == NULL)? "all interfaces" : host));

    g_ptr_array_add(listen_socks, sock);

    if(Sock_listen(sock, SOMAXCONN)) {
        fnc_log(FNC_LOG_ERR, "Sock_listen() error for TCP socket.");
        fprintf(stderr, "[fatal] Sock_listen() error for TCP socket.\n");
        return 1;
    }

    return 0;
}

static inline
int rtsp_sock_read(Sock *sock, int *stream, char *buffer, int size)
{
    int n;
#ifdef HAVE_LIBSCTP
    struct sctp_sndrcvinfo sctp_info;
    if (Sock_type(sock) == SCTP) {
        memset(&sctp_info, 0, sizeof(sctp_info));
        n = Sock_read(sock, buffer, size, &sctp_info, 0);
        *stream = sctp_info.sinfo_stream;
        fnc_log(FNC_LOG_DEBUG,
            "Sock_read() received %d bytes from sctp stream %d\n", n, stream);
    } else    // RTSP protocol is TCP
#endif    // HAVE_LIBSCTP

    n = Sock_read(sock, buffer, size, NULL, 0);

    return n;
}

#ifdef HAVE_LIBSCTP
#   define Sock_interleaved(a) \
     (Sock_type(a) == TCP || Sock_type(a) == SCTP)
#else
#   define Sock_interleaved(a) \
     (Sock_type(a) == TCP)
#endif

#ifdef HAVE_LIBSCTP
static gboolean find_sctp_interleaved(gconstpointer value, gconstpointer target)
{
  RTSP_interleaved *i = (RTSP_interleaved *)value;
  gint m = GPOINTER_TO_INT(target);

  return (i->proto.sctp.rtp.sinfo_stream == m || i->proto.sctp.rtcp.sinfo_stream == m);
}
#endif

static void interleaved_read(gpointer element, gpointer user_data)
{
  RTSP_buffer *rtsp = (RTSP_buffer *)user_data;
  RTSP_interleaved *intlvd = (RTSP_interleaved *)element;

  char buffer[RTSP_BUFFERSIZE + 1];    /* +1 to control the final '\0' */
  int n;
#ifdef HAVE_LIBSCTP
    struct sctp_sndrcvinfo sctp_info;
#endif

  if ( FD_ISSET(Sock_fd(intlvd->rtcp_local), &rset) ) {
    if ( (n = Sock_read(intlvd->rtcp_local, buffer, RTSP_BUFFERSIZE, NULL, 0)) < 0) {
      fnc_log(FNC_LOG_ERR, "Error reading from local socket\n");
      return;
    }
    switch (Sock_type(rtsp->sock)){
    case TCP: {
      uint16_t ne_n = htons((uint16_t)n);
      GString *str = g_string_sized_new(n+4);
      g_string_append_c(str, '$');
      g_string_append_c(str, (unsigned char)intlvd->proto.tcp.rtcp_ch);
      g_string_append_len(str, (gchar *)&ne_n, 2);
      g_string_append_len(str, (const gchar *)buffer, n);
      
      g_async_queue_push(rtsp->out_queue, str);
    }
    break;
#ifdef HAVE_LIBSCTP
    case SCTP:
      memcpy(&sctp_info, &(intlvd->proto.sctp.rtcp), sizeof(struct sctp_sndrcvinfo));
      Sock_write(rtsp->sock, buffer, n, &sctp_info, MSG_DONTWAIT | MSG_EOR);
      break;
#endif
    default:
      break;
    }
  }
  if ( FD_ISSET(Sock_fd(intlvd->rtp_local), &rset) ) {
    if ( (n = Sock_read(intlvd->rtp_local, buffer,
                        RTSP_BUFFERSIZE, NULL, 0)) < 0) {
      // if ( (n = read(intlvd->rtp_fd, intlvd->out_buffer, sizeof(intlvd->out_buffer))) < 0) {
      fnc_log(FNC_LOG_ERR, "Error reading from local socket\n");
      return;
    }
    switch (Sock_type(rtsp->sock)){
    case TCP: {
      uint16_t ne_n = htons((uint16_t)n);
      GString *str = g_string_sized_new(n+4);
      g_string_append_c(str, '$');
      g_string_append_c(str, (unsigned char)intlvd->proto.tcp.rtcp_ch);
      g_string_append_len(str, (gchar *)&ne_n, 2);
      g_string_append_len(str, (const gchar *)buffer, n);
      
      g_async_queue_push(rtsp->out_queue, str);
    }
#ifdef HAVE_LIBSCTP
    case SCTP:
      memcpy(&sctp_info, &(intlvd->proto.sctp.rtp),
	     sizeof(struct sctp_sndrcvinfo));
      Sock_write(rtsp->sock, buffer, n, &sctp_info, MSG_DONTWAIT | MSG_EOR);
      break;
#endif
    default:
      break;
    }<commit_msg>missing break...<commit_after>/* * 
 *  This file is part of Feng
 * 
 * Copyright (C) 2008 by LScube team <<EMAIL>>
 * See AUTHORS for more details 
 *  
 * Feng is free software; you can redistribute it and/or 
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2 of
 * the License, or (at your option) any later version.
 * 
 * Feng 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 Feng; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 *  
 * */

/**
 * @file eventloop.c
 * Network main loop
 */

#include <stdio.h>
#include <sys/types.h>
#include <sys/select.h>
#include <unistd.h>

#include "eventloop.h"
#include "network/rtsp_utils.h"
#include <fenice/utils.h>

#include <fenice/schedule.h>

static GPtrArray *listen_socks;        //!< listen sockets
static fd_set rset;
static fd_set wset;
static int max_fd;
static GSList *clients; // of type RTSP_buffer

/**
 * Bind to the defined listening port
 */

int feng_bind_port(char *host, char *port, specific_config *s)
{
    int is_sctp = s->is_sctp;
    Sock *sock;

    if (is_sctp)
        sock = Sock_bind(host, port, NULL, SCTP, NULL);
    else
        sock = Sock_bind(host, port, NULL, TCP, NULL);
    if(!sock) {
        fnc_log(FNC_LOG_ERR,"Sock_bind() error for port %s.", port);
        fprintf(stderr,
                "[fatal] Sock_bind() error in main() for port %s.\n",
                port);
        return 1;
    }

    fnc_log(FNC_LOG_INFO, "Listening to port %s (%s) on %s",
            port,
            (is_sctp? "SCTP" : "TCP"),
            ((host == NULL)? "all interfaces" : host));

    g_ptr_array_add(listen_socks, sock);

    if(Sock_listen(sock, SOMAXCONN)) {
        fnc_log(FNC_LOG_ERR, "Sock_listen() error for TCP socket.");
        fprintf(stderr, "[fatal] Sock_listen() error for TCP socket.\n");
        return 1;
    }

    return 0;
}

static inline
int rtsp_sock_read(Sock *sock, int *stream, char *buffer, int size)
{
    int n;
#ifdef HAVE_LIBSCTP
    struct sctp_sndrcvinfo sctp_info;
    if (Sock_type(sock) == SCTP) {
        memset(&sctp_info, 0, sizeof(sctp_info));
        n = Sock_read(sock, buffer, size, &sctp_info, 0);
        *stream = sctp_info.sinfo_stream;
        fnc_log(FNC_LOG_DEBUG,
            "Sock_read() received %d bytes from sctp stream %d\n", n, stream);
    } else    // RTSP protocol is TCP
#endif    // HAVE_LIBSCTP

    n = Sock_read(sock, buffer, size, NULL, 0);

    return n;
}

#ifdef HAVE_LIBSCTP
#   define Sock_interleaved(a) \
     (Sock_type(a) == TCP || Sock_type(a) == SCTP)
#else
#   define Sock_interleaved(a) \
     (Sock_type(a) == TCP)
#endif

#ifdef HAVE_LIBSCTP
static gboolean find_sctp_interleaved(gconstpointer value, gconstpointer target)
{
  RTSP_interleaved *i = (RTSP_interleaved *)value;
  gint m = GPOINTER_TO_INT(target);

  return (i->proto.sctp.rtp.sinfo_stream == m || i->proto.sctp.rtcp.sinfo_stream == m);
}
#endif

static void interleaved_read(gpointer element, gpointer user_data)
{
  RTSP_buffer *rtsp = (RTSP_buffer *)user_data;
  RTSP_interleaved *intlvd = (RTSP_interleaved *)element;

  char buffer[RTSP_BUFFERSIZE + 1];    /* +1 to control the final '\0' */
  int n;
#ifdef HAVE_LIBSCTP
    struct sctp_sndrcvinfo sctp_info;
#endif

  if ( FD_ISSET(Sock_fd(intlvd->rtcp_local), &rset) ) {
    if ( (n = Sock_read(intlvd->rtcp_local, buffer, RTSP_BUFFERSIZE, NULL, 0)) < 0) {
      fnc_log(FNC_LOG_ERR, "Error reading from local socket\n");
      return;
    }
    switch (Sock_type(rtsp->sock)){
    case TCP: {
      uint16_t ne_n = htons((uint16_t)n);
      GString *str = g_string_sized_new(n+4);
      g_string_append_c(str, '$');
      g_string_append_c(str, (unsigned char)intlvd->proto.tcp.rtcp_ch);
      g_string_append_len(str, (gchar *)&ne_n, 2);
      g_string_append_len(str, (const gchar *)buffer, n);

      g_async_queue_push(rtsp->out_queue, str);
    }
    break;
#ifdef HAVE_LIBSCTP
    case SCTP:
      memcpy(&sctp_info, &(intlvd->proto.sctp.rtcp), sizeof(struct sctp_sndrcvinfo));
      Sock_write(rtsp->sock, buffer, n, &sctp_info, MSG_DONTWAIT | MSG_EOR);
      break;
#endif
    default:
      break;
    }
  }
  if ( FD_ISSET(Sock_fd(intlvd->rtp_local), &rset) ) {
    if ( (n = Sock_read(intlvd->rtp_local, buffer,
                        RTSP_BUFFERSIZE, NULL, 0)) < 0) {
      // if ( (n = read(intlvd->rtp_fd, intlvd->out_buffer, sizeof(intlvd->out_buffer))) < 0) {
      fnc_log(FNC_LOG_ERR, "Error reading from local socket\n");
      return;
    }
    switch (Sock_type(rtsp->sock)){
    case TCP: {
      uint16_t ne_n = htons((uint16_t)n);
      GString *str = g_string_sized_new(n+4);
      g_string_append_c(str, '$');
      g_string_append_c(str, (unsigned char)intlvd->proto.tcp.rtcp_ch);
      g_string_append_len(str, (gchar *)&ne_n, 2);
      g_string_append_len(str, (const gchar *)buffer, n);

      g_async_queue_push(rtsp->out_queue, str);
    }
    break;
#ifdef HAVE_LIBSCTP
    case SCTP:
      memcpy(&sctp_info, &(intlvd->proto.sctp.rtp),
	     sizeof(struct sctp_sndrcvinfo));
      Sock_write(rtsp->sock, buffer, n, &sctp_info, MSG_DONTWAIT | MSG_EOR);
      break;
#endif
    default:
      break;
    }
{'fraction_non_alphanumeric': 0.12671480144404332, 'fraction_numerical': 0.007581227436823105, 'mean_word_length': 3.1347014925373133, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3691747', 'n_tokens_mistral': 4273, 'n_tokens_neox': 4115, 'n_words': 1028}