<commit_before>/*
 * Copyright (c) 2004-2006 <NAME> <<EMAIL>>
 * Copyright (c) 2006-2014 Sippy Software, Inc., http://www.sippysoft.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 */

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <assert.h>
#include <poll.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "rtpp_types.h"
#include "rtpp_weakref.h"
#include "rtp.h"
#include "rtp_info.h"
#include "rtp_packet.h"
#include "rtp_resizer.h"
#include "rtpp_log.h"
#include "rtpp_log_obj.h"
#include "rtpp_cfg_stable.h"
#include "rtpp_defines.h"
#include "rtpp_network.h"
#include "rtpp_notify.h"
#include "rtpp_netio_async.h"
#include "rtpp_proc.h"
#include "rtpp_record.h"
#include "rtpp_refcnt.h"
#include "rtpp_sessinfo.h"
#include "rtpp_stream.h"
#include "rtpp_session.h"
#include "rtpp_stats.h"
#include "rtpp_util.h"
#include "rtpp_analyzer.h"

struct rtpp_proc_ready_lst {
    struct rtpp_session_obj *sp;
    int ridx;
};

static void send_packet(struct cfg *, struct rtpp_session_obj *, int, \
  struct rtp_packet *, struct sthread_args *, struct rtpp_proc_rstats *);

static int
fill_session_addr(struct rtpp_session_obj *sp, struct rtp_packet *packet, int ridx)
{

    CALL_METHOD(sp->stream[ridx], fill_addr, packet);
    if (sp->rtcp == NULL) {
        return (0);
    }
    return (CALL_METHOD(sp->rtcp->stream[ridx], guess_addr, packet));
}

static struct rtpp_session_obj *
get_rtp(struct cfg *cf, struct rtpp_session_obj *sp)
{
    if (sp->rtcp != NULL) {
        return (sp);
    }
    return (CALL_METHOD(cf->stable->sessions_wrt, get_by_idx, sp->rtp_seuid));
}

static void
rxmit_packets(struct cfg *cf, struct rtpp_proc_ready_lst *rready, int rlen,
  double dtime, int drain_repeat, struct sthread_args *sender,
  struct rtpp_proc_rstats *rsp)
{
    int ndrain, rn, ridx;
    struct rtp_packet *packet = NULL;
    struct rtpp_session_obj *sp, *tsp;

    /* Repeat since we may have several packets queued on the same socket */
    ndrain = -1;
    for (rn = 0; rn < rlen; rn += (ndrain > 0) ? 0 : 1) {
        if (ndrain < 0) {
            ndrain = drain_repeat - 1;
        } else {
            ndrain -= 1;
        }
	if (packet != NULL)
	    rtp_packet_free(packet);

        sp = rready[rn].sp;
        ridx = rready[rn].ridx;

	packet = rtp_recv(sp->stream[ridx]->fd);
	if (packet == NULL) {
            /* Move on to the next session */
            ndrain = -1;
	    continue;
        }
	packet->laddr = sp->stream[ridx]->laddr;
	packet->rport = sp->stream[ridx]->port;
	packet->rtime = dtime;
        rsp->npkts_rcvd.cnt++;

	if (sp->stream[ridx]->addr != NULL) {
	    /* Check that the packet is authentic, drop if it isn't */
	    if (sp->stream[ridx]->asymmetric == 0) {
		if (memcmp(sp->stream[ridx]->addr, &packet->raddr, packet->rlen) != 0) {
		    if (sp->stream[ridx]->latch_info.latched != 0 && \
                      CALL_METHOD(sp->stream[ridx], check_latch_override, packet) == 0) {
			/*
			 * Continue, since there could be good packets in
			 * queue.
			 */
                        ndrain += 1;
                        sp->pcount.nignored++;
                        rsp->npkts_discard.cnt++;
			continue;
		    }
		    /* Signal that an address has to be updated */
		    fill_session_addr(sp, packet, ridx);
		} else if (sp->stream[ridx]->latch_info.latched == 0) {
                    CALL_METHOD(sp->stream[ridx], latch, dtime, packet);
		}
	    } else {
		/*
		 * For asymmetric clients don't check
		 * source port since it may be different.
		 */
		if (!ishostseq(sp->stream[ridx]->addr, sstosa(&packet->raddr))) {
		    /*
		     * Continue, since there could be good packets in
		     * queue.
		     */
                    ndrain += 1;
                    sp->pcount.nignored++;
                    rsp->npkts_discard.cnt++;
		    continue;
                }
	    }
	    sp->stream[ridx]->npkts_in++;
	} else {
	    sp->stream[ridx]->npkts_in++;
	    sp->stream[ridx]->addr = malloc(packet->rlen);
	    if (sp->stream[ridx]->addr == NULL) {
		sp->pcount.ndropped++;
		RTPP_LOG(sp->log, RTPP_LOG_ERR,
		  "can't allocate memory for remote address - "
		  "removing session");
                tsp = get_rtp(cf, sp);
                if (tsp != NULL) {
		    remove_session(cf, tsp);
                    if (tsp != sp)
                        CALL_METHOD(sp->rcnt, decref);
                }
		/* Move on to the next session, sp is invalid now */
                ndrain = -1;
                rsp->npkts_discard.cnt++;
		continue;
	    }
	    /* Update address recorded in the session */
	    fill_session_addr(sp, packet, ridx);
	}
        if (sp->stream[ridx]->analyzer != NULL) {
            rtpp_analyzer_update(sp, sp->stream[ridx]->analyzer, packet);
        }
	if (sp->stream[ridx]->resizer != NULL) {
	    rtp_resizer_enqueue(sp->stream[ridx]->resizer, &packet, rsp);
            if (packet == NULL) {
                rsp->npkts_resizer_in.cnt++;
            }
        }
	if (packet != NULL) {
	    send_packet(cf, sp, ridx, packet, sender, rsp);
            packet = NULL;
        }
    }
    if (packet != NULL)<commit_msg>Use return value of the rtpp_analyzer_update() to find our that the SSRC has changed and stream needs to be re-latched.<commit_after>/*
 * Copyright (c) 2004-2006 <NAME> <<EMAIL>>
 * Copyright (c) 2006-2014 Sippy Software, Inc., http://www.sippysoft.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 */

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <assert.h>
#include <poll.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "rtpp_types.h"
#include "rtpp_weakref.h"
#include "rtp.h"
#include "rtp_info.h"
#include "rtp_packet.h"
#include "rtp_resizer.h"
#include "rtpp_log.h"
#include "rtpp_log_obj.h"
#include "rtpp_cfg_stable.h"
#include "rtpp_defines.h"
#include "rtpp_network.h"
#include "rtpp_notify.h"
#include "rtpp_netio_async.h"
#include "rtpp_proc.h"
#include "rtpp_record.h"
#include "rtpp_refcnt.h"
#include "rtpp_sessinfo.h"
#include "rtpp_stream.h"
#include "rtpp_session.h"
#include "rtpp_stats.h"
#include "rtpp_util.h"
#include "rtp_analyze.h"
#include "rtpp_analyzer.h"

struct rtpp_proc_ready_lst {
    struct rtpp_session_obj *sp;
    int ridx;
};

static void send_packet(struct cfg *, struct rtpp_session_obj *, int, \
  struct rtp_packet *, struct sthread_args *, struct rtpp_proc_rstats *);

static int
fill_session_addr(struct rtpp_session_obj *sp, struct rtp_packet *packet, int ridx)
{

    CALL_METHOD(sp->stream[ridx], fill_addr, packet);
    if (sp->rtcp == NULL) {
        return (0);
    }
    return (CALL_METHOD(sp->rtcp->stream[ridx], guess_addr, packet));
}

static struct rtpp_session_obj *
get_rtp(struct cfg *cf, struct rtpp_session_obj *sp)
{
    if (sp->rtcp != NULL) {
        return (sp);
    }
    return (CALL_METHOD(cf->stable->sessions_wrt, get_by_idx, sp->rtp_seuid));
}

static void
rxmit_packets(struct cfg *cf, struct rtpp_proc_ready_lst *rready, int rlen,
  double dtime, int drain_repeat, struct sthread_args *sender,
  struct rtpp_proc_rstats *rsp)
{
    int ndrain, rn, ridx;
    struct rtp_packet *packet = NULL;
    struct rtpp_session_obj *sp, *tsp;

    /* Repeat since we may have several packets queued on the same socket */
    ndrain = -1;
    for (rn = 0; rn < rlen; rn += (ndrain > 0) ? 0 : 1) {
        if (ndrain < 0) {
            ndrain = drain_repeat - 1;
        } else {
            ndrain -= 1;
        }
	if (packet != NULL)
	    rtp_packet_free(packet);

        sp = rready[rn].sp;
        ridx = rready[rn].ridx;

	packet = rtp_recv(sp->stream[ridx]->fd);
	if (packet == NULL) {
            /* Move on to the next session */
            ndrain = -1;
	    continue;
        }
	packet->laddr = sp->stream[ridx]->laddr;
	packet->rport = sp->stream[ridx]->port;
	packet->rtime = dtime;
        rsp->npkts_rcvd.cnt++;

	if (sp->stream[ridx]->addr != NULL) {
	    /* Check that the packet is authentic, drop if it isn't */
	    if (sp->stream[ridx]->asymmetric == 0) {
		if (memcmp(sp->stream[ridx]->addr, &packet->raddr, packet->rlen) != 0) {
		    if (sp->stream[ridx]->latch_info.latched != 0 && \
                      CALL_METHOD(sp->stream[ridx], check_latch_override, packet) == 0) {
			/*
			 * Continue, since there could be good packets in
			 * queue.
			 */
                        ndrain += 1;
                        sp->pcount.nignored++;
                        rsp->npkts_discard.cnt++;
			continue;
		    }
		    /* Signal that an address has to be updated */
		    fill_session_addr(sp, packet, ridx);
		} else if (sp->stream[ridx]->latch_info.latched == 0) {
                    CALL_METHOD(sp->stream[ridx], latch, dtime, packet);
		}
	    } else {
		/*
		 * For asymmetric clients don't check
		 * source port since it may be different.
		 */
		if (!ishostseq(sp->stream[ridx]->addr, sstosa(&packet->raddr))) {
		    /*
		     * Continue, since there could be good packets in
		     * queue.
		     */
                    ndrain += 1;
                    sp->pcount.nignored++;
                    rsp->npkts_discard.cnt++;
		    continue;
                }
	    }
	    sp->stream[ridx]->npkts_in++;
	} else {
	    sp->stream[ridx]->npkts_in++;
	    sp->stream[ridx]->addr = malloc(packet->rlen);
	    if (sp->stream[ridx]->addr == NULL) {
		sp->pcount.ndropped++;
		RTPP_LOG(sp->log, RTPP_LOG_ERR,
		  "can't allocate memory for remote address - "
		  "removing session");
                tsp = get_rtp(cf, sp);
                if (tsp != NULL) {
		    remove_session(cf, tsp);
                    if (tsp != sp)
                        CALL_METHOD(sp->rcnt, decref);
                }
		/* Move on to the next session, sp is invalid now */
                ndrain = -1;
                rsp->npkts_discard.cnt++;
		continue;
	    }
	    /* Update address recorded in the session */
	    fill_session_addr(sp, packet, ridx);
	}
        if (sp->stream[ridx]->analyzer != NULL) {
            if (rtpp_analyzer_update(sp, sp->stream[ridx]->analyzer, packet) == UPDATE_SSRC_CHG) {
                CALL_METHOD(sp->stream[ridx], latch, dtime, packet);
            }
        }
	if (sp->stream[ridx]->resizer != NULL) {
	    rtp_resizer_enqueue(sp->stream[ridx]->resizer, &packet, rsp);
            if (packet == NULL) {
                rsp->npkts_resizer_in.cnt++;
            }
        }
	if (packet != NULL) {
	    send_packet(cf, sp, ridx, packet, sender, rsp);
            packet = NULL;
        }
    }
    if (packet != NULL)
{'fraction_non_alphanumeric': 0.1266141972950256, 'fraction_numerical': 0.005501642851684878, 'mean_word_length': 2.6774374824388874, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 152, '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': '1823994', 'n_tokens_mistral': 4663, 'n_tokens_neox': 4258, 'n_words': 1208}