package org.apache.commons.jcs.utils.discovery;

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.commons.jcs.engine.CacheInfo;
import org.apache.commons.jcs.engine.behavior.IShutdownObserver;
import org.apache.commons.jcs.io.ObjectInputStreamClassLoaderAware;
import org.apache.commons.jcs.utils.discovery.UDPDiscoveryMessage.BroadcastType;
import org.apache.commons.jcs.utils.threadpool.DaemonThreadFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/** Receives UDP Discovery messages. */
public class UDPDiscoveryReceiver
    implements Runnable, IShutdownObserver
{
    /** The log factory */
    private static final Log log = LogFactory.getLog( UDPDiscoveryReceiver.class );

    /** buffer */
    private final byte[] mBuffer = new byte[65536];

    /** The socket used for communication. */
    private MulticastSocket mSocket;

    /**
     * TODO: Consider using the threadpool manager to get this thread pool. For now place a tight
     * restriction on the pool size
     */
    private static final int maxPoolSize = 2;

    /** The processor */
    private ThreadPoolExecutor pooledExecutor = null;

    /** number of messages received. For debugging and testing. */
    private int cnt = 0;

    /** Service to get cache names and handle request broadcasts */
    private UDPDiscoveryService service = null;

    /** Address */
    private String multicastAddressString = "";

    /** The port */
    private int multicastPort = 0;

    /** Is it shutdown. */
    private boolean shutdown = false;

    /**
     * Constructor for the LateralUDPReceiver object.
     * <p>
     * We determine out own host using InetAddress
     *<p>
     * @param service
     * @param multicastAddressString
     * @param multicastPort
     * @throws IOException
     */
    public UDPDiscoveryReceiver( UDPDiscoveryService service, String multicastAddressString, int multicastPort )
        throws IOException
    {
        this.service = service;
        this.multicastAddressString = multicastAddressString;
        this.multicastPort = multicastPort;

        // create a small thread pool to handle a barrage
        pooledExecutor = (ThreadPoolExecutor)Executors.newFixedThreadPool(maxPoolSize,
                new DaemonThreadFactory("JCS-UDPDiscoveryReceiver-", Thread.MIN_PRIORITY));
        pooledExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        //pooledExecutor.setMinimumPoolSize(1);

        if ( log.isInfoEnabled() )
        {
            log.info( "Constructing listener, [" + this.multicastAddressString + ":" + this.multicastPort + "]" );
        }

        try
        {
            createSocket( this.multicastAddressString, this.multicastPort );
        }
        catch ( IOException ioe )
        {
            // consider eating this so we can go on, or constructing the socket
            // later
            throw ioe;
        }
    }

    /**
     * Creates the socket for this class.
     * <p>
     * @param multicastAddressString
     * @param multicastPort
     * @throws IOException
     */
    private void createSocket( String multicastAddressString, int multicastPort )
        throws IOException
    {
        try
        {
            mSocket = new MulticastSocket( multicastPort );
            if ( log.isInfoEnabled() )
            {
                log.info( "Joining Group: [" + InetAddress.getByName( multicastAddressString ) + "]" );
            }
            mSocket.joinGroup( InetAddress.getByName( multicastAddressString ) );
        }
        catch ( IOException e )
        {
            log.error( "Could not bind to multicast address [" + InetAddress.getByName( multicastAddressString ) + ":" + multicastPort + "]", e );
            throw e;
        }
    }

    /**
     * Highly unreliable. If it is processing one message while another comes in, the second
     * message is lost. This is for low concurrency peppering.
     * <p>
     * @return the object message
     * @throws IOException
     */
    public Object waitForMessage()
        throws IOException
    {
        final DatagramPacket packet = new DatagramPacket( mBuffer, mBuffer.length );
        ObjectInputStream objectStream = null;
        Object obj = null;
        try
        {
            if ( log.isDebugEnabled() )
            {
                log.debug( "Waiting for message." );
            }

            mSocket.receive( packet );

            if ( log.isDebugEnabled() )
            {
                log.debug( "Received packet from address [" + packet.getSocketAddress() + "]" );
            }

            final ByteArrayInputStream byteStream = new ByteArrayInputStream( mBuffer, 0, packet.getLength() );
            objectStream = new ObjectInputStreamClassLoaderAware( byteStream, null );
            obj = objectStream.readObject();

            if ( obj instanceof UDPDiscoveryMessage )
            {
                // Ensure that the address we're supposed to send to is, indeed, the address
                // of the machine on the other end of this connection.  This guards against
                // instances where we don't exactly get the right local host address
                UDPDiscoveryMessage msg = (UDPDiscoveryMessage) obj;
                msg.setHost(packet.getAddress().getHostAddress());

                    if ( log.isDebugEnabled() )
                    {
                        log.debug( "Read object from address [" + packet.getSocketAddress() + "], object=[" + obj + "]" );
                    }
            }
        }
        catch ( Exception e )
        {
            log.error( "Error receiving multicast packet", e );
        }
        finally
        {
            if (objectStream != null)
            {
                try
                {
                    objectStream.close();
                }
                catch (IOException e)
                {
                    log.error( "Error closing object stream", e );
                }
            }
        }
        return obj;
    }

    /** Main processing method for the LateralUDPReceiver object */
    @Override
    public void run()
    {
        try
        {
            while ( !shutdown )
            {
                Object obj = waitForMessage();

                // not thread safe, but just for debugging
                cnt++;

                if ( log.isDebugEnabled() )
                {
                    log.debug( getCnt() + " messages received." );
                }

                UDPDiscoveryMessage message = null;

                try
                {
                    message = (UDPDiscoveryMessage) obj;
                    // check for null
                    if ( message != null )
                    {
                        MessageHandler handler = new MessageHandler( message );

                        pooledExecutor.execute( handler );

                        if ( log.isDebugEnabled() )
                        {
                            log.debug( "Passed handler to executor." );
                        }
                    }
                    else
                    {
                        log.warn( "message is null" );
                    }
                }
                catch ( ClassCastException cce )
                {
                    log.warn( "Received unknown message type " + cce.getMessage() );
                }
            } // end while
        }
        catch ( Exception e )
        {
            log.error( "Unexpected exception in UDP receiver.", e );
            try
            {
                Thread.sleep( 100 );
                // TODO consider some failure count so we don't do this
                // forever.
            }
            catch ( Exception e2 )
            {
                log.error( "Problem sleeping", e2 );
            }
        }
    }

    /**
     * @param cnt The cnt to set.
     */
    public void setCnt( int cnt )
    {
        this.cnt = cnt;
    }

    /**
     * @return Returns the cnt.
     */
    public int getCnt()
    {
        return cnt;
    }

    /**
     * Separate thread run when a command comes into the UDPDiscoveryReceiver.
     */
    public class MessageHandler
        implements Runnable
    {
        /** The message to handle. Passed in during construction. */
        private UDPDiscoveryMessage message = null;

        /**
         * @param message
         */
        public MessageHandler( UDPDiscoveryMessage message )
        {
            this.message = message;
        }

        /**
         * Process the message.
         */
        @SuppressWarnings("synthetic-access")
        @Override
        public void run()
        {
            // consider comparing ports here instead.
            if ( message.getRequesterId() == CacheInfo.listenerId )
            {
                if ( log.isDebugEnabled() )
                {
                    log.debug( "Ignoring message sent from self" );
                }
            }
            else
            {
                if ( log.isDebugEnabled() )
                {
                    log.debug( "Process message sent from another" );
                    log.debug( "Message = " + message );
                }

                if ( message.getHost() == null || message.getCacheNames() == null || message.getCacheNames().isEmpty() )
                {
                    if ( log.isDebugEnabled() )
                    {
                        log.debug( "Ignoring invalid message: " + message );
                    }
                }
                else
                {
                    processMessage();
                }
            }
        }

        /**
         * Process the incoming message.
         */
        @SuppressWarnings("synthetic-access")
        private void processMessage()
        {
            DiscoveredService discoveredService = new DiscoveredService();
            discoveredService.setServiceAddress( message.getHost() );
            discoveredService.setCacheNames( message.getCacheNames() );
            discoveredService.setServicePort( message.getPort() );
            discoveredService.setLastHearFromTime( System.currentTimeMillis() );

            // if this is a request message, have the service handle it and
            // return
            if ( message.getMessageType() == BroadcastType.REQUEST )
            {
                if ( log.isDebugEnabled() )
                {
                    log.debug( "Message is a Request Broadcast, will have the service handle it." );
                }
                service.serviceRequestBroadcast();
                return;
            }
            else if ( message.getMessageType() == BroadcastType.REMOVE )
            {
                if ( log.isDebugEnabled() )
                {
                    log.debug( "Removing service from set " + discoveredService );
                }
                service.removeDiscoveredService( discoveredService );
            }
            else
            {
                service.addOrUpdateService( discoveredService );
            }
        }
    }

    /** Shuts down the socket. */
    @Override
    public void shutdown()
    {
        try
        {
            shutdown = true;
            mSocket.leaveGroup( InetAddress.getByName( multicastAddressString ) );
            mSocket.close();
            pooledExecutor.shutdownNow();
        }
        catch ( IOException e )
        {
            log.error( "Problem closing socket" );
        }
    }
}