<commit_before>/*
 *  Copyright (C) 2008 Pingtel Corp., certain elements licensed under a Contributor Agreement.
 *  Contributors retain copyright to elements licensed under a Contributor Agreement.
 *  Licensed to the User under the LGPL license.
 *
 */
package org.sipfoundry.sipxbridge.symmitron;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.util.TimerTask;

import org.apache.log4j.Logger;

/**
 * Transmitter endpoint.
 * 
 * @author mranga
 * 
 */
final class SymTransmitterEndpoint extends SymEndpoint implements SymTransmitterEndpointInterface {

    private long lastPacketSentTime;

    private ByteBuffer keepAliveBuffer = null;

    private ByteBuffer emptyBuffer = ByteBuffer.allocate(0);

    private AutoDiscoveryFlag remoteAddressAutoDiscovered = AutoDiscoveryFlag.NO_AUTO_DISCOVERY;

    private KeepaliveMethod keepaliveMethod = KeepaliveMethod.USE_EMPTY_PACKET;

    private KeepaliveTimerTask keepaliveTimerTask;

    private static Logger logger = Logger.getLogger(SymTransmitterEndpoint.class.getPackage()
            .getName());

    protected boolean onHold = false;

    private boolean earlyMediaStarted = false;

    private int maxSilence;

    long packetsSent;

    private static final boolean checkForSelfRouting = true;

    private InetSocketAddress socketAddress;

    private InetSocketAddress farEnd;

    /**
     * The keepalive timer task.
     * 
     */

    class KeepaliveTimerTask extends TimerTask {

        KeepaliveTimerTask() {

            logger.debug("Starting early media sender TimerTask : remote endpoint = "
                    + getIpAddress() + "/ " + port);

        }

        public void run() {

            if (datagramChannel == null)
                return;
            long now = System.currentTimeMillis();
            if (now - lastPacketSentTime < maxSilence) {

                return;
            }
            Bridge bridge = ConcurrentSet.getBridge(datagramChannel);
            if (bridge == null || bridge.getState() == BridgeState.INITIAL
                    || bridge.getState() == BridgeState.TERMINATED) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Bridge is not in running state " + bridge.getState());
                }
                return;
            }
            try {

                if (keepaliveMethod.equals(KeepaliveMethod.USE_EMPTY_PACKET)) {
                    if (datagramChannel != null && getSocketAddress() != null
                            && datagramChannel.isOpen() && getSocketAddress() != null) {
                        lastPacketSentTime = now;
                        datagramChannel.send(emptyBuffer, getSocketAddress());
                    }

                } else if (keepaliveMethod.equals(KeepaliveMethod.REPLAY_LAST_SENT_PACKET)
                        || keepaliveMethod.equals(KeepaliveMethod.USE_DUMMY_RTP_PAYLOAD)) {

                    if (keepAliveBuffer != null && datagramChannel != null
                            && getSocketAddress() != null && datagramChannel.isOpen()) {
                        logger.debug("Sending keepalive");
                        lastPacketSentTime = now;
                        datagramChannel.send((ByteBuffer) keepAliveBuffer, getSocketAddress());

                    }
                }
            } catch (ClosedChannelException ex) {
                /*
                 * This is not an error. Somebody closed socket. Just bail.
                 */
                logger.warn("Exitting early media thread due to closed channel", ex);
            } catch (Exception ex) {
                logger.error("Unexpected exception in sending early media ", ex);
                if ( bridge != null ) {
                    bridge.stop();
                }
            }

        }

    }

    private void setFarEnd() {
        try {
            InetAddress remoteAddress = this.getSocketAddress().getAddress();
            int remotePort = this.getSocketAddress().getPort();
            InetAddress farEndAddress = (SymmitronServer.getPublicInetAddress() != null
                    && remoteAddress.equals(SymmitronServer.getPublicInetAddress()) ? SymmitronServer
                    .getLocalInetAddress()
                    : remoteAddress);
            this.farEnd = new InetSocketAddress(farEndAddress, remotePort);
        } catch (Exception ex) {
            logger.error("Unexpected exception in setting far end address", ex);

        }

    }

    private static boolean isPacketSelfRouted(InetSocketAddress destinationSockaddr) {
        try {
            int port = destinationSockaddr.getPort();
            InetAddress destination = destinationSockaddr.getAddress();
            if (port >= SymmitronServer.getRangeLowerBound()
                    && port < SymmitronServer.getRangeUpperBound()) {
                if (!destination.equals(SymmitronServer.getLocalInetAddress())) {
                    return false;
                } else {
                    logger.trace("Self routed!");
                    return true;
                }
            } else {
                return false;
            }
        } catch (Exception ex) {
            logger.error("Unexpected exception",ex);
            throw new SymmitronException("Unexpected exception ", ex);
        }

    }

    private void startKeepaliveTimer() {

        if (earlyMediaStarted) {
            return;
        }
        logger.debug("startEarlyMediaThread " + this.toString());
        this.earlyMediaStarted = true;
        this.keepaliveTimerTask = new KeepaliveTimerTask();
        SymmitronServer.timer.schedule(this.keepaliveTimerTask, 0, this.maxSilence);

    }

    public SymTransmitterEndpoint() {
        super();
    }

    /**
     * @return the earlyMediaStarted
     */
    public boolean isKeepaliveStarted() {
        return earlyMediaStarted;
    }

    public void send(ByteBuffer byteBuffer) throws IOException {

        if (this.getSocketAddress() == null) {
            logger.debug("Cannot send -- remote address is null!");
            return;
        }
        this.lastPacketSentTime = System.currentTimeMillis();
<commit_msg>XECS-2088 : 0 byte sends from sipxrelay. Tighten up keepalive --  use AtomicLong for time recording. Initialize to current wallclock time.<commit_after>/*
 *  Copyright (C) 2008 Pingtel Corp., certain elements licensed under a Contributor Agreement.
 *  Contributors retain copyright to elements licensed under a Contributor Agreement.
 *  Licensed to the User under the LGPL license.
 *
 */
package org.sipfoundry.sipxbridge.symmitron;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;

/**
 * Transmitter endpoint.
 * 
 * @author mranga
 * 
 */
final class SymTransmitterEndpoint extends SymEndpoint implements SymTransmitterEndpointInterface {

    /*
     * Initialized to currentTime to prevent timer from kicking in too early.
     */
    private AtomicLong lastPacketSentTime = new AtomicLong(System.currentTimeMillis());

    private ByteBuffer keepAliveBuffer = null;

    private ByteBuffer emptyBuffer = ByteBuffer.allocate(0);

    private AutoDiscoveryFlag remoteAddressAutoDiscovered = AutoDiscoveryFlag.NO_AUTO_DISCOVERY;

    private KeepaliveMethod keepaliveMethod = KeepaliveMethod.USE_EMPTY_PACKET;

    private KeepaliveTimerTask keepaliveTimerTask;

    private static Logger logger = Logger.getLogger(SymTransmitterEndpoint.class.getPackage()
            .getName());

    protected boolean onHold = false;

    private boolean earlyMediaStarted = false;

    private int maxSilence;

    long packetsSent;

    private static final boolean checkForSelfRouting = true;

    private InetSocketAddress socketAddress;

    private InetSocketAddress farEnd;

    /**
     * The keepalive timer task.
     * 
     */

    class KeepaliveTimerTask extends TimerTask {

        KeepaliveTimerTask() {

            logger.debug("Starting early media sender TimerTask : remote endpoint = "
                    + getIpAddress() + "/ " + port);

        }

        public void run() {

            if (datagramChannel == null)
                return;
            long now = System.currentTimeMillis();
            if (now - lastPacketSentTime.get() < maxSilence) {

                return;
            }
            Bridge bridge = ConcurrentSet.getBridge(datagramChannel);
            if (bridge == null || bridge.getState() == BridgeState.INITIAL
                    || bridge.getState() == BridgeState.TERMINATED) {
                if (logger.isDebugEnabled()) {
                    logger.debug("Bridge is not in running state " + bridge.getState());
                }
                return;
            }
            try {

                if (keepaliveMethod.equals(KeepaliveMethod.USE_EMPTY_PACKET)) {
                    if (datagramChannel != null && getSocketAddress() != null
                            && datagramChannel.isOpen() && getSocketAddress() != null) {
                        lastPacketSentTime.set(now);
                        datagramChannel.send(emptyBuffer, getSocketAddress());
                    }

                } else if (keepaliveMethod.equals(KeepaliveMethod.REPLAY_LAST_SENT_PACKET)
                        || keepaliveMethod.equals(KeepaliveMethod.USE_DUMMY_RTP_PAYLOAD)) {

                    if (keepAliveBuffer != null && datagramChannel != null
                            && getSocketAddress() != null && datagramChannel.isOpen()) {
                        logger.debug("Sending keepalive");
                        lastPacketSentTime.set(now);
                        datagramChannel.send((ByteBuffer) keepAliveBuffer, getSocketAddress());

                    }
                }
            } catch (ClosedChannelException ex) {
                /*
                 * This is not an error. Somebody closed socket. Just bail.
                 */
                logger.warn("Exitting early media thread due to closed channel", ex);
            } catch (Exception ex) {
                logger.error("Unexpected exception in sending early media ", ex);
                if ( bridge != null ) {
                    bridge.stop();
                }
            }

        }

    }

    private void setFarEnd() {
        try {
            InetAddress remoteAddress = this.getSocketAddress().getAddress();
            int remotePort = this.getSocketAddress().getPort();
            InetAddress farEndAddress = (SymmitronServer.getPublicInetAddress() != null
                    && remoteAddress.equals(SymmitronServer.getPublicInetAddress()) ? SymmitronServer
                    .getLocalInetAddress()
                    : remoteAddress);
            this.farEnd = new InetSocketAddress(farEndAddress, remotePort);
        } catch (Exception ex) {
            logger.error("Unexpected exception in setting far end address", ex);

        }

    }

    private static boolean isPacketSelfRouted(InetSocketAddress destinationSockaddr) {
        try {
            int port = destinationSockaddr.getPort();
            InetAddress destination = destinationSockaddr.getAddress();
            if (port >= SymmitronServer.getRangeLowerBound()
                    && port < SymmitronServer.getRangeUpperBound()) {
                if (!destination.equals(SymmitronServer.getLocalInetAddress())) {
                    return false;
                } else {
                    logger.trace("Self routed!");
                    return true;
                }
            } else {
                return false;
            }
        } catch (Exception ex) {
            logger.error("Unexpected exception",ex);
            throw new SymmitronException("Unexpected exception ", ex);
        }

    }

    private void startKeepaliveTimer() {

        if (earlyMediaStarted) {
            return;
        }
        logger.debug("startEarlyMediaThread " + this.toString());
        this.earlyMediaStarted = true;
        this.keepaliveTimerTask = new KeepaliveTimerTask();
        SymmitronServer.timer.schedule(this.keepaliveTimerTask, 0, this.maxSilence);

    }

    public SymTransmitterEndpoint() {
        super();
    }

    /**
     * @return the earlyMediaStarted
     */
    public boolean isKeepaliveStarted() {
        return earlyMediaStarted;
    }

    public void send(ByteBuffer byteBuffer) throws IOException {

        if (this.getSocketAddress() == null) {
            if ( logger.isTraceEnabled() ) {
                logger.trace("Cannot send -- remote address is null!");
            }
            return;
        }
        
        /*
         * Record the time when packet is sent out.
         */
        this.lastPacketSentTime.set(System.currentTimeMillis());

{'fraction_non_alphanumeric': 0.08656761673741661, 'fraction_numerical': 0.001440266828380837, 'mean_word_length': 2.001820250284414, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, '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': '2018369', 'n_tokens_mistral': 3410, 'n_tokens_neox': 3198, 'n_words': 750}