package org.apache.commons.jcs3.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.IOException;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.StandardProtocolFamily;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.MembershipKey;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.jcs3.engine.CacheInfo;
import org.apache.commons.jcs3.engine.behavior.IElementSerializer;
import org.apache.commons.jcs3.engine.behavior.IShutdownObserver;
import org.apache.commons.jcs3.log.Log;
import org.apache.commons.jcs3.log.LogManager;
import org.apache.commons.jcs3.utils.net.HostNameUtil;
import org.apache.commons.jcs3.utils.threadpool.PoolConfiguration;
import org.apache.commons.jcs3.utils.threadpool.PoolConfiguration.WhenBlockedPolicy;
import org.apache.commons.jcs3.utils.threadpool.ThreadPoolManager;

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

    /** The channel used for communication. */
    private DatagramChannel multicastChannel;

    /** The group membership key. */
    private MembershipKey multicastGroupKey;

    /** The selector. */
    private Selector selector;

    /**
     * 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 final ExecutorService pooledExecutor;

    /** number of messages received. For debugging and testing. */
    private final AtomicInteger cnt = new AtomicInteger(0);

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

    /** Serializer */
    private IElementSerializer serializer;

    /** Is it shutdown. */
    private final AtomicBoolean shutdown = new AtomicBoolean(false);

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

    /**
     * Constructor for the UDPDiscoveryReceiver object.
     * <p>
     * @param service
     * @param multicastInterfaceString
     * @param multicastAddress
     * @param multicastPort
     * @throws IOException
     * @since 3.1
     */
    public UDPDiscoveryReceiver( final UDPDiscoveryService service,
            final String multicastInterfaceString,
            final InetAddress multicastAddress,
            final int multicastPort )
        throws IOException
    {
        this.service = service;
        if (service != null)
        {
            this.serializer = service.getSerializer();
        }

        // create a small thread pool to handle a barrage
        this.pooledExecutor = ThreadPoolManager.getInstance().createPool(
                new PoolConfiguration(false, 0, maxPoolSize, maxPoolSize, 0,
                        WhenBlockedPolicy.DISCARDOLDEST, maxPoolSize),
                "JCS-UDPDiscoveryReceiver-", Thread.MIN_PRIORITY);

        log.info( "Constructing listener, [{0}:{1}]", multicastAddress, multicastPort );
        createSocket( multicastInterfaceString, multicastAddress, multicastPort );
    }

    /**
     * Creates the socket for this class.
     * <p>
     * @param multicastInterfaceString
     * @param multicastAddress
     * @param multicastPort
     * @throws IOException
     */
    private void createSocket( final String multicastInterfaceString, final InetAddress multicastAddress,
            final int multicastPort )
        throws IOException
    {
        try
        {
            // Use dedicated interface if specified
            NetworkInterface multicastInterface = null;
            if (multicastInterfaceString != null)
            {
                multicastInterface = NetworkInterface.getByName(multicastInterfaceString);
            }
            else
            {
                multicastInterface = HostNameUtil.getMulticastNetworkInterface();
            }
            if (multicastInterface != null)
            {
                log.info("Using network interface {0}", multicastInterface::getDisplayName);
            }

            multicastChannel = DatagramChannel.open(
                    multicastAddress instanceof Inet6Address ?
                            StandardProtocolFamily.INET6 : StandardProtocolFamily.INET)
                    .setOption(StandardSocketOptions.SO_REUSEADDR, true)
                    .setOption(StandardSocketOptions.IP_MULTICAST_IF, multicastInterface)
                    .bind(new InetSocketAddress(multicastPort));
            multicastChannel.configureBlocking(false);

            log.info("Joining Group: [{0}] on {1}", multicastAddress, multicastInterface);
            multicastGroupKey = multicastChannel.join(multicastAddress, multicastInterface);

            selector = Selector.open();
            multicastChannel.register(selector, SelectionKey.OP_READ);
        }
        catch ( final IOException e )
        {
            log.error( "Could not bind to multicast address [{0}:{1}]", multicastAddress,
                    multicastPort, e );
            throw e;
        }
    }

    private final ArrayBlockingQueue<UDPDiscoveryMessage> msgQueue =
            new ArrayBlockingQueue<>(maxPoolSize);

    /**
     * Wait for multicast message
     *
     * @return the object message
     * @throws IOException
     * @deprecated no longer used
     */
    @Deprecated
    public Object waitForMessage()
        throws IOException
    {
        try
        {
            return msgQueue.take();
        }
        catch (InterruptedException e)
        {
            throw new IOException("Interrupted waiting for message", e);
        }
    }

    /** Main processing method for the UDPDiscoveryReceiver object */
    @Override
    public void run()
    {
        try
        {
            log.debug( "Waiting for message." );

            while (!shutdown.get())
            {
                int activeKeys = selector.select();
                if (activeKeys == 0)
                {
                    continue;
                }

                for (Iterator<SelectionKey> i = selector.selectedKeys().iterator(); i.hasNext();)
                {
                    if (shutdown.get())
                    {
                        break;
                    }

                    SelectionKey key = i.next();

                    if (!key.isValid())
                    {
                        continue;
                    }

                    if (key.isReadable())
                    {
                        cnt.incrementAndGet();
                        log.debug( "{0} messages received.", this::getCnt );

                        DatagramChannel mc = (DatagramChannel) key.channel();

                        ByteBuffer byteBuffer = ByteBuffer.allocate(65536);
                        InetSocketAddress sourceAddress =
                                (InetSocketAddress) mc.receive(byteBuffer);
                        byteBuffer.flip();

                        try
                        {
                            log.debug("Received packet from address [{0}]", sourceAddress);
                            byte[] bytes = new byte[byteBuffer.limit()];
                            byteBuffer.get(bytes);
                            Object obj = serializer.deSerialize(bytes, null);

                            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
                                final UDPDiscoveryMessage msg = (UDPDiscoveryMessage) obj;
                                msg.setHost(sourceAddress.getHostString());

                                log.debug( "Read object from address [{0}], object=[{1}]",
                                        sourceAddress, obj );

                                // Just to keep the functionality of the deprecated waitForMessage method
                                synchronized (msgQueue)
                                {
                                    // Check if queue full already?
                                    if (msgQueue.remainingCapacity() == 0)
                                    {
                                        // remove oldest element from queue
                                        msgQueue.remove();
                                    }

                                    msgQueue.add(msg);
                                }

                                pooledExecutor.execute(() -> handleMessage(msg));
                                log.debug( "Passed handler to executor." );
                            }
                        }
                        catch ( final IOException | ClassNotFoundException e )
                        {
                            log.error( "Error receiving multicast packet", e );
                        }

                        i.remove();
                    }
                }
            } // end while
        }
        catch ( final IOException e )
        {
            log.error( "Unexpected exception in UDP receiver.", e );
        }
    }

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

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

    /**
     * For testing
     *
     * @param serializer the serializer to set
     * @since 3.1
     */
    protected void setSerializer(IElementSerializer serializer)
    {
        this.serializer = serializer;
    }

    /**
     * Separate thread run when a command comes into the UDPDiscoveryReceiver.
     * @deprecated No longer used
     */
    @Deprecated
    public class MessageHandler
        implements Runnable
    {
        /** The message to handle. Passed in during construction. */
        private final UDPDiscoveryMessage message;

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

        /**
         * Process the message.
         */
        @Override
        public void run()
        {
            handleMessage(message);
        }
    }

    /**
     * Separate thread run when a command comes into the UDPDiscoveryReceiver.
     */
    private void handleMessage(UDPDiscoveryMessage message)
    {
        // consider comparing ports here instead.
        if ( message.getRequesterId() == CacheInfo.listenerId )
        {
            log.debug( "Ignoring message sent from self" );
        }
        else
        {
            log.debug( "Process message sent from another" );
            log.debug( "Message = {0}", message );

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

    /**
     * Process the incoming message.
     */
    private void processMessage(UDPDiscoveryMessage message)
    {
        final DiscoveredService discoveredService = new DiscoveredService(message);

        switch (message.getMessageType())
        {
            case REMOVE:
                log.debug( "Removing service from set {0}", discoveredService );
                service.removeDiscoveredService( discoveredService );
                break;
            case REQUEST:
                // if this is a request message, have the service handle it and
                // return
                log.debug( "Message is a Request Broadcast, will have the service handle it." );
                service.serviceRequestBroadcast();
                break;
            case PASSIVE:
            default:
                log.debug( "Adding or updating service to set {0}", discoveredService );
                service.addOrUpdateService( discoveredService );
                break;
        }
    }

    /** Shuts down the socket. */
    @Override
    public void shutdown()
    {
        if (shutdown.compareAndSet(false, true))
        {
            try
            {
                selector.close();
                multicastGroupKey.drop();
                multicastChannel.close();
            }
            catch ( final IOException e )
            {
                log.error( "Problem closing socket" );
            }
        }
    }
}