RTPConnectorOutputStream.java

/*
 * Jitsi, the OpenSource Java VoIP and Instant Messaging client.
 *
 * Distributable under LGPL license.
 * See terms of license at gnu.org.
 */
package org.jitsi.impl.neomedia;

import java.io.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;

import javax.media.rtp.*;

import org.jitsi.service.libjitsi.*;
import org.jitsi.service.packetlogging.*;
import org.jitsi.util.*;

/**
 * @author Bing SU (nova.su@gmail.com)
 * @author Lyubomir Marinov
 */
public abstract class RTPConnectorOutputStream
    implements OutputDataStream
{
    /**
     * The <tt>Logger</tt> used by the <tt>RTPConnectorOutputStream</tt> class
     * and its instances for logging output.
     */
    private static final Logger logger
        = Logger.getLogger(RTPConnectorOutputStream.class);

    /**
     * The maximum number of packets to be sent to be kept in the queue of
     * <tt>MaxPacketsPerMillisPolicy</tt>. When the maximum is reached, the next
     * attempt to write a new packet in the queue will block until at least one
     * packet from the queue is sent. Defined in order to prevent
     * <tt>OutOfMemoryError</tt>s which, technically, may arise if the capacity
     * of the queue is unlimited.
     */
    public static final int
        MAX_PACKETS_PER_MILLIS_POLICY_PACKET_QUEUE_CAPACITY
            = 256;

    /**
     * The functionality which allows this <tt>OutputDataStream</tt> to control
     * how many RTP packets it sends through its <tt>DatagramSocket</tt> per a
     * specific number of milliseconds.
     */
    private MaxPacketsPerMillisPolicy maxPacketsPerMillisPolicy;

    /**
     * Stream targets' IP addresses and ports.
     */
    protected final List<InetSocketAddress> targets
        = new LinkedList<InetSocketAddress>();

    /**
     * The pool of <tt>RawPacket</tt> instances which reduces the number of
     * allocations performed by {@link #createRawPacket(byte[], int, int)}.
     */
    private final LinkedBlockingQueue<RawPacket> rawPacketPool
        = new LinkedBlockingQueue<RawPacket>();

    /**
     * Used for debugging. As we don't log every packet
     * we must count them and decide which to log.
     */
    private long numberOfPackets = 0;

    /**
     * Initializes a new <tt>RTPConnectorOutputStream</tt> which is to send
     * packet data out through a specific socket.
     */
    public RTPConnectorOutputStream()
    {
    }

    /**
     * Add a target to stream targets list
     *
     * @param remoteAddr target ip address
     * @param remotePort target port
     */
    public void addTarget(InetAddress remoteAddr, int remotePort)
    {
        InetSocketAddress target
            = new InetSocketAddress(remoteAddr, remotePort);

        if (!targets.contains(target))
            targets.add(target);
    }

    /**
     * Close this output stream.
     */
    public void close()
    {
        if (maxPacketsPerMillisPolicy != null)
        {
            maxPacketsPerMillisPolicy.close();
            maxPacketsPerMillisPolicy = null;
        }
        removeTargets();
    }

    /**
     * Creates a new <tt>RawPacket</tt> from a specific <tt>byte[]</tt> buffer
     * in order to have this instance send its packet data through its
     * {@link #write(byte[], int, int)} method. Allows extenders to intercept
     * the packet data and possibly filter and/or modify it.
     *
     * @param buffer the packet data to be sent to the targets of this instance
     * @param offset the offset of the packet data in <tt>buffer</tt>
     * @param length the length of the packet data in <tt>buffer</tt>
     * @return a new <tt>RawPacket</tt> containing the packet data of the
     * specified <tt>byte[]</tt> buffer or possibly its modification;
     * <tt>null</tt> to ignore the packet data of the specified <tt>byte[]</tt>
     * buffer and not send it to the targets of this instance through its
     * {@link #write(byte[], int, int)} method
     */
    protected RawPacket createRawPacket(byte[] buffer, int offset, int length)
    {
        RawPacket pkt = rawPacketPool.poll();

        if ((pkt == null) || (pkt.getBuffer().length < length))
            pkt = new RawPacket(new byte[length], 0, 0);

        System.arraycopy(buffer, offset, pkt.getBuffer(), 0, length);
        pkt.setLength(length);
        pkt.setOffset(0);
        return pkt;
    }

    /**
     * Remove a target from stream targets list
     *
     * @param remoteAddr target ip address
     * @param remotePort target port
     * @return <tt>true</tt> if the target is in stream target list and can be
     * removed; <tt>false</tt>, otherwise
     */
    public boolean removeTarget(InetAddress remoteAddr, int remotePort)
    {
        for (Iterator<InetSocketAddress> targetIter = targets.iterator();
                targetIter.hasNext();)
        {
            InetSocketAddress target = targetIter.next();

            if (target.getAddress().equals(remoteAddr)
                    && (target.getPort() == remotePort))
            {
                targetIter.remove();
                return true;
            }
        }
        return false;
    }

    /**
     * Remove all stream targets from this session.
     */
    public void removeTargets()
    {
        targets.clear();
    }

    /**
     * Determines whether a <tt>RawPacket</tt> which has a specific number in
     * the total number of sent <tt>RawPacket</tt>s is to be logged by
     * {@link PacketLoggingService}.
     *
     * @param numOfPacket the number of the <tt>RawPacket</tt> in the total
     * number of sent <tt>RawPacket</tt>s
     * @return <tt>true</tt> if the <tt>RawPacket</tt> with the specified
     * <tt>numOfPacket</tt> is to be logged by <tt>PacketLoggingService</tt>;
     * otherwise, <tt>false</tt>
     */
    static boolean logPacket(long numOfPacket)
    {
        return
            (numOfPacket == 1)
                || (numOfPacket == 300)
                || (numOfPacket == 500)
                || (numOfPacket == 1000)
                || ((numOfPacket % 5000) == 0);
    }

    /**
     * Sends a specific <tt>RawPacket</tt> through this
     * <tt>OutputDataStream</tt> to a specific <tt>InetSocketAddress</tt>.
     *
     * @param packet the <tt>RawPacket</tt> to send through this
     * <tt>OutputDataStream</tt> to the specified <tt>target</tt>
     * @param target the <tt>InetSocketAddress</tt> to which the specified
     * <tt>packet</tt> is to be sent through this <tt>OutputDataStream</tt>
     * @throws IOException if anything goes wrong while sending the specified
     * <tt>packet</tt> through this <tt>OutputDataStream</tt> to the specified
     * <tt>target</tt>
     */
    protected abstract void sendToTarget(
            RawPacket packet,
            InetSocketAddress target)
        throws IOException;

    /**
     * Logs a specific <tt>RawPacket</tt> associated with a specific remote
     * address.
     *
     * @param packet packet to log
     * @param target the remote address associated with the <tt>packet</tt>
     */
    protected abstract void doLogPacket(
        RawPacket packet,
        InetSocketAddress target);

    /**
     * Returns whether or not this <tt>RTPConnectorOutputStream</tt> has a valid
     * socket.
     *
     * @return <tt>true</tt> if this <tt>RTPConnectorOutputStream</tt> has a
     * valid socket; <tt>false</tt>, otherwise
     */
    protected abstract boolean isSocketValid();

    /**
     * Sends a specific RTP packet through the <tt>DatagramSocket</tt> of this
     * <tt>OutputDataSource</tt>.
     *
     * @param packet the RTP packet to be sent through the
     * <tt>DatagramSocket</tt> of this <tt>OutputDataSource</tt>
     * @return <tt>true</tt> if the specified <tt>packet</tt> was successfully
     * sent; otherwise, <tt>false</tt>
     */
    private boolean send(RawPacket packet)
    {
        if(!isSocketValid())
            return false;

        numberOfPackets++;
        for (InetSocketAddress target : targets)
        {
            try
            {
                sendToTarget(packet, target);

                if(logPacket(numberOfPackets))
                {
                    PacketLoggingService packetLogging
                        = LibJitsi.getPacketLoggingService();

                    if ((packetLogging != null)
                            && packetLogging.isLoggingEnabled(
                                    PacketLoggingService.ProtocolName.RTP))
                        doLogPacket(packet, target);
                }
            }
            catch (IOException ioe)
            {
                rawPacketPool.offer(packet);
                // TODO error handling
                return false;
            }
        }
        rawPacketPool.offer(packet);
        return true;
    }

    /**
     * Sets the maximum number of RTP packets to be sent by this
     * <tt>OutputDataStream</tt> through its <tt>DatagramSocket</tt> per
     * a specific number of milliseconds.
     *
     * @param maxPackets the maximum number of RTP packets to be sent by this
     * <tt>OutputDataStream</tt> through its <tt>DatagramSocket</tt> per the
     * specified number of milliseconds; <tt>-1</tt> if no maximum is to be set
     * @param perMillis the number of milliseconds per which <tt>maxPackets</tt>
     * are to be sent by this <tt>OutputDataStream</tt> through its
     * <tt>DatagramSocket</tt>
     */
    public void setMaxPacketsPerMillis(int maxPackets, long perMillis)
    {
        if (maxPacketsPerMillisPolicy == null)
        {
            if (maxPackets > 0)
            {
                if (perMillis < 1)
                    throw new IllegalArgumentException("perMillis");

                maxPacketsPerMillisPolicy
                    = new MaxPacketsPerMillisPolicy(maxPackets, perMillis);
            }
        }
        else
        {
            maxPacketsPerMillisPolicy
                .setMaxPacketsPerMillis(maxPackets, perMillis);
        }
    }

    /**
     * Implements {@link OutputDataStream#write(byte[], int, int)}.
     *
     * @param buffer the <tt>byte[]</tt> that we'd like to copy the content
     * of the packet to.
     * @param offset the position where we are supposed to start writing in
     * <tt>buffer</tt>.
     * @param length the number of <tt>byte</tt>s available for writing in
     * <tt>inBuffer</tt>.
     *
     * @return the number of bytes read
     */
    public int write(byte[] buffer, int offset, int length)
    {
        /*
         * While calling write without targets can be carried out without a
         * problem, such a situation may be a symptom of a problem. For example,
         * it was discovered during testing that RTCP was seemingly-endlessly
         * sent after hanging up a call.
         */
        if (logger.isDebugEnabled() && targets.isEmpty())
            logger.debug("Write called without targets!", new Throwable());

        RawPacket packet = createRawPacket(buffer, offset, length);

        /*
         * If we got extended, the delivery of the packet may have been
         * canceled.
         */
        if (packet != null)
        {
            if (maxPacketsPerMillisPolicy == null)
            {
                if (!send(packet))
                    return -1;
            }
            else
                maxPacketsPerMillisPolicy.write(packet);
        }
        return length;
    }

    /**
     * Changes current thread priority.
     * @param priority the new priority.
     */
    public void setPriority(int priority)
    {
        // currently no priority is set
//        if ((maxPacketsPerMillisPolicy != null)
//                && (maxPacketsPerMillisPolicy.sendThread != null))
//            maxPacketsPerMillisPolicy.sendThread.setPriority(priority);
    }

    /**
     * Implements the functionality which allows this <tt>OutputDataStream</tt>
     * to control how many RTP packets it sends through its
     * <tt>DatagramSocket</tt> per a specific number of milliseconds.
     */
    private class MaxPacketsPerMillisPolicy
    {

        /**
         * The maximum number of RTP packets to be sent by this
         * <tt>OutputDataStream</tt> through its <tt>DatagramSocket</tt> per
         * {@link #perNanos} nanoseconds.
         */
        private int maxPackets = -1;

        /**
         * The time stamp in nanoseconds of the start of the current
         * <tt>perNanos</tt> interval.
         */
        private long millisStartTime = 0;

        /**
         * The list of RTP packets to be sent through the
         * <tt>DatagramSocket</tt> of this <tt>OutputDataSource</tt>.
         */
        private final ArrayBlockingQueue<RawPacket> packetQueue
            = new ArrayBlockingQueue<RawPacket>(
                    MAX_PACKETS_PER_MILLIS_POLICY_PACKET_QUEUE_CAPACITY);

        /**
         * The number of RTP packets already sent during the current
         * <tt>perNanos</tt> interval.
         */
        private long packetsSentInMillis = 0;

        /**
         * The time interval in nanoseconds during which {@link #maxPackets}
         * number of RTP packets are to be sent through the
         * <tt>DatagramSocket</tt> of this <tt>OutputDataSource</tt>.
         */
        private long perNanos = -1;

        /**
         * The <tt>Thread</tt> which is to send the RTP packets in
         * {@link #packetQueue} through the <tt>DatagramSocket</tt> of this
         * <tt>OutputDataSource</tt>.
         */
        private Thread sendThread;

        /**
         * To signal run or stop condition to send thread.
         */
        private boolean sendRun = true;

        /**
         * Initializes a new <tt>MaxPacketsPerMillisPolicy</tt> instance which
         * is to control how many RTP packets this <tt>OutputDataSource</tt> is
         * to send through its <tt>DatagramSocket</tt> per a specific number of
         * milliseconds.
         *
         * @param maxPackets the maximum number of RTP packets to be sent per
         * <tt>perMillis</tt> milliseconds through the <tt>DatagramSocket</tt>
         * of this <tt>OutputDataStream</tt>
         * @param perMillis the number of milliseconds per which a maximum of
         * <tt>maxPackets</tt> RTP packets are to be sent through the
         * <tt>DatagramSocket</tt> of this <tt>OutputDataStream</tt>
         */
        public MaxPacketsPerMillisPolicy(int maxPackets, long perMillis)
        {
            setMaxPacketsPerMillis(maxPackets, perMillis);
            synchronized (this) {
                if (sendThread == null)
                {
                    sendThread
                        = new Thread(getClass().getName())
                        {
                            @Override
                            public void run()
                            {
                                runInSendThread();
                            }
                        };
                    sendThread.setDaemon(true);
                    sendThread.start();
                }
            }
        }

        /**
         * Closes the connector.
         */
        synchronized void close()
        {
            if (!sendRun)
                return;
            sendRun = false;
            // just offer a new packet to wakeup thread in case it waits for
            // a packet.
            packetQueue.offer(new RawPacket(null, 0, 0));
        }

        /**
         * Sends the RTP packets in {@link #packetQueue} in accord with
         * {@link #maxPackets} and {@link #perNanos}.
         */
        private void runInSendThread()
        {
            try
            {
                while (sendRun)
                {
                    RawPacket packet = null;

                    while (true)
                    {
                        try
                        {
                            packet = packetQueue.take();
                            break;
                        }
                        catch (InterruptedException iex)
                        {
                            continue;
                        }
                    }
                    if (!sendRun)
                        break;

                    long time = System.nanoTime();
                    long millisRemainingTime = time - millisStartTime;

                    if ((perNanos < 1)
                            || (millisRemainingTime >= perNanos))
                    {
                        millisStartTime = time;
                        packetsSentInMillis = 0;
                    }
                    else if ((maxPackets > 0)
                            && (packetsSentInMillis >= maxPackets))
                    {
                        while (true)
                        {
                            millisRemainingTime = System.nanoTime()
                                    - millisStartTime;
                            if (millisRemainingTime >= perNanos)
                                break;
                            LockSupport.parkNanos(millisRemainingTime);
                        }
                        millisStartTime = System.nanoTime();
                        packetsSentInMillis = 0;
                    }

                    send(packet);
                    packetsSentInMillis++;
                }
            }
            finally
            {
                packetQueue.clear();
                synchronized (packetQueue)
                {
                    if (Thread.currentThread().equals(sendThread))
                        sendThread = null;
                }
            }
        }

        /**
         * Sets the maximum number of RTP packets to be sent by this
         * <tt>OutputDataStream</tt> through its <tt>DatagramSocket</tt> per
         * a specific number of milliseconds.
         *
         * @param maxPackets the maximum number of RTP packets to be sent by
         * this <tt>OutputDataStream</tt> through its <tt>DatagramSocket</tt>
         * per the specified number of milliseconds; <tt>-1</tt> if no maximum
         * is to be set
         * @param perMillis the number of milliseconds per which
         * <tt>maxPackets</tt> are to be sent by this <tt>OutputDataStream</tt>
         * through its <tt>DatagramSocket</tt>
         */
        public void setMaxPacketsPerMillis(int maxPackets, long perMillis)
        {
            if (maxPackets < 1)
            {
                this.maxPackets = -1;
                this.perNanos = -1;
            }
            else
            {
                if (perMillis < 1)
                    throw new IllegalArgumentException("perMillis");

                this.maxPackets = maxPackets;
                this.perNanos = perMillis * 1000000;
            }
        }

        /**
         * Queues a specific RTP packet to be sent through the
         * <tt>DatagramSocket</tt> of this <tt>OutputDataStream</tt>.
         *
         * @param packet the RTP packet to be queued for sending through the
         * <tt>DatagramSocket</tt> of this <tt>OutputDataStream</tt>
         */
        public void write(RawPacket packet)
        {
            while (true)
            {
                try
                {
                    packetQueue.put(packet);
                    break;
                }
                catch (InterruptedException iex)
                {
                }
            }
        }
    }
}