

package gov.va.med.cds.client.functional.concurrent;


import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

import gov.va.med.cds.client.functional.AbstractFunctionalTest;
import gov.va.med.cds.client.webservice.ClinicalDataServiceSynchronousInterface;
import gov.va.med.cds.exception.TestException;
import gov.va.med.cds.junit.runners.BeforeTestsOnce;
import gov.va.med.cds.junit.runners.Suite;
import gov.va.med.cds.testharness.xml.Assert;
import gov.va.med.cds.util.StreamUtil;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.springframework.core.io.Resource;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
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;


public abstract class AbstractConcurrentTest
    extends
        AbstractFunctionalTest
{
    protected static final String CLINICAL_DATA_SERVICE_BEAN_ID = "clinicalDataService";
    protected String expectedResults;
    protected String readFilter;// a read filter of some kind will be used by specific concurrent tests (write or read)
    private Log logger = LogFactory.getLog( AbstractConcurrentTest.class );
    protected static final int FIXED_THREAD_POOL_SIZE = 50;
    protected static final int CALLABLE_COUNT = FIXED_THREAD_POOL_SIZE;
    protected static final int MAX_MILLIS_PER_THREAD = 10000;
    protected static final int MAX_MILLIS_PER_FULL_POOL_EXECUTION = MAX_MILLIS_PER_THREAD * CALLABLE_COUNT;
    protected static List<Callable<Boolean>> callables = null;
    private static final String APPLICATION_NAME = "CDS FNTEST";


    @BeforeTestsOnce
    @Suite( groups = "brokentest" )
    public void beforeAbstractConcurrentTestClassSetUp( )
        throws Exception
    {
        callables = new LinkedList<Callable<Boolean>>();
        for ( int i = 0; i < CALLABLE_COUNT; i++ )
        {
            callables.add( new TestRunner() );
        }
    }


    /**
     * 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 = "brokentest" )
    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();

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

        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." ) );
        }

        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;
            }
        }

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

        // Make sure all the threads succeeded.
        //        Assert.assertEquals( executedCount, CALLABLE_COUNT, "Expected " + CALLABLE_COUNT + " completed invocation, but only finished "
        //                        + executedCount + " in " + elapsedMillis + " millis." );
        String message = "Expected " + CALLABLE_COUNT + " completed invocation, but only finished " + executedCount + " in " + elapsedMillis
                        + " millis.";
        assertEquals( message, 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 abstract String performClinicalOperation( )
        throws Exception;

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

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


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

            session = getClinicalDataService();

            if ( !session.isAlive() )
            {
                fail( "The clinical data service is not alive for thread: " + Thread.currentThread().getName() );
            }
        }


        public int getInvocationId( )
        {
            return invocationId;
        }

        private boolean alreadyCalled = false;


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

            String result = performClinicalOperation();

            try
            {
                Assert.assertXmlSimilar( result, expectedResults );
            }
            finally
            {
                if ( logger.isDebugEnabled() )
                {
                    logger.debug( gov.va.med.cds.util.LogMessageUtil.buildMessage( null, null, APPLICATION_NAME, "ID=" + invocationId
                                    + ". The time is " + System.currentTimeMillis() ) );
                }
                synchronized ( AbstractConcurrentTest.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 setReadFilterResource( Resource readFilterResource )
        throws IOException
    {
        this.readFilter = StreamUtil.resourceToString( readFilterResource );
    }


    @javax.annotation.Resource
    public void setExpectedResultsResource( Resource expectedResultsResource )
        throws IOException
    {
        this.expectedResults = StreamUtil.resourceToString( expectedResultsResource );
    }
}
