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.util.Map;
import java.util.Set;

import org.apache.commons.jcs3.JCS;
import org.apache.commons.jcs3.auxiliary.lateral.LateralCacheAttributes;
import org.apache.commons.jcs3.auxiliary.lateral.LateralCommand;
import org.apache.commons.jcs3.auxiliary.lateral.LateralElementDescriptor;
import org.apache.commons.jcs3.engine.CacheElement;
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.control.CompositeCache;
import org.apache.commons.jcs3.engine.control.CompositeCacheManager;
import org.apache.commons.jcs3.engine.control.MockCompositeCacheManager;
import org.apache.commons.jcs3.engine.control.group.GroupAttrName;
import org.apache.commons.jcs3.engine.control.group.GroupId;
import org.apache.commons.jcs3.utils.serialization.EncryptingSerializer;
import org.apache.commons.jcs3.utils.serialization.StandardSerializer;
import org.apache.commons.jcs3.utils.timing.SleepUtil;

import junit.framework.TestCase;

/**
 * Basic unit tests for the sending and receiving portions of the lateral cache.
 */
public class TestTCPLateralUnitTest
    extends TestCase
{
    private final MockCompositeCacheManager cacheMgr = new MockCompositeCacheManager();

    /**
     * Test setup
     */
    @Override
    public void setUp()
    {
        JCS.setConfigFilename( "/TestTCPLateralCache.ccf" );
    }

    private <K,V> CompositeCache<K, V> createCache(int port)
    {
        final TCPLateralCacheAttributes lattr = new TCPLateralCacheAttributes();
        lattr.setTcpListenerPort(port);
        lattr.setTransmissionType(LateralCacheAttributes.Type.TCP);

        final CompositeCache<K, V> cache = cacheMgr.getCache( "test" );

        // get the listener started
        // give it our mock cache manager
        //LateralTCPListener listener = (LateralTCPListener)
        LateralTCPListener.getInstance( lattr, cacheMgr, new StandardSerializer());

        return cache;
    }

    private <K, V> LateralTCPService<K, V> createService(int listenerPort, int serverPort, long listenerId) throws IOException
    {
        final TCPLateralCacheAttributes lattr2 = new TCPLateralCacheAttributes();
        lattr2.setTcpListenerPort(listenerPort);
        lattr2.setTransmissionType(LateralCacheAttributes.Type.TCP);
        lattr2.setTcpServer("localhost:" + serverPort);

        final LateralTCPService<K, V> service = new LateralTCPService<>(lattr2,  new StandardSerializer());
        service.setListenerId(listenerId);

        return service;
    }

    /**
     * Make sure we can send a bunch to the listener. This would be better if we could plugin a Mock
     * CacheManger. The listener will instantiate it on its own. We have to configure one before
     * that.
     * <p>
     * @throws Exception
     */
    public void testSimpleSend()
        throws Exception
    {
    	simpleSend(new StandardSerializer(), 8111);
    }
    
    /**
     * @throws Exception
     */
    public void testSimpleEncriptedSend()
            throws Exception
    {
    	EncryptingSerializer serializer = new EncryptingSerializer();
    	serializer.setPreSharedKey("my_key");
    	simpleSend(serializer, 8112);
    }
    
    private void simpleSend(final IElementSerializer serializer, final int port ) throws IOException {
    	// SETUP
        // force initialization
        JCS.getInstance( "test" );

        final TCPLateralCacheAttributes lac = new TCPLateralCacheAttributes();
        lac.setTransmissionType(LateralCacheAttributes.Type.TCP);
        lac.setTcpServer( "localhost:" + port );
        lac.setTcpListenerPort( port );

        final ICompositeCacheManager cacheMgr = CompositeCacheManager.getInstance();


        // start the listener
        final LateralTCPListener<String, String> listener = LateralTCPListener.getInstance( lac, cacheMgr, serializer );

        // send to the listener
        final LateralTCPSender lur = new LateralTCPSender(lac,  serializer);

        // DO WORK
        final int numMes = 10;
        for ( int i = 0; i < numMes; i++ )
        {
            final String message = "adsfasasfasfasdasf";
            final CacheElement<String, String> ce = new CacheElement<>( "test", "test", message );
            final LateralElementDescriptor<String, String> led =
                    new LateralElementDescriptor<>(ce, LateralCommand.UPDATE, 1);
            lur.send( led );
        }

        SleepUtil.sleepAtLeast( numMes * 4 ); // this may need to be adjusted ...

        // VERIFY
        assertEquals( "Should have received " + numMes + " by now.", numMes, listener.getPutCnt() );
    }

    /**
     * @throws Exception
     */
    public void testReceive()
        throws Exception
    {
        // VERIFY
        createCache(1101);

        final LateralTCPService<String, String> service = createService(1102, 1101, 123456);

        // DO WORK
        final int cnt = 100;
        for ( int i = 0; i < cnt; i++ )
        {
            final ICacheElement<String, String> element = new CacheElement<>( "test", "key" + i, "value1" );
            service.update( element );
        }

        SleepUtil.sleepAtLeast( 1000 );

        // VERIFY
        assertEquals( "Didn't get the correct number", cnt, cacheMgr.getCache().getUpdateCount() );
    }

    /**
     * Send objects with the same key but different values.
     * <p>
     * @throws Exception
     */
    public void testSameKeyDifferentObject()
        throws Exception
    {
        // SETUP
        final CompositeCache<String, String> cache = createCache(1103);

        // setup a service to talk to the listener started above.
        final LateralTCPService<String, String> service = createService(1104, 1103, 123456);

        // DO WORK
        final ICacheElement<String, String> element = new CacheElement<>( "test", "key", "value1" );
        service.update( element );

        SleepUtil.sleepAtLeast( 300 );

        final ICacheElement<String, String> element2 = new CacheElement<>( "test", "key", "value2" );
        service.update( element2 );

        SleepUtil.sleepAtLeast( 1000 );

        // VERIFY
        final ICacheElement<String, String> cacheElement = cache.get( "key" );
        assertEquals( "Didn't get the correct object "+ cacheElement, element2.getVal(), cacheElement.getVal() );
    }

    /**
     * Send objects with the same key but different values.
     * <p>
     * @throws Exception
     */
    public void testSameKeyObjectDifferentValueObject()
        throws Exception
    {
        final CompositeCache<String, String> cache = createCache(1105);

        final LateralTCPService<String, String> service = createService(1106, 1105, 123456);

        // DO WORK
        final String key = "key";
        final ICacheElement<String, String> element = new CacheElement<>( "test", key, "value1" );
        service.update( element );

        SleepUtil.sleepAtLeast( 300 );

        final ICacheElement<String, String> element2 = new CacheElement<>( "test", key, "value2" );
        service.update( element2 );

        SleepUtil.sleepAtLeast( 1000 );

        // VERIFY
        final ICacheElement<String, String> cacheElement = cache.get( "key" );
        assertEquals( "Didn't get the correct object: " + cacheElement , element2.getVal(), cacheElement.getVal() );
    }

    /**
     * Create a listener. Add an element to the listeners cache. Setup a service. Try to get from
     * the service.
     * <p>
     * @throws Exception
     */
    public void testGet_SendAndReceived()
        throws Exception
    {
        // SETUP
        final CompositeCache<String, String> cache = createCache(1107);

        // add the item to the listeners cache
        final ICacheElement<String, String> element = new CacheElement<>( "test", "key", "value1" );
        cache.update( element );

        // setup a service to talk to the listener started above.
        final LateralTCPService<String, String> service = createService(1108, 1107, 123456);

        SleepUtil.sleepAtLeast( 300 );

        // DO WORK
        final ICacheElement<String, String> result = service.get( "test", "key" );

        // VERIFY
        assertNotNull( "Result should not be null.", result );
        assertEquals( "Didn't get the correct object", element.getVal(), result.getVal() );
    }

    /**
     * Create a listener. Add an element to the listeners cache. Setup a service. Try to get keys from
     * the service.
     * <p>
     * @throws Exception
     */
    public void testGetGroupKeys_SendAndReceived()  throws Exception
    {
        // SETUP
        final CompositeCache<GroupAttrName<String>, String> cache = createCache(1150);

        // add the item to the listeners cache
        final GroupAttrName<String> groupKey = new GroupAttrName<>(new GroupId("test", "group"), "key");
        final ICacheElement<GroupAttrName<String>, String> element =
            new CacheElement<>( "test", groupKey, "value1" );
        cache.update( element );

        // setup a service to talk to the listener started above.
        final LateralTCPService<GroupAttrName<String>, String>service = createService(1151, 1150, 123459);

        SleepUtil.sleepAtLeast( 500 );

        // DO WORK
        final Set<GroupAttrName<String>> result = service.getKeySet("test");

       // SleepUtil.sleepAtLeast( 5000000 );

        // VERIFY
        assertNotNull( "Result should not be null.", result );
        assertEquals( "Didn't get the correct object", "key", result.iterator().next().attrName );
    }

    /**
     * Create a listener. Add an element to the listeners cache. Setup a service. Try to get from
     * the service.
     * <p>
     * @throws Exception
     */
    public void testGetMatching_WithData()
        throws Exception
    {
        // SETUP
        final CompositeCache<String, Integer> cache = createCache(1108);

        final String keyprefix1 = "MyPrefix1";
        final int numToInsertPrefix1 = 10;
        // insert with prefix1
        for ( int i = 0; i < numToInsertPrefix1; i++ )
        {
            // add the item to the listeners cache
            final ICacheElement<String, Integer> element = new CacheElement<>( "test", keyprefix1 + String.valueOf( i ), Integer.valueOf( i ) );
            cache.update( element );
        }

        // setup a service to talk to the listener started above.
        final LateralTCPService<String, Integer> service = createService(1108, 1108, 123456);

        SleepUtil.sleepAtLeast( 300 );

        // DO WORK
        final Map<String, ICacheElement<String, Integer>> result = service.getMatching( "test", keyprefix1 + ".+" );

        // VERIFY
        assertNotNull( "Result should not be null.", result );
        assertEquals( "Wrong number returned 1:", numToInsertPrefix1, result.size() );
    }
}
