

package gov.va.med.cds.filter;


import static org.junit.Assert.assertEquals;
import gov.va.med.cds.junit.runners.AfterTestsOnce;
import gov.va.med.cds.junit.runners.BeforeTestsOnce;
import gov.va.med.cds.junit.runners.SuiteAwareSpringRunner;
import gov.va.med.cds.junit.runners.Suite;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.junit.runner.RunWith;


@RunWith( SuiteAwareSpringRunner.class )
public abstract class AbstractConcurrentFilterTest
    extends
        AbstractFilterManagerTest
{
    private Log logger = LogFactory.getLog( AbstractConcurrentFilterTest.class );
    protected static final int RUN_TIME_IN_MILLI_SECONDS = 5000; //5000 Sync with MAX_MILLIS_PER_THREAD
    protected static final int FIXED_THREAD_POOL_SIZE = 50; //50;
    protected static final int CALLABLE_COUNT = FIXED_THREAD_POOL_SIZE;
    protected static final int MAX_MILLIS_PER_THREAD = 10000; //10000;
    protected static final int MAX_MILLIS_PER_FULL_POOL_EXECUTION = MAX_MILLIS_PER_THREAD * CALLABLE_COUNT;
    protected static List<Callable<Boolean>> callables = new LinkedList<Callable<Boolean>>();
    private static final String APPLICATION_NAME = "CDS FILTER TEST";
    protected static final long MAX_MEMORY_USED_IN_RUN = 50000000;
    protected List<String> moreFiltersThanCache = null;
    protected List<String> sameFiltersAsCache = null;
    protected List<String> lessFiltersThanCache = null;
    protected Map<String, List<String>> filterMap = null;


    @BeforeTestsOnce
    @Suite( groups = { "smoketest" } )
    public void beforeAbstractConcurrentFilterTestClassSetUp( )
        throws Exception
    {
        for ( int i = 0; i < CALLABLE_COUNT; i++ )
        {
            callables.add( new TestRunner() );
        }
    }


    @AfterTestsOnce
    @Suite( groups = { "smoketest" } )
    public void afterAbstractConcurrentFilterTestClassTearDown( )
    {
        callables = null;
    }


    /**
     * Tests concurrent clinical operations and invokes a set of <callable> objects concurrently and then loops through
     * the set of <future> objects returned to ensure that all threads have completed - any exceptions would then be
     * captured and thrown at that point.
     * 
     * @throws Exception
     */
    @Test
    @Suite( groups = { "smoketest" } )
    public void testConcurrentOperation( )
        throws Exception
    {
        ThreadPoolExecutor executor = createExecutor();

        if ( logger.isDebugEnabled() )
        {
            logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "invoking all runners." ) );
        }

        long startMillis = System.currentTimeMillis();

        long startMemory = Runtime.getRuntime().freeMemory();

        List<Future<Boolean>> futures = executor.invokeAll( callables, MAX_MILLIS_PER_FULL_POOL_EXECUTION, TimeUnit.MILLISECONDS );

        long endMemory = Runtime.getRuntime().freeMemory();
        long usedMemory = startMemory - endMemory;
        long memoryUsedInRun = ( usedMemory * 100 ) / startMemory;

        long elapsedMillis = System.currentTimeMillis() - startMillis;

        if ( logger.isDebugEnabled() )
        {
            logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "finished invoking runners in "
                            + elapsedMillis + " millis." ) );
            logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "% memory used in run " + memoryUsedInRun
                            + "%." ) );
            logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "Total memory used in run " + usedMemory
                            + "." ) );
        }

        int failedRunCount = 0;
        Exception lastException = null;

        // Collect results. This will capture any exceptions thrown and ensure that all tasks complete.
        for ( Future<Boolean> future : futures )
        {
            try
            {
                future.get();
            }
            catch ( Exception e )
            {
                failedRunCount++ ;
                lastException = e;
                e.printStackTrace();
            }
        }

        if ( lastException != null )
        {
            throw new Exception( failedRunCount + " of " + CALLABLE_COUNT + " runs failed.", lastException );
        }

        // Make sure all the threads succeeded.
        assertEquals( "Expected " + CALLABLE_COUNT + " completed invocation, but only finished " + executedCount + " in " + elapsedMillis
                        + " millis.", executedCount, CALLABLE_COUNT );
    }


    protected ThreadPoolExecutor createExecutor( )
    {
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<Runnable>( CALLABLE_COUNT );

        ThreadPoolExecutor executor = new ThreadPoolExecutor( FIXED_THREAD_POOL_SIZE, FIXED_THREAD_POOL_SIZE, MAX_MILLIS_PER_FULL_POOL_EXECUTION,
                        TimeUnit.MILLISECONDS, workQueue );

        if ( logger.isDebugEnabled() )
        {
            logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "starting up all threads." ) );
        }
        executor.prestartAllCoreThreads();
        return executor;
    }


    protected String getFilterXml( String filterId, String vhimVersion, String domainEntryPoint )
    {
        StringBuffer xmlWorkBuffer = new StringBuffer();

        xmlWorkBuffer.setLength( 0 );

        xmlWorkBuffer.append( "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" );
        xmlWorkBuffer.append( "<filter:filter vhimVersion=\"" );
        xmlWorkBuffer.append( vhimVersion );
        xmlWorkBuffer.append( "\" xmlns:filter=\"Filter\" xmlns:xsi=\"http://www.w3.org/2001/XMLSchema-instance\">" );
        xmlWorkBuffer.append( "<filterId>" );
        xmlWorkBuffer.append( filterId ); // Filter Id
        xmlWorkBuffer.append( "</filterId>" );
        xmlWorkBuffer.append( "<patients>" );
        xmlWorkBuffer.append( "<NationalId>" );
        xmlWorkBuffer.append( "1234567890" );
        xmlWorkBuffer.append( "</NationalId>" );
        xmlWorkBuffer.append( "</patients>" );

        // Build each entry point independently
        xmlWorkBuffer.append( "<entryPointFilter queryName=\"" );
        xmlWorkBuffer.append( filterId + "Query" );
        xmlWorkBuffer.append( "\">" );
        xmlWorkBuffer.append( "<domainEntryPoint>" );
        xmlWorkBuffer.append( domainEntryPoint );
        xmlWorkBuffer.append( "</domainEntryPoint>" );

        xmlWorkBuffer.append( "</entryPointFilter>" );
        xmlWorkBuffer.append( "</filter:filter>" );

        return xmlWorkBuffer.toString();
    }


    protected abstract void performFilterManagerOperation( )
        throws Exception;

    public static int invocationCount = 0;
    public static int executedCount = 0;

    class TestRunner
        implements
            Callable<Boolean>
    {
        private int invocationId;


        public TestRunner( )
        {
            synchronized ( AbstractConcurrentFilterTest.this )
            {
                invocationId = ++invocationCount;
                if ( logger.isDebugEnabled() )
                {
                    logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "Constructing TestRunner ID="
                                    + invocationId ) );
                }
            }
        }


        public int getInvocationId( )
        {
            return invocationId;
        }

        private boolean alreadyCalled = false;


        public Boolean call( )
            throws Exception
        {
            if ( alreadyCalled )
            {
                throw new Exception( "Already Called" );
            }
            alreadyCalled = true;
            if ( logger.isDebugEnabled() )
            {
                logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "Executing - ID = " + invocationId ) );
            }

            performFilterManagerOperation();

            if ( logger.isDebugEnabled() )
            {
                logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "ID=" + invocationId + ". The time is "
                                + System.currentTimeMillis() ) );
            }
            synchronized ( AbstractConcurrentFilterTest.this )
            {
                ++executedCount;
            }

            if ( logger.isDebugEnabled() )
            {
                logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "Finished Read - ID = " + invocationId ) );
            }
            return new Boolean( true );
        }

    }


    @javax.annotation.Resource
    public void setMoreFiltersThanCache( List<String> moreFiltersThanCache )
    {
        this.moreFiltersThanCache = moreFiltersThanCache;
    }


    @javax.annotation.Resource
    public void setSameFiltersAsCache( List<String> sameFiltersAsCache )
    {
        this.sameFiltersAsCache = sameFiltersAsCache;
    }


    @javax.annotation.Resource
    public void setLessFiltersThanCache( List<String> lessFiltersThanCache )
    {
        this.lessFiltersThanCache = lessFiltersThanCache;
    }


    @javax.annotation.Resource
    public void setFilterMap( Map<String, List<String>> filterMap )
    {
        this.filterMap = filterMap;
    }

}
