package org.apache.commons.jcs3.auxiliary.lateral.socket.tcp;

/*
 * 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.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.jcs3.auxiliary.lateral.LateralElementDescriptor;
import org.apache.commons.jcs3.auxiliary.lateral.behavior.ILateralCacheListener;
import org.apache.commons.jcs3.auxiliary.lateral.socket.tcp.behavior.ITCPLateralCacheAttributes;
import org.apache.commons.jcs3.engine.CacheInfo;
import org.apache.commons.jcs3.engine.behavior.ICacheElement;
import org.apache.commons.jcs3.engine.behavior.ICompositeCacheManager;
import org.apache.commons.jcs3.engine.behavior.IElementSerializer;
import org.apache.commons.jcs3.engine.behavior.IShutdownObserver;
import org.apache.commons.jcs3.engine.control.CompositeCache;
import org.apache.commons.jcs3.log.Log;
import org.apache.commons.jcs3.log.LogManager;
import org.apache.commons.jcs3.utils.serialization.StandardSerializer;

/**
 * Listens for connections from other TCP lateral caches and handles them. The initialization method
 * starts a listening thread, which creates a socket server. When messages are received they are
 * passed to a pooled executor which then calls the appropriate handle method.
 */
public class LateralTCPListener<K, V>
    implements ILateralCacheListener<K, V>, IShutdownObserver
{
    /** The logger */
    private static final Log log = LogManager.getLog( LateralTCPListener.class );

    /** How long the server will block on an accept(). 0 is infinite. */
    private static final int acceptTimeOut = 1000;

    /** The CacheHub this listener is associated with */
    private transient ICompositeCacheManager cacheManager;

    /** Map of available instances, keyed by port */
    private static final ConcurrentHashMap<String, ILateralCacheListener<?, ?>> instances =
        new ConcurrentHashMap<>();

    /** Configuration attributes */
    private ITCPLateralCacheAttributes tcpLateralCacheAttributes;

    /** The listener thread */
    private Thread listenerThread;

    /**
     * Serializer for reading and writing
     */
    private IElementSerializer serializer;

    /** put count */
    private int putCnt;

    /** remove count */
    private int removeCnt;

    /** get count */
    private int getCnt;

    /**
     * Use the vmid by default. This can be set for testing. If we ever need to run more than one
     * per vm, then we need a new technique.
     */
    private long listenerId = CacheInfo.listenerId;

    /** is this shut down? */
    private final AtomicBoolean shutdown = new AtomicBoolean();

    /** is this terminated? */
    private final AtomicBoolean terminated = new AtomicBoolean();

    /**
     * Gets the instance attribute of the LateralCacheTCPListener class.
     * <p>
     * @param ilca ITCPLateralCacheAttributes
     * @param cacheMgr
     * @return The instance value
     * @deprecated Specify serializer
     */
    @Deprecated
    @SuppressWarnings("unchecked") // Need to cast because of common map for all instances
    public static <K, V> LateralTCPListener<K, V>
        getInstance( final ITCPLateralCacheAttributes ilca, final ICompositeCacheManager cacheMgr)
    {
        return (LateralTCPListener<K, V>) instances.computeIfAbsent(
                String.valueOf( ilca.getTcpListenerPort() ),
                k -> {
                    final LateralTCPListener<K, V> newIns = new LateralTCPListener<>( ilca, new StandardSerializer() );

                    newIns.init();
                    newIns.setCacheManager( cacheMgr );

                    log.info("Created new listener {0}", ilca::getTcpListenerPort);

                    return newIns;
                });
    }

    /**
     * Gets the instance attribute of the LateralCacheTCPListener class.
     * <p>
     * @param ilca ITCPLateralCacheAttributes
     * @param cacheMgr
     * @param serializer the serializer to use when receiving
     * @return The instance value
     */
    @SuppressWarnings("unchecked") // Need to cast because of common map for all instances
    public static <K, V> LateralTCPListener<K, V>
        getInstance( final ITCPLateralCacheAttributes ilca, final ICompositeCacheManager cacheMgr, final IElementSerializer serializer )
    {
        return (LateralTCPListener<K, V>) instances.computeIfAbsent(
                String.valueOf( ilca.getTcpListenerPort() ),
                k -> {
                    final LateralTCPListener<K, V> newIns = new LateralTCPListener<>( ilca, serializer );

                    newIns.init();
                    newIns.setCacheManager( cacheMgr );

                    log.info("Created new listener {0}", ilca::getTcpListenerPort);

                    return newIns;
                });
    }

    /**
     * Only need one since it does work for all regions, just reference by multiple region names.
     * <p>
     * @param ilca ITCPLateralCacheAttributes
     * @deprecated Specify serializer
     */
    @Deprecated
    protected LateralTCPListener( final ITCPLateralCacheAttributes ilca )
    {
        this(ilca, new StandardSerializer());
    }

    /**
     * Only need one since it does work for all regions, just reference by multiple region names.
     * <p>
     * @param ilca ITCPLateralCacheAttributes
     * @param serializer the serializer to use when receiving
     */
    protected LateralTCPListener( final ITCPLateralCacheAttributes ilca, final IElementSerializer serializer )
    {
        this.setTcpLateralCacheAttributes( ilca );
        this.serializer = serializer;
    }

    /**
     * This starts the ListenerThread on the specified port.
     */
    @Override
    public synchronized void init()
    {
        try
        {
            final int port = getTcpLateralCacheAttributes().getTcpListenerPort();
            final String host = getTcpLateralCacheAttributes().getTcpListenerHost();

            terminated.set(false);
            shutdown.set(false);

            final ServerSocketChannel serverSocket = ServerSocketChannel.open();

            SocketAddress endPoint;

            if (host != null && !host.isEmpty())
            {
                log.info( "Listening on {0}:{1}", host, port );
                //Bind the SocketAddress with host and port
                endPoint = new InetSocketAddress(host, port);
            }
            else
            {
                log.info( "Listening on port {0}", port );
                endPoint = new InetSocketAddress(port);
            }

            serverSocket.bind(endPoint);
            serverSocket.configureBlocking(false);

            listenerThread = new Thread(() -> runListener(serverSocket),
                    "JCS-LateralTCPListener-" + host + ":" + port);
            listenerThread.setDaemon(true);
            listenerThread.start();
        }
        catch ( final IOException ex )
        {
            throw new IllegalStateException( ex );
        }
    }

    /**
     * Let the lateral cache set a listener_id. Since there is only one listener for all the
     * regions and every region gets registered? the id shouldn't be set if it isn't zero. If it is
     * we assume that it is a reconnect.
     * <p>
     * By default, the listener id is the vmid.
     * <p>
     * The service should set this value. This value will never be changed by a server we connect
     * to. It needs to be non static, for unit tests.
     * <p>
     * The service will use the value it sets in all send requests to the sender.
     * <p>
     * @param id The new listenerId value
     * @throws IOException
     */
    @Override
    public void setListenerId( final long id )
        throws IOException
    {
        this.listenerId = id;
        log.debug( "set listenerId = {0}", id );
    }

    /**
     * Gets the listenerId attribute of the LateralCacheTCPListener object
     * <p>
     * @return The listenerId value
     * @throws IOException
     */
    @Override
    public long getListenerId()
        throws IOException
    {
        return this.listenerId;
    }

    /**
     * Increments the put count. Gets the cache that was injected by the lateral factory. Calls put
     * on the cache.
     * <p>
     * @see org.apache.commons.jcs3.engine.behavior.ICacheListener#handlePut(org.apache.commons.jcs3.engine.behavior.ICacheElement)
     */
    @Override
    public void handlePut( final ICacheElement<K, V> element )
        throws IOException
    {
        putCnt++;
        if ( log.isInfoEnabled() && getPutCnt() % 100 == 0 )
        {
            log.info( "Put Count (port {0}) = {1}",
                    () -> getTcpLateralCacheAttributes().getTcpListenerPort(),
                    this::getPutCnt);
        }

        log.debug( "handlePut> cacheName={0}, key={1}",
                element::getCacheName, element::getKey);

        getCache( element.getCacheName() ).localUpdate( element );
    }

    /**
     * Increments the remove count. Gets the cache that was injected by the lateral factory. Calls
     * remove on the cache.
     * <p>
     * @see org.apache.commons.jcs3.engine.behavior.ICacheListener#handleRemove(String,
     *      Object)
     */
    @Override
    public void handleRemove( final String cacheName, final K key )
        throws IOException
    {
        removeCnt++;
        if ( log.isInfoEnabled() && getRemoveCnt() % 100 == 0 )
        {
            log.info( "Remove Count = {0}", this::getRemoveCnt);
        }

        log.debug( "handleRemove> cacheName={0}, key={1}", cacheName, key );

        getCache( cacheName ).localRemove( key );
    }

    /**
     * Gets the cache that was injected by the lateral factory. Calls removeAll on the cache.
     * <p>
     * @see org.apache.commons.jcs3.engine.behavior.ICacheListener#handleRemoveAll(String)
     */
    @Override
    public void handleRemoveAll( final String cacheName )
        throws IOException
    {
        log.debug( "handleRemoveAll> cacheName={0}", cacheName );

        getCache( cacheName ).localRemoveAll();
    }

    /**
     * Gets the cache that was injected by the lateral factory. Calls get on the cache.
     * <p>
     * @param cacheName
     * @param key
     * @return a ICacheElement
     * @throws IOException
     */
    public ICacheElement<K, V> handleGet( final String cacheName, final K key )
        throws IOException
    {
        getCnt++;
        if ( log.isInfoEnabled() && getGetCnt() % 100 == 0 )
        {
            log.info( "Get Count (port {0}) = {1}",
                    () -> getTcpLateralCacheAttributes().getTcpListenerPort(),
                    this::getGetCnt);
        }

        log.debug( "handleGet> cacheName={0}, key={1}", cacheName, key );

        return getCache( cacheName ).localGet( key );
    }

    /**
     * Gets the cache that was injected by the lateral factory. Calls get on the cache.
     * <p>
     * @param cacheName the name of the cache
     * @param pattern the matching pattern
     * @return Map
     * @throws IOException
     */
    public Map<K, ICacheElement<K, V>> handleGetMatching( final String cacheName, final String pattern )
        throws IOException
    {
        getCnt++;
        if ( log.isInfoEnabled() && getGetCnt() % 100 == 0 )
        {
            log.info( "GetMatching Count (port {0}) = {1}",
                    () -> getTcpLateralCacheAttributes().getTcpListenerPort(),
                    this::getGetCnt);
        }

        log.debug( "handleGetMatching> cacheName={0}, pattern={1}", cacheName, pattern );

        return getCache( cacheName ).localGetMatching( pattern );
    }

    /**
     * Gets the cache that was injected by the lateral factory. Calls getKeySet on the cache.
     * <p>
     * @param cacheName the name of the cache
     * @return a set of keys
     * @throws IOException
     */
    public Set<K> handleGetKeySet( final String cacheName ) throws IOException
    {
        return getCache( cacheName ).getKeySet(true);
    }

    /**
     * This marks this instance as terminated.
     * <p>
     * @see org.apache.commons.jcs3.engine.behavior.ICacheListener#handleDispose(String)
     */
    @Override
    public void handleDispose( final String cacheName )
        throws IOException
    {
        log.info( "handleDispose > cacheName={0} | Ignoring message. "
                + "Do not dispose from remote.", cacheName );

        // TODO handle active deregistration, rather than passive detection
        dispose();
    }

    @Override
    public synchronized void dispose()
    {
        if (terminated.compareAndSet(false, true))
        {
            notify();
            listenerThread.interrupt();
        }
    }

    /**
     * Gets the cacheManager attribute of the LateralCacheTCPListener object.
     * <p>
     * Normally this is set by the factory. If it wasn't set the listener defaults to the expected
     * singleton behavior of the cache manager.
     * <p>
     * @param name
     * @return CompositeCache
     */
    protected CompositeCache<K, V> getCache( final String name )
    {
        return getCacheManager().getCache( name );
    }

    /**
     * This is roughly the number of updates the lateral has received.
     * <p>
     * @return Returns the putCnt.
     */
    public int getPutCnt()
    {
        return putCnt;
    }

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

    /**
     * @return Returns the removeCnt.
     */
    public int getRemoveCnt()
    {
        return removeCnt;
    }

    /**
     * @param cacheMgr The cacheMgr to set.
     */
    @Override
    public void setCacheManager( final ICompositeCacheManager cacheMgr )
    {
        this.cacheManager = cacheMgr;
    }

    /**
     * @return Returns the cacheMgr.
     */
    @Override
    public ICompositeCacheManager getCacheManager()
    {
        return cacheManager;
    }

    /**
     * @param tcpLateralCacheAttributes The tcpLateralCacheAttributes to set.
     */
    public void setTcpLateralCacheAttributes( final ITCPLateralCacheAttributes tcpLateralCacheAttributes )
    {
        this.tcpLateralCacheAttributes = tcpLateralCacheAttributes;
    }

    /**
     * @return Returns the tcpLateralCacheAttributes.
     */
    public ITCPLateralCacheAttributes getTcpLateralCacheAttributes()
    {
        return tcpLateralCacheAttributes;
    }

    /**
     * Processes commands from the server socket. There should be one listener for each configured
     * TCP lateral.
     * @deprecated No longer used
     */
    @Deprecated
    public class ListenerThread
        extends Thread
    {
        /** The socket listener */
        private final ServerSocket serverSocket;

        /**
         * Constructor
         *
         * @param serverSocket
         */
        public ListenerThread(final ServerSocket serverSocket)
        {
            this.serverSocket = serverSocket;
        }

        /** Main processing method for the ListenerThread object */
        @Override
        public void run()
        {
            runListener(serverSocket.getChannel());
        }
    }

    /**
     * Processes commands from the server socket. There should be one listener for each configured
     * TCP lateral.
     */
    private void runListener(final ServerSocketChannel serverSocket)
    {
        try (Selector selector = Selector.open())
        {
            serverSocket.register(selector, SelectionKey.OP_ACCEPT);
            log.debug("Waiting for clients to connect");

            // Check to see if we've been asked to exit, and exit
            while (!terminated.get())
            {
                int activeKeys = selector.select(acceptTimeOut);
                if (activeKeys == 0)
                {
                    continue;
                }

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

                    SelectionKey key = i.next();

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

                    if (key.isAcceptable())
                    {
                        ServerSocketChannel server = (ServerSocketChannel) key.channel();
                        SocketChannel client = server.accept();
                        if (client == null)
                        {
                            //may happen in non-blocking mode
                            continue;
                        }

                        log.info("Connected to client at {0}", client.getRemoteAddress());

                        client.configureBlocking(false);
                        client.register(selector, SelectionKey.OP_READ);
                    }

                    if (key.isReadable())
                    {
                        handleClient(key);
                    }

                    i.remove();
                }
            }

            log.debug("Thread terminated, exiting gracefully");

            //close all registered channels
            selector.keys().forEach(key -> {
                try
                {
                    key.channel().close();
                }
                catch (IOException e)
                {
                    log.warn("Problem closing channel", e);
                }
            });
        }
        catch (final IOException e)
        {
            log.error( "Exception caught in TCP listener", e );
        }
        finally
        {
            try
            {
                serverSocket.close();
            }
            catch (IOException e)
            {
                log.error( "Exception closing TCP listener", e );
            }
        }
    }

    /**
     * A Separate thread that runs when a command comes into the LateralTCPReceiver.
     * @deprecated No longer used
     */
    @Deprecated
    public class ConnectionHandler
        implements Runnable
    {
        /** The socket connection, passed in via constructor */
        private final Socket socket;

        /**
         * Construct for a given socket
         * @param socket
         */
        public ConnectionHandler( final Socket socket )
        {
            this.socket = socket;
        }

        /**
         * Main processing method for the LateralTCPReceiverConnection object
         */
        @Override
        public void run()
        {
            try (InputStream is = socket.getInputStream())
            {
                while ( true )
                {
                    final LateralElementDescriptor<K, V> led =
                            serializer.deSerializeFrom(is, null);

                    if ( led == null )
                    {
                        log.debug( "LateralElementDescriptor is null" );
                        continue;
                    }
                    if ( led.getRequesterId() == getListenerId() )
                    {
                        log.debug( "from self" );
                    }
                    else
                    {
                        log.debug( "receiving LateralElementDescriptor from another led = {0}",
                                led );

                        Object obj = handleElement(led);
                        if (obj != null)
                        {
                            OutputStream os = socket.getOutputStream();
                            serializer.serializeTo(obj, os);
                            os.flush();
                        }
                    }
                }
            }
            catch (final IOException e)
            {
                log.info("Caught {0}, closing connection.", e.getClass().getSimpleName(), e);
            }
            catch (final ClassNotFoundException e)
            {
                log.error( "Deserialization failed reading from socket", e );
            }
        }
    }

    /**
     * A Separate thread that runs when a command comes into the LateralTCPReceiver.
     */
    private void handleClient(final SelectionKey key)
    {
        final SocketChannel socketChannel = (SocketChannel) key.channel();

        try
        {
            final LateralElementDescriptor<K, V> led =
                    serializer.deSerializeFrom(socketChannel, null);

            if ( led == null )
            {
                log.debug("LateralElementDescriptor is null");
                return;
            }

            if ( led.getRequesterId() == getListenerId() )
            {
                log.debug( "from self" );
            }
            else
            {
                log.debug( "receiving LateralElementDescriptor from another led = {0}",
                        led );

                Object obj = handleElement(led);
                if (obj != null)
                {
                    serializer.serializeTo(obj, socketChannel);
                }
            }
        }
        catch (final IOException e)
        {
            log.info("Caught {0}, closing connection.", e.getClass().getSimpleName(), e);
            try
            {
                socketChannel.close();
            }
            catch (IOException e1)
            {
                log.error("Error while closing connection", e );
            }
        }
        catch (final ClassNotFoundException e)
        {
            log.error( "Deserialization failed reading from socket", e );
        }
    }

    /**
     * This calls the appropriate method, based on the command sent in the Lateral element
     * descriptor.
     * <p>
     * @param led the lateral element
     * @return a possible response
     * @throws IOException
     */
    private Object handleElement(final LateralElementDescriptor<K, V> led) throws IOException
    {
        final String cacheName = led.getPayload().getCacheName();
        final K key = led.getPayload().getKey();
        Object obj = null;

        switch (led.getCommand())
        {
            case UPDATE:
                handlePut(led.getPayload());
                break;

            case REMOVE:
                // if a hash code was given and filtering is on
                // check to see if they are the same
                // if so, then don't remove, otherwise issue a remove
                if (led.getValHashCode() != -1 &&
                    getTcpLateralCacheAttributes().isFilterRemoveByHashCode())
                {
                    final ICacheElement<K, V> test = getCache( cacheName ).localGet( key );
                    if ( test != null )
                    {
                        if ( test.getVal().hashCode() == led.getValHashCode() )
                        {
                            log.debug( "Filtering detected identical hashCode [{0}], "
                                    + "not issuing a remove for led {1}",
                                    led.getValHashCode(), led );
                            return null;
                        }
                        log.debug( "Different hash codes, in cache [{0}] sent [{1}]",
                                test.getVal()::hashCode, led::getValHashCode );
                    }
                }
                handleRemove( cacheName, key );
                break;

            case REMOVEALL:
                handleRemoveAll( cacheName );
                break;

            case GET:
                obj = handleGet( cacheName, key );
                break;

            case GET_MATCHING:
                obj = handleGetMatching( cacheName, (String) key );
                break;

            case GET_KEYSET:
                obj = handleGetKeySet(cacheName);
                break;

            default: break;
        }

        return obj;
    }

    /**
     * Shuts down the receiver.
     */
    @Override
    public void shutdown()
    {
        if ( shutdown.compareAndSet(false, true) )
        {
            log.info( "Shutting down TCP Lateral receiver." );
            dispose();
        }
        else
        {
            log.debug( "Shutdown already called." );
        }
    }
}