/*
 * 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.
 */

package org.apache.uima.ee.test;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.jms.Connection;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;

import junit.framework.Assert;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQMessageConsumer;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.log4j.Logger;
import org.apache.uima.UIMAFramework;
import org.apache.uima.UIMA_IllegalStateException;
import org.apache.uima.aae.UimaClassFactory;
import org.apache.uima.aae.client.UimaASProcessStatus;
import org.apache.uima.aae.client.UimaAsBaseCallbackListener;
import org.apache.uima.aae.client.UimaAsynchronousEngine;
import org.apache.uima.aae.controller.Endpoint;
import org.apache.uima.aae.error.ServiceShutdownException;
import org.apache.uima.aae.error.UimaASPingTimeout;
import org.apache.uima.aae.error.UimaASProcessCasTimeout;
import org.apache.uima.aae.monitor.statistics.AnalysisEnginePerformanceMetrics;
import org.apache.uima.adapter.jms.JmsConstants;
import org.apache.uima.adapter.jms.activemq.JmsInputChannel;
import org.apache.uima.adapter.jms.activemq.JmsOutputChannel;
import org.apache.uima.adapter.jms.activemq.SpringContainerDeployer;
import org.apache.uima.adapter.jms.client.BaseUIMAAsynchronousEngine_impl;
import org.apache.uima.adapter.jms.message.JmsMessageContext;
import org.apache.uima.analysis_engine.AnalysisEngineDescription;
import org.apache.uima.cas.CAS;
import org.apache.uima.cas.TypeSystem;
import org.apache.uima.cas.impl.XmiCasDeserializer;
import org.apache.uima.collection.CollectionException;
import org.apache.uima.collection.CollectionReader;
import org.apache.uima.collection.CollectionReaderDescription;
import org.apache.uima.collection.EntityProcessStatus;
import org.apache.uima.ee.test.utils.BaseTestSupport;
import org.apache.uima.ee.test.utils.UimaASTestRunner;
import org.apache.uima.internal.util.XMLUtils;
import org.apache.uima.resource.ResourceInitializationException;
import org.apache.uima.resource.ResourceProcessException;
import org.apache.uima.resource.ResourceSpecifier;
import org.apache.uima.resource.metadata.ProcessingResourceMetaData;
import org.apache.uima.resource.metadata.TypePriorityList;
import org.apache.uima.resourceSpecifier.factory.DeploymentDescriptorFactory;
import org.apache.uima.resourceSpecifier.factory.ServiceContext;
import org.apache.uima.resourceSpecifier.factory.UimaASPrimitiveDeploymentDescriptor;
import org.apache.uima.resourceSpecifier.factory.impl.ServiceContextImpl;
import org.apache.uima.util.XMLInputSource;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.xml.sax.SAXException;
import org.xml.sax.SAXParseException;

@RunWith(UimaASTestRunner.class)
public class TestUimaASExtended extends BaseTestSupport {

  private CountDownLatch getMetaCountLatch = null;

  private static final int MaxGetMetaRetryCount = 2;

  private static final String primitiveServiceQueue1 = "NoOpAnnotatorQueue";

  private static final String PrimitiveDescriptor1 = "resources/descriptors/analysis_engine/NoOpAnnotator.xml";

  private int getMetaRequestCount = 0;

  public BaseTestSupport superRef = null;

  /**
   * bring up testing for compressed binary serialization
   */
  
/*
  public static void main(String[] args ) {
	  JUnitCore core= new JUnitCore();
	    core.addListener(new UimaASJunitTestFailFastListener());
	    core.run(TestUimaASExtended.class);
  }
  */
  
    private String getMasterConnectorURI(BrokerService b) {

	return b.getDefaultSocketURIString();
    }  
    /*
    @Test
    public void testBrokerRestartWithPrimitiveMultiplier() throws Exception {
      System.out.println("-------------- testBrokerRestartWithPrimitiveMultiplier -------------");
      System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());

      BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
      
      deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplier.xml");
     
      
      broker.stop();
      broker.waitUntilStopped();

      broker = createBroker();
      broker.start();
      broker.waitUntilStarted();

      String burl = broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString();
      Map<String, Object> appCtx = 
      buildContext(burl, "TestMultiplierQueue");

      // reduce the cas pool size and reply window
      appCtx.remove(UimaAsynchronousEngine.ShadowCasPoolSize);
      appCtx.put(UimaAsynchronousEngine.ShadowCasPoolSize, Integer.valueOf(2));
      runTest(appCtx, eeUimaEngine,burl,
              "TestMultiplierQueue", 1, PROCESS_LATCH);
      
      eeUimaEngine.stop();
    }
*/
    
    /**
     * This test starts a secondary broker, starts NoOp Annotator, and
     * using synchronous sendAndReceive() sends 5 CASes for analysis. Before sending 6th, the test
     * stops the secondary broker and sends 5 more CASes. All CASes sent after
     * the broker shutdown result in GetMeta ping and a subsequent timeout. Before
     * sending 11th CAS the test starts the broker again and sends 10 more CASes 
     * @throws Exception
     */
    @Test
    public void testSyncClientRecoveryFromBrokerStopAndRestart() throws Exception  {
      System.out.println("-------------- testSyncClientRecoveryFromBrokerStopAndRestart -------------");
      //System.setProperty("uima.as.enable.jmx","false");  
      // Instantiate Uima AS Client
        BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
        BrokerService broker2 = setupSecondaryBroker(true);
        // Deploy Uima AS Primitive Service
        deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
        
        Map<String, Object> appCtx = 
        buildContext(broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");
        appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
        appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
        appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 20000);
        initialize(uimaAsEngine, appCtx);
        waitUntilInitialized();
        int errorCount=0;
        int delay = 1;
        for (int i = 0; i < 20; i++) {
          
          if ( i == 5 ) {
            broker2.stop();
            broker2.waitUntilStopped();
        	  System.out.println("..... Stopped broker ............................");
        	  Timer timer = new Timer();
  		  timer.schedule(new StartBrokerTask(broker2, this),500);
  		  delay = 200;
          } 
          
          synchronized(appCtx) {
          	try {
          		appCtx.wait(delay);
          	} catch( InterruptedException eee) {
          		
          	}
          }
          /*
          else if ( i == 10 ) {
            //  restart the broker 
            System.setProperty("activemq.broker.jmx.domain","org.apache.activemq.test");
            broker2 = setupSecondaryBroker(true);
            
            broker2.start();
            broker2.waitUntilStarted();

          }
          */
          CAS cas = uimaAsEngine.getCAS();
          cas.setDocumentText("Some Text");
        //  System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
          try {
            uimaAsEngine.sendAndReceiveCAS(cas);
          } catch( Exception e) {
            errorCount++;
            System.out.println("Client Received Expected Error on CAS:"+(i+1));
          } finally {
            cas.release();
          }
        }
        
        uimaAsEngine.stop();
        super.cleanBroker(broker2);

        broker2.stop();

        //  expecting 5 failures due to broker missing
//        if ( errorCount != 5 ) {
//          fail("Expected 5 failures due to broker down, instead received:"+errorCount+" failures");
//        }
        broker2.waitUntilStopped();

    }
    /**
     * This test creates 4 UIMA AS clients and runs each in a separate thread. There is a single 
     * shared jms connection to a broker that each client uses. After initialization a client
     * sends 1000 CASes to a remote service. While clients are processing the test kills 
     * the broker, waits for 4 seconds and restarts it. While the broker is down, clients
     * keep trying sending CASes, receiving Ping timeouts. Once the broker is available again
     * all clients should recover and begin processing CASes again. This tests recovery of a 
     * shared connection.
     * 
     * @throws Exception
     */
    @Test
    public void testMultipleSyncClientsRecoveryFromBrokerStopAndRestart() throws Exception  {
      System.out.println("-------------- testMultipleSyncClientsRecoveryFromBrokerStopAndRestart -------------");
      //System.setProperty("uima.as.enable.jmx","false"); 
      final BrokerService broker2 = setupSecondaryBroker(true);
      final CountDownLatch latch = new CountDownLatch(8);
      Thread[] clientThreads = new Thread[8];
      
      //  Create 4 Uima AS clients each running in a separate thread
      for(int i=0; i < 8; i++) {
        clientThreads[i] = new Thread() {
          public void run() {
            BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
            try {
              // Deploy Uima AS Primitive Service
              deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
              Map<String, Object> appCtx = 
                buildContext(broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");

              appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
              appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
              initialize(uimaAsEngine, appCtx);
              waitUntilInitialized();
              for (int i = 0; i < 500; i++) {
                if ( i == 5 ) {
                  latch.countDown();   // indicate that some CASes were processed
                }
                CAS cas = uimaAsEngine.getCAS();
                cas.setDocumentText("Some Text");
//                System.out.println("UIMA AS Client#"+ Thread.currentThread().getId()+" Sending CAS#"+(i + 1) + " Request to a Service");
                try {
                  uimaAsEngine.sendAndReceiveCAS(cas);
                } catch( Exception e) {
                  System.out.println("Client Received Expected Error on CAS:"+(i+1));
                } finally {
                  cas.release();
                }
                synchronized(uimaAsEngine) {
              	  uimaAsEngine.wait(100);
                }
              }
              System.out.println("Thread:"+Thread.currentThread().getId()+" Completed run()");
              uimaAsEngine.stop();
            } catch( Exception e) {
              e.printStackTrace();
              return;
            }
          }
        };
        clientThreads[i].start();
      }
      BrokerService broker3 =  null;
      try {
        latch.await();  // wait for all threads to process a few CASes

        broker2.stop();
        System.out.println("Stopping Broker - wait ...");
        broker2.waitUntilStopped();

        System.out.println("Restarting Broker - wait ...");
        //  restart the broker 
        broker3 = setupSecondaryBroker(true);
        broker3.waitUntilStarted();

      } catch ( Exception e ) {
        
      } finally {
        for(int i=0; i < 4; i++ ) {
          clientThreads[i].join();
        }
        System.out.println("Stopping Broker - wait ...");
        if ( broker3 != null ) {
          super.cleanBroker(broker3);

          broker3.stop();
          broker3.waitUntilStopped();

        }
      }
  }
    @Test
    public void testClient() throws Exception {
      System.out.println("-------------- testClient -------------");
      System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());

      BaseUIMAAsynchronousEngine_impl uimaAsEngine1 = new BaseUIMAAsynchronousEngine_impl();
      BaseUIMAAsynchronousEngine_impl uimaAsEngine2 = new BaseUIMAAsynchronousEngine_impl();
      File directory = new File("./");
      System.out.println(directory.getAbsolutePath());
      String sid1= deployService(uimaAsEngine1, relativePath + "/Deploy_AggregateMultiplierWith30SecDelay.xml");
      String sid2 = deployService(uimaAsEngine2, relativePath + "/Deploy_AggregateMultiplierWith30SecDelay.xml");
      
      uimaAsEngine1.undeploy(sid1);
      
      uimaAsEngine2.undeploy(sid2);
    }
    
    @Test
    public void testClientWithPrimitives() throws Exception {
      System.out.println("-------------- testClientWithPrimitives -------------");
      System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());

      BaseUIMAAsynchronousEngine_impl uimaAsEngine1 = new BaseUIMAAsynchronousEngine_impl();
      BaseUIMAAsynchronousEngine_impl uimaAsEngine2 = new BaseUIMAAsynchronousEngine_impl();

      String sid1= deployService(uimaAsEngine1, relativePath + "/Deploy_NoOpAnnotator.xml");
      String sid2 = deployService(uimaAsEngine2, relativePath + "/Deploy_NoOpAnnotator.xml");
      
      uimaAsEngine1.undeploy(sid1);
      
      uimaAsEngine2.undeploy(sid2);
      
    }
    
    /**
     * Tests error handling of the client. It deploys Aggregate service Cas Multiplier. initializes
     * the client and sends a CAS for processing. The child CAS is than held in NoOp Annotator for
     * 30 secs to simulate heavy processing. While the CAS is being processed, a broker is stopped.
     * The client should timeout after 40 secs and attempt to send 2 more CASes. Since the broker
     * is down, each of these 2 CASes goes into a retry list while a Connection is being retried.
     * Both should timeout, and sendAndReceive() should fail due to a timeout.
     *
     * @throws Exception
     */
    /*
    @Test
    public void testClientRecoveryFromBrokerFailure() throws Exception {
      System.out.println("-------------- testClientRecoveryFromBrokerFailure -------------");
      System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());

      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      deployService(uimaAsEngine, relativePath + "/Deploy_AggregateMultiplierWith30SecDelay.xml");

      Map<String, Object> appCtx = 
         buildContext(broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString(), "TopLevelTaeQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 40000);   // AE will hold the CAS for 30 secs so this needs to be larger
      appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 20000);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();

      ScheduledExecutorService s = Executors.newSingleThreadScheduledExecutor();

      // schedule a thread that will stop the broker after 10 secs
      s.schedule(
    		  new Runnable() {

				@Override
				public void run() {
					try {
						System.out.println("Stopping Broker ...");
						broker.stop();
 				        broker.waitUntilStopped();
						System.out.println("Broker Stopped...");

					} catch( Exception e) {
						
					}
					
				}
    			  
    		  }
    		  , 10, TimeUnit.SECONDS);
      int timeoutCount=0;

      // try to send 3 CASes, each should timeout
      for (int i = 0; i < 3; i++) {
          CAS cas = uimaAsEngine.getCAS();
          cas.setDocumentText("Some Text");
          try {
        	  System.out.println("............... Client Sending CAS #"+(i+1));
              uimaAsEngine.sendAndReceiveCAS(cas);
            } catch( Exception e) {
            	if ( e instanceof UimaASProcessCasTimeout ) {
            		timeoutCount++;
            		System.out.println("Client .............. "+e.getMessage());
            		if ( e.getCause() != null && e instanceof UimaASPingTimeout) {
                		System.out.println("Client .............. "+e.getCause().getMessage());
            		}
            	} else if ( e.getCause() instanceof UimaASProcessCasTimeout ) {
            		timeoutCount++;
            		System.out.println("Client .............. "+e.getCause().getMessage());
            		if ( e.getCause().getCause() != null && e.getCause().getCause() instanceof UimaASPingTimeout) {
                		System.out.println("Client .............. "+e.getCause().getCause().getMessage());
            		}
            	} else {
                	e.printStackTrace();
            	}
            	//              System.out.println("Client Received Expected Error on CAS:"+(i+1));
            } finally {
              cas.release();
            }
      }
       if ( timeoutCount != 3) {
          uimaAsEngine.stop();
    	  fail("Expected 3 Errors Due to Timeout, Instead Got "+timeoutCount+" Timeouts");
      } else {
  		System.out.println("Client .............. Stopping");
          uimaAsEngine.stop();
      }

 }
    
    @Test
    public void testBrokerRestartWithAggregateMultiplier() throws Exception {
      System.out.println("-------------- testBrokerRestartWithAggregateMultiplier -------------");
      System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());

      BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
      deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplier.xml");
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      deployService(eeUimaEngine, relativePath + "/Deploy_AggregateMultiplier.xml");

      String burl = broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString();
      Map<String, Object> appCtx = buildContext(burl, "TopLevelTaeQueue");
      synchronized(this) {
    	  this.wait(2000);
      }
      broker.stop();
      broker.waitUntilStopped();

      synchronized(this) {
    	  this.wait(2000);
      }

      broker = createBroker();
      broker.start();
      broker.waitUntilStarted();
      synchronized(this) {
    	  this.wait(2000);
      }


      // reduce the cas pool size and reply window
      appCtx.remove(UimaAsynchronousEngine.ShadowCasPoolSize);
      appCtx.put(UimaAsynchronousEngine.ShadowCasPoolSize, Integer.valueOf(2));
      runTest(appCtx, eeUimaEngine,burl,
              "TopLevelTaeQueue", 1, PROCESS_LATCH);
  //    eeUimaEngine.stop();
 }
    */
    /**
     * Tests client and service recovery from broker restart. It deploys CM service, dispatches
     * a CAS for processing and while the CAS is in process, it bounces a broker. The service
     * listeners should be restored and the CAS should fail due to invalid destination. Once
     * the client times out, it should send 2 more CASes which should force client to re-establish
     * connection with a broker and replies should come back.
     * 
     * @throws Exception
     */

    @Test
    public void testBrokerRestartWithAggregateMultiplierWhileProcessingCAS() throws Exception {
      System.out.println("-------------- testBrokerRestartWithAggregateMultiplierWhileProcessingCAS -------------");
      System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());

      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      deployService(uimaAsEngine, relativePath + "/Deploy_AggregateMultiplierWith30SecDelay.xml");

      Map<String, Object> appCtx = 
         buildContext(broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString(), "TopLevelTaeQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 40000);
      appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 20000);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();


      ScheduledExecutorService s = Executors.newSingleThreadScheduledExecutor();

      s.schedule(
    		  new Runnable() {

				@Override
				public void run() {
					try {
						System.out.println("Stopping Broker ...");
						broker.stop();
 				        broker.waitUntilStopped();
						System.out.println("Broker Stopped...");

					    broker = createBroker();
					    broker.start();
					    broker.waitUntilStarted();
					    System.out.println("Broker Restarted...");

					
					} catch( Exception e) {
						
					}
					
				}
    			  
    		  }
    		  , 10, TimeUnit.SECONDS);


      for (int i = 0; i < 3; i++) {
          CAS cas = uimaAsEngine.getCAS();
          cas.setDocumentText("Some Text");
          try {
        	  System.out.println("............... Client Sending CAS #"+(i+1));
              uimaAsEngine.sendAndReceiveCAS(cas);
            } catch( Exception e) {
            	e.printStackTrace();
            	//              System.out.println("Client Received Expected Error on CAS:"+(i+1));
            } finally {
              cas.release();
            }


      }
      s.shutdownNow();
      s.awaitTermination(10, TimeUnit.HOURS);
      uimaAsEngine.stop();
 }

    
    @Test
    public void testBrokerRestartWhileProcessingCAS() throws Exception {
      System.out.println("--------------  testBrokerRestartWhileProcessingCAS -------------");
      System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());

      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      
      deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWith30SecDelay.xml");
      Map<String, Object> appCtx = 
      buildContext(broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString(), "NoOpAnnotatorQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 40000);
      appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 20000);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();


      ScheduledExecutorService s = Executors.newSingleThreadScheduledExecutor();
      
      s.schedule(
    		  new Runnable() {

				@Override
				public void run() {
					try {
						System.out.println("Stopping Broker ...");
						broker.stop();
 				        broker.waitUntilStopped();
						System.out.println("Broker Stopped...");

					    broker = createBroker();
					    broker.start();
					    broker.waitUntilStarted();
					    System.out.println("Broker Restarted...");

					
					} catch( Exception e) {
						
					}
					
				}
    			  
    		  }
    		  , 10, TimeUnit.SECONDS);

      for (int i = 0; i < 1; i++) {
          CAS cas = uimaAsEngine.getCAS();
          cas.setDocumentText("Some Text");
          try {
              uimaAsEngine.sendAndReceiveCAS(cas);
            } catch( Exception e) {
            	e.printStackTrace();
            	//              System.out.println("Client Received Expected Error on CAS:"+(i+1));
            } finally {
              cas.release();
            }


      }
      s.shutdownNow();
      s.awaitTermination(10, TimeUnit.HOURS);
      uimaAsEngine.stop();
      
      
    }
    /**
     * This tests if broker keep-alive protocol is working. With AMQ 5.13.2 the test
     * fails due to broker bug. What happens is that when a jms client uses http
     * protocol, the connection is made but the keep-alive chat between broker and
     * client is not causing a timeout and an exception. 
     * 
     * The exception is internal to the broker but it also happens within amq
     * client code. To get to this, a custom spring based listener is deployed 
     * with some of its exception handling methods overriden to capture an exception. 
     *  
     * @throws Exception
     */
    
    @Test
    public void testServiceWithHttpListeners() throws Exception {
  	    System.out.println("-------------- testServiceWithHttpListeners -------------");
  	    // Need java monitor object on which to sleep
  	    Object waitObject = new Object();
  	    // Custom spring listener with handleListenerSetupFailure() overriden to 
  	    // capture AMQ exception.
  	    TestDefaultMessageListenerContainer c = new TestDefaultMessageListenerContainer();
  	    c.setConnectionFactory(new ActiveMQConnectionFactory("http://localhost:18888"));
  	    c.setDestinationName("TestQ");
  	    c.setConcurrentConsumers(2);
  	    c.setBeanName("testServiceWithHttpListeners() - JUnit Test Listener");
  	    c.setMessageListener(new JmsInputChannel());
  	    //c.initialize();
  	    //c.afterPropertiesSet();
  	    c.start();
  	    
  	    if ( c.isRunning() ) {
  		    System.out.println("... Listener Ready");
  	    	
  	    }
  	    // Keep-alive has a default 30 secs timeout. Sleep for bit longer than that
  	    // If there is an exception due to keep-alive, an exception handler will be
  	    // called on the TestDefaultMessageListenerContainer instance where we 
  	    // capture the error.
  	    System.out.println("... Waiting for 40 secs");
  	    try {
  	    	synchronized(waitObject) {
  	    		waitObject.wait(40000);
  	    	}
  	    	// had there been broker issues relateds to keep-alive the listener's failed
  	    	// flag would have been set by now. Check it and fail the test 
  	    	if ( c.failed() ) {
  		    	fail("Broker Failed - Reason:"+c.getReasonForFailure());
  	    	} else {
 	    	}
  	    } catch( Exception e) {
  	    	e.printStackTrace();
  	    	fail(e.getMessage());
  	    }
  		System.out.println("Stopping Listener");
  		c.stop();
  		c.shutdown();

    }

    

 
    
  /*
  public void testContinueOnRetryFailure2() throws Exception {
	    System.out.println("-------------- testContinueOnRetryFailure -------------");
	    File tempDir = new File("target/temp");
	    deleteAllFiles(tempDir);
	    try {
	        tempDir.mkdir();
	    } catch( Exception e) {
	    	e.printStackTrace();
	    	throw e;
	    }
	    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
	    deployService(eeUimaEngine, relativePath + "/Deploy_WriterAnnotatorA.xml");
	    deployService(eeUimaEngine, relativePath + "/Deploy_WriterAnnotatorB.xml");
	    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithContinueOnRetryFailures.xml");
	    runTest(null, eeUimaEngine, String.valueOf(broker.getMasterConnectorURI()), "TopLevelTaeQueue",
	            1, EXCEPTION_LATCH);
	    if (!(new File(tempDir, "WriterAnnotatorB.3")).exists()
	            || (new File(tempDir, "WriterAnnotatorB.4")).exists()) {
	      fail("Second annotator should have run 3 times");
	    }
	    if ((new File(tempDir, "WriterAnnotatorC.1")).exists()) {
	      fail("Third annotator should not have seen CAS");
	    }
	  }
*/
  
    
    
    /*
     * Tests 
     */
//    @Test
//    public void testSyncClientRecoveryFromBrokerStopAndRestart3() throws Exception  {
//      System.out.println("-------------- testSyncClientRecoveryFromBrokerStopAndRestart -------------");
//      System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());
// 
//      // Instantiate Uima AS Client
//        BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
//        //BrokerService broker2 = setupSecondaryBroker(true);
//        // Deploy Uima AS Primitive Service
//        deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
//        
//        Map<String, Object> appCtx = 
//        buildContext(broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString(), "NoOpAnnotatorQueue");
//        appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
//        appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
//        appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 20000);
//        initialize(uimaAsEngine, appCtx);
//        waitUntilInitialized();
//
//
//        broker.stop();
//        broker.waitUntilStopped();
//
//        //System.setProperty("activemq.broker.jmx.domain","org.apache.activemq.test");
//        //broker2 = setupSecondaryBroker(true);
//        broker = createBroker();
//        broker.start();
//        broker.waitUntilStarted();
//        int errorCount = 0;
//        System.out.println("Sending CASes");
//        for (int i = 0; i < 60; i++) {
//            CAS cas = uimaAsEngine.getCAS();
//            cas.setDocumentText("Some Text");
//            try {
//                uimaAsEngine.sendAndReceiveCAS(cas);
//              } catch( Exception e) {
//                System.out.println("Client Received Expected Error on CAS:"+(i+1));
//              } finally {
//                cas.release();
//              }
//
//
//        }
//        uimaAsEngine.stop();
//        
//        /*
//        int errorCount=0;
//        for (int i = 0; i < 20; i++) {
//          
//          if ( i == 5 ) {
//            broker2.stop();
//            broker2.waitUntilStopped();
//          } else if ( i == 10 ) {
//            //  restart the broker 
//            System.setProperty("activemq.broker.jmx.domain","org.apache.activemq.test");
//            broker2 = setupSecondaryBroker(true);
//            
//            broker2.start();
//            broker2.waitUntilStarted();
//
//          }
//          CAS cas = uimaAsEngine.getCAS();
//          cas.setDocumentText("Some Text");
//        //  System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
//          try {
//            uimaAsEngine.sendAndReceiveCAS(cas);
//          } catch( Exception e) {
//            errorCount++;
//            System.out.println("Client Received Expected Error on CAS:"+(i+1));
//          } finally {
//            cas.release();
//          }
//        }
//        
//        uimaAsEngine.stop();
//        super.cleanBroker(broker2);
//
//        broker2.stop();
//
//        //  expecting 5 failures due to broker missing
//        if ( errorCount != 5 ) {
//          fail("Expected 5 failures due to broker down, instead received:"+errorCount+" failures");
//        }
//        broker2.waitUntilStopped();
//*/
//    }

    /*
    
    @Test
    public void testSyncClientRecoveryFromBrokerStopAndRestart2() throws Exception  {
        broker.stop();
        broker.waitUntilStopped();
    	System.out.println("-------------- testSyncClientRecoveryFromBrokerStopAndRestart -------------");
         // Instantiate Uima AS Client
          BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
          BrokerService broker2 = setupSecondaryBroker(true);
          // Deploy Uima AS Primitive Service
          deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
          
          Map<String, Object> appCtx = 
          buildContext(broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");
          appCtx.put(UimaAsynchronousEngine.Timeout, 5100);
          appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
          appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 20000);
          initialize(uimaAsEngine, appCtx);
          waitUntilInitialized();
          
          // Get meta received, bounce the broker now. 
          broker2.stop();
          broker2.waitUntilStopped();

          
//          ActiveMQConnectionFactory f = new ActiveMQConnectionFactory("");
//          ActiveMQConnection c = (ActiveMQConnection)f.createConnection();
          //  restart the broker 
          System.setProperty("activemq.broker.jmx.domain","org.apache.activemq.test");
          broker2 = setupSecondaryBroker(true);
          
          broker2.start();
          broker2.waitUntilStarted();
        
          // new broker is up. Send a few CASes now
          for( int i=0; i < 5; i++) {
              CAS cas = uimaAsEngine.getCAS();
              cas.setDocumentText("Some Text");
              try {
                uimaAsEngine.sendAndReceiveCAS(cas);
              } catch( Exception e) {
            	  e.printStackTrace();
            	 // fail("Unexpected exception from sendAndReceive()- test Failed");
              } finally {
                cas.release();
              }
        	  
          }

          
          
          uimaAsEngine.stop();
          super.cleanBroker(broker2);

          broker2.stop();
          broker2.waitUntilStopped();

      }

 */
  
  /**
   * Test use of a JMS Service Adapter. Invoke from a synchronous aggregate to emulate usage from
   * RunAE or RunCPE.
   * 
   * @throws Exception
   */
    @Test
    public void testJmsServiceAdapter() throws Exception {
	  Logger.getLogger(this.getClass()).info("-------------- testJmsServiceAdapter -------------");
	  //setUp();
	  BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    try {
        deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
        deployService(eeUimaEngine, relativePath + "/Deploy_SyncAggregateWithJmsService.xml");
        runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
                1, PROCESS_LATCH);
       
    } catch( Exception e ) {
    	throw e;
    }
  }
    /**
     * Test use of a JMS Service Adapter. Invoke from a synchronous aggregate to emulate usage from
     * RunAE or RunCPE.
     * 
     * @throws Exception
     */
      @Test
      public void testJmsServiceAdapterInAsyncAggregate() throws Exception {
  	  Logger.getLogger(this.getClass()).info("-------------- testJmsServiceAdapter -------------");
  	  //setUp();
  	  BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
      try {
          deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
          deployService(eeUimaEngine, relativePath + "/Deploy_AsyncAggregateWithJmsService.xml");
          runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
                  1, PROCESS_LATCH);
         
      } catch( Exception e ) {
      	throw e;
      }
    }
  /*
   * Tests Uima AS client placeholder handling and substitution. The Uima Aggregate instantiates
   * UIMA AS client proxy using Jms Client Descriptor that contains a placeholder
   * ${defaultBrokerURL} instead of hard coded Broker URL. The client parses the 
   * placeholder string, retrieves the name (defaultBrokerURL) and uses it to look
   * up tha actual broker URL in System properties.
   */

    @Test
    public void testJmsServiceAdapterWithPlaceholder() throws Exception {
    System.out.println("-------------- testJmsServiceAdapterWithPlaceholder -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_SyncAggregateWithJmsServiceUsingPlaceholder.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            10, PROCESS_LATCH);
  }

  /**
   * Tests use of a JMS Service Adapter and an override of the MultipleDeploymentAllowed. 
   * In this test, the AE descriptor of the remote service is configured with MultipleDeploymentAllowed=false
   * Without the override this causes an exception when instantiating Uima aggregate with
   * MultipleDeploymentAllowed=true. 
   * 
   * @throws Exception
   */
  @Test
  public void testJmsServiceAdapterWithOverride() throws Exception {
    System.out.println("-------------- testJmsServiceAdapterWithOverride -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_SingleInstancePersonTitleAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_SyncAggregateWithJmsServiceAndScaleoutOverride.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            10, PROCESS_LATCH);
  }

  @Test
  public void testJmsServiceAdapterWithException() throws Exception {
    System.out.println("-------------- testJmsServiceAdapterWithException -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithException.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_SyncAggregateWithJmsService.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 0);
    initialize(eeUimaEngine, appCtx);
    waitUntilInitialized();
    CAS cas = eeUimaEngine.getCAS();
    try {
      eeUimaEngine.sendAndReceiveCAS(cas);
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      eeUimaEngine.collectionProcessingComplete();
      cas.reset();
    }
    eeUimaEngine.stop();
  }

  @Test
  public void testJmsServiceAdapterWithProcessTimeout() throws Exception {
    System.out.println("-------------- testJmsServiceAdapterWithProcessTimeout -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithLongDelay.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_SyncAggregateWithJmsServiceLongDelay.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);

  }

  @Test
  public void testJmsServiceAdapterWithGetmetaTimeout() throws Exception {
    System.out.println("-------------- testJmsServiceAdapterWithGetmetaTimeout -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    try {
      deployService(eeUimaEngine, relativePath + "/Deploy_SyncAggregateWithJmsService.xml");
    } catch (ResourceInitializationException e) {
      System.out.println("Received Expected ResourceInitializationException");
      return;
    }
    Assert.fail("Expected ResourceInitializationException Not Thrown. Instead Got Clean Run");
  }

  
  /**
   * Test binary compressed serialization between client and svc, and between 
   * service and remote delegate
   * 
   * @throws Exception
   */
  @Test
  public void testCompressedTypeFiltering() throws Exception {
    System.out.println("-------------- testCompressedTypeFiltering -------------");
    // Instantiate Uima-AS Client
    final BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(uimaAsEngine, relativePath + "/Deploy_RoomNumberAnnotator.xml");
    deployService(uimaAsEngine, relativePath + "/Deploy_MeetingDetectorTAE_RemoteRoomNumberBinary.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)), "MeetingDetectorTaeQueue");
    // Set an explicit getMeta (Ping)timeout
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 2000);
    // Set an explicit process timeout so to test the ping on timeout
    appCtx.put(UimaAsynchronousEngine.Timeout, 1000);
    appCtx.put(UimaAsynchronousEngine.SERIALIZATION_STRATEGY, "binary");

    runTest(null, uimaAsEngine, String.valueOf(getMasterConnectorURI(broker)),
            "MeetingDetectorTaeQueue", 3, PROCESS_LATCH);
  }

  
  /**
   * Tests Broker startup and shutdown
   */
  @Test
  public void testBrokerLifecycle() {
    System.out.println("-------------- testBrokerLifecycle -------------");
    System.out.println("UIMA_HOME=" + System.getenv("UIMA_HOME")
            + System.getProperty("file.separator") + "bin" + System.getProperty("file.separator")
            + "dd2spring.xsl");
  }
  @Test
  public void testForHang() throws Exception {
		System.out
		            .println("-------------- testForHang -------------");
		System.setProperty("BrokerURL", getMasterConnectorURI(broker));

		BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
		deployService(eeUimaEngine, relativePath + "/Deploy_ComplexAggregateMultiplier.xml");
		runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
		            100, PROCESS_LATCH);
  }

  /**
   * Tests programmatic generation of DD for deployment
   * 
   * @throws Exception
   */
  @Test
  public void testGenerateAndDeployPrimitiveDD() throws Exception {
	    System.out.println("-------------- testGenerateAndDeployPrimitiveDD -------------");
	  File directory = new File (".");
	  // Set up a context object containing basic service deployment
	  // information
	  ServiceContext context = new ServiceContextImpl("PersonTitle",
			  "PersonTitle Annotator Description",
			  directory.getCanonicalPath() + 
			  System.getProperty("file.separator")+
			  resourceDirPath+
			  System.getProperty("file.separator")+
			  "descriptors" +
			  System.getProperty("file.separator")+
			  "analysis_engine" +
			  System.getProperty("file.separator")+
			  "PersonTitleAnnotator.xml", 
			  "PersonTitleAnnotatorQueue",
			  getMasterConnectorURI(broker));
	  context.setCasPoolSize(2);
	  // create DD with default settings
	  UimaASPrimitiveDeploymentDescriptor dd = DeploymentDescriptorFactory
			  .createPrimitiveDeploymentDescriptor(context);

	  // Get default Error Handler for process and change error threshold
	  dd.getProcessErrorHandlingSettings().setThresholdCount(4);

	  // Two instances of AE in a jvm
	  dd.setScaleup(2);

	  // Generate deployment descriptor in xml format
	  String ddXML = dd.toXML();
	  System.out.println(ddXML);
	  
	  File tempFile = File.createTempFile("Deploy_PersonTitle", ".xml");
	  BufferedWriter out = new BufferedWriter(new FileWriter(tempFile));
	  out.write(ddXML);
	  out.close();
	  char FS = System.getProperty("file.separator").charAt(0);
	  
	// create Map to hold required parameters
	  Map<String,Object> appCtx = new HashMap<String,Object>();
	  appCtx.put(UimaAsynchronousEngine.DD2SpringXsltFilePath,
	            "../src/main/scripts/dd2spring.xsl".replace('/', FS));
	  appCtx.put(UimaAsynchronousEngine.SaxonClasspath,
	            "file:../src/main/saxon/saxon8.jar".replace('/', FS));	  
	  
	  BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
	  String aSpringContainerId =
	      eeUimaEngine.deploy(tempFile.getAbsolutePath(), appCtx);
	  
	  eeUimaEngine.undeploy(aSpringContainerId);
	  eeUimaEngine.stop();
	  
	  
  }
  @Test
  public void testDeployAggregateServiceWithFailingCollocatedCM() throws Exception {
    System.out.println("-------------- testDeployAggregateServiceWithFailingCollocatedCM -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithFailingCollocatedCM.xml");
    
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    appCtx.put(UimaAsynchronousEngine.Timeout, 0);
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 0);
    
    //addExceptionToignore(org.apache.uima.aae.error.UimaEEServiceException.class); 
    
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
  }
  
  
 
  public void getLargeCAS(CAS aCAS, File xmiFile) throws IOException, CollectionException {
	    FileInputStream inputStream = new FileInputStream(xmiFile);
	    try {
	    	XmiCasDeserializer.deserialize(inputStream, aCAS, false);
	    } catch (SAXException e) {
	      throw new CollectionException(e);
	    } finally {
	      inputStream.close();
	    }
	    
	  }
  /*
  @Test
  public void testLargeCAS() {
	    System.out.println("-------------- testLargeCAS -------------");
	    try {
		    // Instantiate Uima AS Client
		    BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
		    // Deploy Uima AS Primitive Service
//		    deployService(uimaAsEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");
		    deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
//		    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
//		            "PersonTitleAnnotatorQueue");
		    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
		            "NoOpAnnotatorQueue");
		    
		    
		    appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
		    appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
		    initialize(uimaAsEngine, appCtx);
		    waitUntilInitialized();
	        CAS cas = uimaAsEngine.getCAS();
	        getLargeCAS(cas, new File("C:/uima/largeCASTest/NYT_ENG_20070514.0065.out.xmi"));
	       
	        System.out.println("UIMA AS Client Sending CAS Request to a Service");
		    uimaAsEngine.sendCAS(cas);
		    uimaAsEngine.collectionProcessingComplete();
		    System.clearProperty("DefaultBrokerURL");
		    uimaAsEngine.stop();	  
	    	
	    } catch( Exception e) {
	    	e.printStackTrace();
	    }
  }
  */
  /**
   * Tests service quiesce and stop support. This test sets a CasPool to 1 to send just one CAS at a
   * time. After the first CAS is sent, a thread is started with a timer to expire before the reply
   * is received. When the timer expires, the client initiates quiesceAndStop on the top level
   * controller. As part of this, the top level controller stops its listeners on the input queue
   * (GetMeta and Process Listeners), and registers a callback with the InProcess cache. When the
   * cache is empty, meaning all CASes are processed, the cache notifies the controller which then
   * begins the service shutdown. Meanwhile, the client receives a reply for the first CAS, and
   * sends a second CAS. This CAS, will remain in the queue as the service has previously stopped
   * its listeners. The client times out after 10 seconds and shuts down.
   * 
   * @throws Exception
   */
  @Test
  public void testQuiesceAndStop() throws Exception {
    System.out.println("-------------- testQuiesceAndStop -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    // Set an explicit process timeout so to test the ping on timeout
    appCtx.put(UimaAsynchronousEngine.Timeout, 10000);
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 300);
    appCtx.put(UimaAsynchronousEngine.CasPoolSize, 10);
    String containers[] = new String[1];
    containers[0] = deployService(eeUimaEngine, relativePath + "/Deploy_ScaledPrimitiveAggregateAnnotator.xml");

    
//    String containers[] = new String[2];
 //   containers[0] = deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
   // containers[1] =  deployService(eeUimaEngine, relativePath
     //       + "/Deploy_AggregateAnnotatorWithInternalCM1000Docs.xml");
//    spinShutdownThread(eeUimaEngine, 5000, containers, SpringContainerDeployer.QUIESCE_AND_STOP);
    spinShutdownThread(eeUimaEngine, 5000, containers, SpringContainerDeployer.QUIESCE_AND_STOP);
    
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue", 5000, EXCEPTION_LATCH);
    //eeUimaEngine.stop();
  }
/*
  @Test
  public void testQuiesceAndStop2() throws Exception {
	    System.out.println("-------------- testQuiesceAndStop -------------");
	    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
	    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
	            "TopLevelTaeQueue");
	    // Set an explicit process timeout so to test the ping on timeout
	    appCtx.put(UimaAsynchronousEngine.Timeout, 10000);
	    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 300);
	    appCtx.put(UimaAsynchronousEngine.CasPoolSize, 1);
	    String containers[] = new String[1];
	    containers[0] = deployService(eeUimaEngine, relativePath + "/Deploy_ScaledPrimitiveAggregateAnnotator.xml");
	    
	    
	    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
	            "TopLevelTaeQueue", 100, PROCESS_LATCH);
	    System.out.println("------------ Undeploying ----------------");
	    eeUimaEngine.undeploy(containers[0] , SpringContainerDeployer.QUIESCE_AND_STOP);
//	    eeUimaEngine.stop();
	  }

  */
  
  
  
  @Test
  public void testStopNow() throws Exception {
    System.out.println("-------------- testAggregateWithFailedRemoteDelegate -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    String containers[] = new String[2];

    containers[0] = deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    containers[1] = deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateAnnotatorWithInternalCM1000Docs.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    // Set an explicit process timeout so to test the ping on timeout
    appCtx.put(UimaAsynchronousEngine.Timeout, 4000);
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 300);
    spinShutdownThread(eeUimaEngine, 3000, containers, SpringContainerDeployer.STOP_NOW);
    //  send may fail since we forcefully stop the service. Tolerate
    //  ResourceProcessException
    addExceptionToignore(ResourceProcessException.class); 
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue", 10, EXCEPTION_LATCH);
  }
  @Test
  public void testSendAndReceive() throws Exception  {
      BaseUIMAAsynchronousEngine_impl uimaAsEngine 
      	= new BaseUIMAAsynchronousEngine_impl();
      
      /*
      String id = deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      deployService(uimaAsEngine, relativePath + "/Deploy_AggregateAnnotator.xml");

      
      Map<String, Object> appCtx = buildContext(getMasterConnectorURI(broker),"MeetingDetectorTaeQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      //appCtx.put(UimaAsynchronousEngine.TargetSelectorProperty, "MeetingDetector2");
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();
      for (int i = 0; i < 15; i++) {
          CAS cas = uimaAsEngine.getCAS();
          cas.setDocumentText("Some Text");
          System.out.println(".... Sending CAS "+i);
          uimaAsEngine.sendAndReceiveCAS(cas);
          cas.release();
      }
      */
      
      String id = deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      deployService(uimaAsEngine, relativePath + "/Deploy_AggregateAnnotator.xml");
      // Deploy Uima AS Primitive Service
 //     deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      Map<String, Object> appCtx = buildContext(getMasterConnectorURI(broker).toString(),"TopLevelTaeQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();
      int errorCount = 0;
      List<AnalysisEnginePerformanceMetrics> componentMetricsList = 
    		  new ArrayList<AnalysisEnginePerformanceMetrics>();
      for (int i = 0; i < 15; i++) {
        CAS cas = uimaAsEngine.getCAS();
        cas.setDocumentText("Some Text");
  //      System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
        try {
          uimaAsEngine.sendAndReceiveCAS(cas,componentMetricsList);
          System.out.println("-------> Client Received Performance Metrics of Size:"+componentMetricsList.size());
          for( AnalysisEnginePerformanceMetrics m :componentMetricsList ) {
        	  System.out.println(".............. Component:"+m.getName()+" AnalysisTime:"+m.getAnalysisTime());
          }
        	//uimaAsEngine.sendCAS(cas);
          System.out.println("----------------------------------------------------");
          componentMetricsList.clear();
        } catch( Exception e) {
          errorCount++;
        } finally {
          cas.release();
          componentMetricsList.clear();
        }
      }
      uimaAsEngine.stop();
      
  }
  @Test
  public void testMultipleSyncClientsWithMultipleBrokers() throws Exception  {
	    System.out.println("-------------- testMultipleSyncClientsWithMultipleBrokers -------------");
	    
	    class RunnableClient implements Runnable {
	    	String brokerURL;
	    	BaseTestSupport testSupport;
            BaseUIMAAsynchronousEngine_impl uimaAsEngine;
	    	
	    	RunnableClient(String brokerURL,BaseTestSupport testSupport) {
	    		this.brokerURL = brokerURL;
	    		this.testSupport = testSupport;
	    	}
	    	public void initialize(String dd, String serviceEndpoint) throws Exception {
	    		uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
	            // Deploy Uima AS Primitive Service
	            deployService(uimaAsEngine, dd);

	    		@SuppressWarnings("unchecked")
			  Map<String, Object> appCtx = buildContext(brokerURL, serviceEndpoint);
		  	  appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
		  	  appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
		  	  testSupport.initialize(uimaAsEngine, appCtx);
		  	  waitUntilInitialized();
	    	}
			public void run() {
				try {
		            for (int i = 0; i < 1000; i++) {
			              CAS cas = uimaAsEngine.getCAS();
			              cas.setDocumentText("Some Text");
		//	              System.out.println("UIMA AS Client#"+ Thread.currentThread().getId()+" Sending CAS#"+(i + 1) + " Request to a Service Managed by Broker:"+brokerURL);
			              try {
				                uimaAsEngine.sendAndReceiveCAS(cas);
			              } catch( Exception e) {
			            	  e.printStackTrace();
			              } finally {
			                cas.release();
			              }
			            }
			            System.out.println("Thread:"+Thread.currentThread().getId()+" Completed run()");
			            uimaAsEngine.stop();
				} catch( Exception e) {
					e.printStackTrace();
				}

			}
	    	
	    }
	    
	    ExecutorService executor = Executors.newCachedThreadPool();

	    //	change broker URl in system properties
	    System.setProperty("BrokerURL", getMasterConnectorURI(broker).toString());
	    
	    RunnableClient client1 = 
	    		new RunnableClient(getMasterConnectorURI(broker), this);
	    client1.initialize(relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml", "NoOpAnnotatorQueue");

	    final BrokerService broker2 = setupSecondaryBroker(true);

	    //	change broker URl in system properties
	    System.setProperty("BrokerURL", broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());

	    RunnableClient client2 = 
	    		new RunnableClient(broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), this);
	    client2.initialize(relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml", "NoOpAnnotatorQueue");

	    Future<?> f1 = executor.submit(client1);
	    Future<?> f2 = executor.submit(client2);
	    f1.get();
	    f2.get();
	    executor.shutdownNow();
	    while( !executor.isShutdown() ) {
	    	synchronized(broker) {
	    		broker.wait(500);
	    	}
	    }
	    try {
		    broker2.stop();
//		    broker.stop();
//		    broker.waitUntilStopped();

	    } catch( Exception e) {
	    	// ignore this one. Always thrown by AMQ on stop().
	    }
	}
  

  @Test
  public void testAggregateHttpTunnelling() throws Exception {
    System.out.println("-------------- testAggregateHttpTunnelling -------------");
    // Create Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy remote service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    // Deploy top level aggregate that communicates with the remote via Http Tunnelling
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithHttpDelegate.xml");

    // Initialize and run the Test. Wait for a completion and cleanup resources.
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            10, CPC_LATCH);
  }

  
  
  @Test
  public void testClientHttpTunnellingToAggregate() throws Exception {
	  System.out.println("-------------- testClientHttpTunnellingToAggregate -------------");
    // Add HTTP Connector to the broker. 
    String httpURI = getHttpURI();
    // Create Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy remote service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotator.xml");
    // Initialize and run the Test. Wait for a completion and cleanup resources.
    System.out.println("-------- Connecting Client To Service: "+httpURI);
    runTest(null, eeUimaEngine, httpURI, "TopLevelTaeQueue", 1, CPC_LATCH);
  }
  @Test
  public void testClientHttpTunnelling() throws Exception {
    System.out.println("-------------- testClientHttpTunnelling -------------");
    String httpURI = getHttpURI();
    // Create Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy remote service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    // Initialize and run the Test. Wait for a completion and cleanup resources.
    System.out.println("-------- Connecting Client To Service: "+httpURI);
    runTest(null, eeUimaEngine, httpURI, "NoOpAnnotatorQueue", 1, PROCESS_LATCH);
  }


  @Test
  public void testClientHttpTunnellingWithDoubleByteText() throws Exception {
    System.out.println("-------------- testClientHttpTunnellingWithDoubleByteText -------------");

    BufferedReader in = null;
    try {
      File file = new File(relativeDataPath + "/DoubleByteText.txt");
      System.out.println("Checking for existence of File:" + file.getAbsolutePath());
      // Process only if the file exists
      if (file.exists()) {
        System.out
                .println(" *** DoubleByteText.txt exists and will be sent through http connector.");
        System.out.println(" ***   If the vanilla activemq release is being used,");
        System.out
                .println(" ***   and DoubleByteText.txt is bigger than 64KB or so, this test case will hang.");
        System.out
                .println(" *** To fix, override the classpath with the jar files in and under the");
        System.out
                .println(" ***   apache-uima-as/uima-as/src/main/apache-activemq-X.y.z directory");
        System.out.println(" ***   in the apache-uima-as source distribution.");

        String httpURI = getHttpURI();
      // Create Uima-AS Client
        BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
        // Deploy remote service
        deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");

        InputStream fis = new FileInputStream(file);
        Reader rd = new InputStreamReader(fis, "UTF-8");
        in = new BufferedReader(rd);
        // Set the double-byte text. This is what will be sent to the service
        String line = in.readLine();
        super.setDoubleByteText(line);
        int err = XMLUtils.checkForNonXmlCharacters(line);
        if (err >= 0) {
          fail("Illegal XML char at offset " + err);
        }
        System.out.println("-------- Connecting Client To Service: "+httpURI);
        // Initialize and run the Test. Wait for a completion and cleanup resources.
        runTest(null, eeUimaEngine, httpURI, "NoOpAnnotatorQueue", 1, CPC_LATCH);
      }
    } catch (Exception e) {
      // Double-Byte Text file not present. Continue on with the next test
      e.printStackTrace();
      fail("Could not complete test");
    } finally {
      if (in != null) {
        in.close();
      }
    }
  }
  
  /**
   * Tests support for ActiveMQ failover protocol expressed in broker
   * URL as follows "failover:(tcp:IP:Port1,tcp:IP:Port2)". The test launches a secondary
   * broker, launches a Primitive service that uses that broker,
   * and finally configures the UIMA AS Client to connect to the secondary broker
   * and specifies an alternative broker on a different port. This test 
   * only tests ability of UIMA AS to handle a complex URL, and it does *not*
   * test the actual failover from one broker to the next. 
   * 
   * @throws Exception
   */
  @Test
  public void testBrokerFailoverSupportUsingHTTP() throws Exception  {
    System.out.println("-------------- testBrokerFailoverSupportUsingTCP -------------");
     // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      BrokerService broker2 = setupSecondaryBroker(false);

      String bhttpURL = addHttpConnector(broker2, DEFAULT_HTTP_PORT2);
      String burl = "failover:("+bhttpURL+","+broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString().replaceAll("tcp","http")+")?randomize=false";
      System.setProperty("BrokerURL", burl); 
      // Deploy Uima AS Primitive Service
      deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
      Map<String, Object> appCtx = buildContext(burl,"NoOpAnnotatorQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();
      int errorCount = 0;
      for (int i = 0; i < 15; i++) {
        CAS cas = uimaAsEngine.getCAS();
        cas.setDocumentText("Some Text");
      //  System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
        try {
          uimaAsEngine.sendAndReceiveCAS(cas);
        } catch( Exception e) {
          errorCount++;
        } finally {
          cas.release();
        }
      }
      uimaAsEngine.stop();
      super.cleanBroker(broker2);
      broker2.stop();
      broker2.waitUntilStopped();

  }
  
  /**
   * Tests support for ActiveMQ failover protocol expressed in broker
   * URL as follows "failover:(tcp:IP:Port1,tcp:IP:Port2)". The test launches a secondary
   * broker, launches a Primitive service that uses that broker,
   * and finally configures the UIMA AS Client to connect to the secondary broker
   * and specifies an alternate broker on a different port. This test 
   * only tests ability of UIMA AS to handle a complex URL, and it does *not*
   * test the actual failover from one broker to the next. 
   * 
   * @throws Exception
   */
  @Test
  public void testBrokerFailoverSupportUsingTCP() throws Exception  {
    System.out.println("-------------- testBrokerFailoverSupportUsingTCP -------------");
     // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      
      BrokerService broker2 = setupSecondaryBroker(true);
      // Deploy Uima AS Primitive Service
      deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
      Map<String, Object> appCtx = buildContext("failover:("+System.getProperty("BrokerURL")+","+getBrokerUri()+")?randomize=false","NoOpAnnotatorQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();
      int errorCount = 0;
      for (int i = 0; i < 15; i++) {
        CAS cas = uimaAsEngine.getCAS();
        cas.setDocumentText("Some Text");
      //  System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
        try {
          uimaAsEngine.sendAndReceiveCAS(cas);
        } catch( Exception e) {
          errorCount++;
        } finally {
          cas.release();
        }
      }
      uimaAsEngine.stop();
      super.cleanBroker(broker2);

      broker2.stop();
      broker2.waitUntilStopped();

  }
  
  /**
   * This test starts a secondary broker, starts NoOp Annotator, and
   * using synchronous sendAndReceive() sends 10 CASes for analysis. Before sending 11th, the test
   * stops the secondary broker and sends 5 more CASes. All CASes sent after
   * the broker shutdown result in GetMeta ping and a subsequent timeout.
   * @throws Exception
   */
  
  @Test
  public void testSyncClientRecoveryFromBrokerStop() throws Exception  {
    System.out.println("-------------- testSyncClientRecoveryFromBrokerStop -------------");
    //System.setProperty("uima.as.enable.jmx","false"); 
    // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      BrokerService broker2 = setupSecondaryBroker(true);
      // Deploy Uima AS Primitive Service
      deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
      Map<String, Object> appCtx = 
      buildContext(broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");
      appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 1100);
      appCtx.put(UimaAsynchronousEngine.Timeout, 0);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 300);
      appCtx.put(UimaAsynchronousEngine.CasPoolSize,15);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();
      int errorCount = 0;
      for (int i = 0; i < 15; i++) {
        
        if ( i == 10 ) {
          //  Stop the broker
          broker2.stop();
          broker2.waitUntilStopped();
          Timer timer = new Timer();
		  timer.schedule(new StartBrokerTask(broker2, this),10000);
        }
        CAS cas = uimaAsEngine.getCAS();
        cas.setDocumentText("Some Text");
     //   System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
        try {
          uimaAsEngine.sendAndReceiveCAS(cas);
        } catch( Exception e) {
          errorCount++;
          System.out.println("Client Received Expected Error on CAS:"+(i+1)+" ErrorCount:"+errorCount);
        } finally {
          cas.release();
        }
      }
     
      uimaAsEngine.stop();
      //  expecting 5 failures due to broker missing
//      if ( errorCount != 5 ) {
//        fail("Expected 5 failures due to broker down, instead received:"+errorCount+" failures");
//      }
      broker2.stop();
      broker2.waitUntilStopped();

  }

  @Test
  public void testAsyncClientRecoveryFromBrokerStopAndRestart() throws Exception  {
    System.out.println("-------------- testAsyncClientRecoveryFromBrokerStopAndRestart -------------");

     BrokerService broker2 = setupSecondaryBroker(true);
     // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      // Deploy Uima AS Primitive Service
      String id = deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
      String brokerUri = broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString();
      Map<String, Object> appCtx = 
        buildContext(brokerUri, "NoOpAnnotatorQueue");

      appCtx.put(UimaAsynchronousEngine.Timeout, 200);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 200);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();
      int delay = 1;
      for (int i = 0; i < 50; i++) {
        if ( i == 10 ) {
          broker2.stop();
          broker2.waitUntilStopped();
      	  System.out.println("..... Stopped broker ............................");
      	  Timer timer = new Timer();
		  timer.schedule(new StartBrokerTask(broker2, this),500);
		  delay = 200;
	     }
        synchronized(appCtx) {
        	try {
        		appCtx.wait(delay);
        	} catch( InterruptedException eee) {
        		
        	}
        }
/*

        } else if ( i == 15 ) {
          broker2 = setupSecondaryBroker(true);
          broker2.waitUntilStarted();
      	System.out.println("..... Restarted broker ............................");

        }
        
       */
        CAS cas = uimaAsEngine.getCAS();
        cas.setDocumentText("Some Text");
        System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
        uimaAsEngine.sendCAS(cas);
      }
      
      
      uimaAsEngine.stop();

      super.cleanBroker(broker2);

      broker2.stop();
      broker2.waitUntilStopped();
  }

  /**
   * Tests recovery from a broker restart when running multiple instances of 
   * UIMA AS client in the same JVM. The scenario is:
   * 1) start broker
   * 2) start service
   * 3) create 1st client and initialize it
   * 4) create 2nd client and initialize it
   * 5) send 1 CAS from client#1
   * 6) when reply is received kill broker and restart it
   * 7) send 2 CAS from client#1
   * 8) CAS #2 fails, but forces SharedConnection to reconnect to broker
   * 9) CAS#3 and #4 are sent from client#1
   * 10) CASes 1-4 are sent from client#2 without error
   * 
   * @throws Exception
   */

  /*
  @Test
  public void testMultipleClientsRecoveryFromBrokerStopAndRestart() throws Exception  {
    System.out.println("-------------- testMultipleClientsRecoveryFromBrokerStopAndRestart -------------");
      
    BrokerService broker2 = setupSecondaryBroker(true);
      // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaClient1 = new BaseUIMAAsynchronousEngine_impl();
      // Deploy Uima AS Primitive Service
      deployService(uimaClient1, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
      Map<String, Object> appCtx = 
        buildContext(broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");
    
      appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 400);
      initialize(uimaClient1, appCtx);
      waitUntilInitialized();
      
      // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaClient2 = new BaseUIMAAsynchronousEngine_impl();
      Map<String, Object> appCtx2 = 
      buildContext(broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");

      
      appCtx2.put(UimaAsynchronousEngine.Timeout, 1100);
      appCtx2.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      appCtx2.put(UimaAsynchronousEngine.GetMetaTimeout, 400);
      initialize(uimaClient2, appCtx2);
      waitUntilInitialized();
      
      
      int errorCount=0;
      for (int i = 0; i < 4; i++) {
        //  Stop broker before second CAS is sent to the service
        if ( i == 1 ) {
          broker2.stop();
          broker2.waitUntilStopped();

          //  restart broker before 3rd CAS is sent
          //  restart the broker 
          broker2 = setupSecondaryBroker(true);
          broker2.waitUntilStarted();

        } 
        CAS cas = uimaClient1.getCAS();
        cas.setDocumentText("Some Text");
        System.out.println("UIMA AS Client#1 Sending CAS#" + (i + 1) + " Request to a Service");
        try {
          uimaClient1.sendAndReceiveCAS(cas);
          System.out.println("UIMA AS Client#1 Received Reply For CAS:"+(i+1));

        } catch( Exception e) {
          errorCount++;
          System.out.println("UIMA AS Client#1 Received Expected Error on CAS:"+(i+1));
        } finally {
          cas.release();
        }
      }
      System.out.println("Done with UIMA AS Client#1");

      for (int i = 0; i < 4; i++) {
        CAS cas = uimaClient2.getCAS();
        cas.setDocumentText("Some Text");
        System.out.println("UIMA AS Client#2 Sending CAS#" + (i + 1) + " Request to a Service");
        try {
 
          uimaClient2.sendAndReceiveCAS(cas);
        } catch( Exception e) {
          errorCount++;
          System.out.println("UIMA AS Client#2 Received Expected Error on CAS:"+(i+1));
        } finally {
          cas.release();
        }
      }
      
      uimaClient1.stop();
      uimaClient2.stop();
      super.cleanBroker(broker2);

      broker2.stop();
      broker2.waitUntilStopped();

  }
  */
  /**
   * Tests ability of an aggregate to recover from a Broker restart. The broker managing
   * delegate's input queue is stopped after 1st CAS is fully processed. As part of error
   * handling the listener on delegate temp reply queue is stopped and a delegate marked 
   * as FAILED. The aggregate error handling is configured to retry the command and as
   * part of retry a new temp queue and a listener are created for the delegate when
   * the broker is restarted. When a 2nd CAS is sent from a client to the aggregate the 
   * aggregate will force retry of the previous CAS. Once this is done, 2nd CAS is sent 
   * to the delegate and processing continues.
   * @throws Exception
   */
  @Test
  public void testAggregateRecoveryFromBrokerStopAndRestart() throws Exception  {
    System.out.println("-------------- testAggregateRecoveryFromBrokerStopAndRestart -------------");
    runAggregateRecoveryFromBrokerStopAndRestart("Deploy_AggregateWithRemoteNoOpOnBroker8200.xml");

  }
  /**
   * This test starts a secondary broker, starts NoOp Annotator, and
   * using asynchronous send() sends a total of 15 CASes for analysis. After processing 11th
   * the test stops the secondary broker and sends 4 more CASes which fails due to broker not running.
   * 
   * @throws Exception
   */
  @Test
  public void testAsyncClientRecoveryFromBrokerStop() throws Exception  {
    System.out.println("-------------- testAsyncClientRecoveryFromBrokerStop -------------");
    
    BrokerService broker2 = setupSecondaryBroker(true);
     // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
      Map<String, Object> appCtx = 
      buildContext(broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      appCtx.put(UimaAsynchronousEngine.CasPoolSize, 15);

      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();
      
      for (int i = 0; i < 15; i++) {
        
        if ( i == 10 ) {
          broker2.stop();
          broker2.waitUntilStopped();

        }
        CAS cas = uimaAsEngine.getCAS();
        cas.setDocumentText("Some Text");
       // System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
        uimaAsEngine.sendCAS(cas);
      }
      
      uimaAsEngine.stop();
      super.cleanBroker(broker2);
      broker2.stop();
      broker2.waitUntilStopped();

  }
/**
   * Tests ability of an aggregate to recover from a Broker restart. The broker managing
   * delegate's input queue is stopped after 1st CAS is fully processed. As part of error
   * handling the listener on delegate temp reply queue is stopped and a delegate marked 
   * as FAILED. The aggregate error handling is configured with no retries. After the broker
   * is stopped, the listener on a temp reply queue is shutdown. When the broker is restarted
   * the client sends a new CAS which forces creation of a new temp queue and new listener.
   * @throws Exception
   */
  @Test
  public void testAggregateRecoveryFromBrokerStopAndRestartNoDelegateRetries() throws Exception  {
    System.out.println("-------------- testAggregateRecoveryFromBrokerStopAndRestartNoDelegateRetries -------------");
    runAggregateRecoveryFromBrokerStopAndRestart("Deploy_AggregateWithRemoteNoOpOnBroker8200NoRetry.xml");
  }
  
  private void runAggregateRecoveryFromBrokerStopAndRestart(String aggregateDescriptor ) throws Exception {
    BrokerService broker2 = setupSecondaryBroker(false);
    System.setProperty("BrokerURL", broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());

      // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaClient1 = new BaseUIMAAsynchronousEngine_impl();
      // Deploy Uima AS Primitive Service
      deployService(uimaClient1, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");
      deployService(uimaClient1, relativePath + "/"+aggregateDescriptor);
      Map<String, Object> appCtx = 
      buildContext(getMasterConnectorURI(broker), "TopLevelTaeQueue");

      appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      initialize(uimaClient1, appCtx);
      waitUntilInitialized();
      
      
      int errorCount=0;
      for (int i = 0; i < 10; i++) {
        //  Stop broker before second CAS is sent to the service
        if ( i == 1 ) {
          System.out.println("Stopping Secondary Broker Running on Port:"+broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());
          broker2.stop();
          broker2.waitUntilStopped();

          //  restart broker before 3rd CAS is sent
          //  restart the broker 
          broker2 = setupSecondaryBroker(true);
          broker2.waitUntilStarted();

        } 
        CAS cas = uimaClient1.getCAS();
        cas.setDocumentText("Some Text");
//        System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
        try {
          uimaClient1.sendAndReceiveCAS(cas);
   //       System.out.println("UIMA AS Client Received Reply For CAS#" + (i + 1) );
        } catch( Exception e) {
          errorCount++;
          System.out.println("UIMA AS Client Received Expected Error on CAS:"+(i+1));
        } finally {
          cas.release();
        }
      }
      uimaClient1.stop();
      System.out.println("Stopping Broker - wait ...");
      super.cleanBroker(broker2);

      broker2.stop();
      broker2.waitUntilStopped();

  }
  /**
   * Tests sending CPC after CAS timeout. The service is a Primitive taking 
   * 6 seconds to process a CAS. The client waits for 5 secs to force
   * a timeout. This test forces the client to send GetMeta Ping and to
   * 'hold' the subsequent CPC request.
   *   
   * @throws Exception
   */
  @Test
  public void testCpcAfterCasTimeout() throws Exception  {
    System.out.println("-------------- testCpcAfterCasTimeout -------------");
     // Instantiate Uima AS Client
      BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
      deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorAWithLongDelay.xml");
      Map<String, Object> appCtx = buildContext(getMasterConnectorURI(broker),
      "NoOpAnnotatorAQueue");
      appCtx.put(UimaAsynchronousEngine.Timeout, 5000);
      appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
      initialize(uimaAsEngine, appCtx);
      waitUntilInitialized();
      
      for( int i=0; i < 3; i++ ) {
        CAS cas = uimaAsEngine.getCAS();
        cas.setDocumentText("Some Text");
        uimaAsEngine.sendCAS(cas);  // will timeout after 5 secs
        uimaAsEngine.collectionProcessingComplete();  // the CPC should not
        // be sent to a service until the timeout occurs.
      }
      uimaAsEngine.stop();
  }
  @Test
  public void testClientCRProcess() throws Exception {
	    System.out.println("-------------- testClientCRProcess -------------");
	    super.resetCASesProcessed();
	    
	    // Instantiate Uima AS Client
	    final BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
//	    UIMAFramework.getLogger(BaseUIMAAsynchronousEngineCommon_impl.class).setLevel(Level.FINEST);
//	    UIMAFramework.getLogger(BaseUIMAAsynchronousEngine_impl.class).setLevel(Level.FINEST);
//	    UIMAFramework.getLogger().setLevel(Level.FINEST);
//	    UIMAFramework.getLogger().setOutputStream(System.out);

	    // Deploy Uima AS Primitive Service
	    deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithLongDelay.xml");
	    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
	            "NoOpAnnotatorQueueLongDelay");
	    appCtx.put(UimaAsynchronousEngine.Timeout, 0);
	    appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
	    appCtx.put(UimaAsynchronousEngine.CasPoolSize,10);
	    
	    String collectionReaderDescriptor =
	    		resourceDirPath + System.getProperty("file.separator") +
	    		"descriptors"+ System.getProperty("file.separator") +
	    		"collection_reader"+ System.getProperty("file.separator") +
	    		"FileSystemCollectionReader.xml";
	   		  
         // add Collection Reader if specified
         try {
             CollectionReaderDescription collectionReaderDescription = 
                     UIMAFramework.getXMLParser()
                             .parseCollectionReaderDescription(new XMLInputSource(collectionReaderDescriptor));
             collectionReaderDescription.getCollectionReaderMetaData().
             	getConfigurationParameterSettings().
             		setParameterValue("InputDirectory", relativeDataPath);
             CollectionReader collectionReader = UIMAFramework
                     .produceCollectionReader(collectionReaderDescription);
             uimaAsEngine.setCollectionReader(collectionReader);	    
         } catch( Throwable e) {
        	 e.printStackTrace();
         }
	    
	    initialize(uimaAsEngine, appCtx);
	    waitUntilInitialized();
	    
	    uimaAsEngine.process();

	    Assert.assertEquals(8, getNumberOfCASesProcessed());
	    System.clearProperty("DefaultBrokerURL");
	    uimaAsEngine.stop();
	  }
private class Killer {
	Timer timer;
	public Killer(int secs) {
		timer = new Timer();
		timer.schedule(new KillerTask(), secs*1000);
	}
	class KillerTask extends TimerTask {
		public void run() {
			System.out.println("----------------- KillerTask calling System.exit()");
			timer.cancel();
			System.exit(0);
		}
	}
}
  @Test
  public void testClientProcess() throws Exception {
    System.out.println("-------------- testClientProcess -------------");
    
    // Instantiate Uima AS Client
    final BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima AS Primitive Service
    deployService(uimaAsEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
			"PersonTitleAnnotatorQueue");
    appCtx.put(UimaAsynchronousEngine.Timeout, 0);
    appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
    appCtx.put(UimaAsynchronousEngine.CasPoolSize,2);
    appCtx.put(UimaAsynchronousEngine.SERIALIZATION_STRATEGY, "xmi");
    
    initialize(uimaAsEngine, appCtx);
    waitUntilInitialized();

    for (int i = 0; i < 50; i++) {
      CAS cas = uimaAsEngine.getCAS();
      cas.setDocumentText("Some Text");
//      System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
      uimaAsEngine.sendCAS(cas);
    }
    
    uimaAsEngine.collectionProcessingComplete();
    System.clearProperty("DefaultBrokerURL");
    uimaAsEngine.stop();
  }
  
  
  @Test
  public void testClientProcessTimeout() throws Exception {
    System.out
            .println("-------------- testClientProcessTimeout -------------");
    // Instantiate Uima AS Client
    BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima AS Primitive Service
    deployService(uimaAsEngine, relativePath + "/Deploy_NoOpAnnotatorWithLongDelay.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueueLongDelay");
    appCtx.put(UimaAsynchronousEngine.Timeout, 300);
    initialize(uimaAsEngine, appCtx);
    waitUntilInitialized();
    Object o = new Object();
    for (int i = 0; i < 6; i++) {
      CAS cas = uimaAsEngine.getCAS();
      cas.setDocumentText("Some Text");
 //     System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
      uimaAsEngine.sendCAS(cas);
      synchronized(o) {
    	  o.wait(1000);
      }
    }
    
    uimaAsEngine.collectionProcessingComplete();
    uimaAsEngine.stop();
  
  
  }
  
  
  @Test
  public void testClientBrokerPlaceholderSubstitution() throws Exception {
    System.out.println("-------------- testClientBrokerPlaceholderSubstitution -------------");
    // Instantiate Uima AS Client
    BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima AS Primitive Service
    deployService(uimaAsEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");

    System.setProperty( "defaultBrokerURL", getMasterConnectorURI(broker));
    Map<String, Object> appCtx = buildContext("${defaultBrokerURL}","PersonTitleAnnotatorQueue");

    initialize(uimaAsEngine, appCtx);
    waitUntilInitialized();
    for (int i = 0; i < 10; i++) {
      CAS cas = uimaAsEngine.getCAS();
      cas.setDocumentText("Some Text");
 //     System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
      uimaAsEngine.sendCAS(cas);
    }
    uimaAsEngine.collectionProcessingComplete();
    uimaAsEngine.stop();
    
  }

  @Test
  public void testClientEndpointPlaceholderSubstitution() throws Exception {
	    System.out.println("-------------- testClientEndpointPlaceholderSubstitution -------------");
	    // Instantiate Uima AS Client
	    BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
	    // Deploy Uima AS Primitive Service
	    deployService(uimaAsEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");

	    // Nest the placeholders in the broker & endpoint strings
	    String url = getMasterConnectorURI(broker);
	    System.setProperty( "defaultBrokerURL", url.substring(2,url.length()-2));
	    String brokerUrl = url.substring(0,2) + "${defaultBrokerURL}" + url.substring(url.length()-2);	    
	    System.setProperty( "PersonTitleEndpoint", "TitleAnnotator");
	    String endpoint = "Person${PersonTitleEndpoint}Queue";  // "PersonTitleAnnotatorQueue"
	    Map<String, Object> appCtx = buildContext(brokerUrl, endpoint);

	    initialize(uimaAsEngine, appCtx);
	    waitUntilInitialized();
	    for (int i = 0; i < 10; i++) {
	      CAS cas = uimaAsEngine.getCAS();
	      cas.setDocumentText("Some Text");
//	      System.out.println("UIMA AS Client Sending CAS#" + (i + 1) + " Request to a Service");
	      uimaAsEngine.sendCAS(cas);
	    }
	    uimaAsEngine.collectionProcessingComplete();
	    uimaAsEngine.stop();
	    
	  }
  
  @Test
  public void testClientCpcTimeout() throws Exception {
    System.out.println("-------------- testClientCpcTimeout -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    addExceptionToignore(org.apache.uima.aae.error.UimaASCollectionProcessCompleteTimeout.class);

    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithCpCDelay.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueue");
    // Set an explicit CPC timeout as exceptions thrown in the 2nd annotator's CPC don't reach the
    // client.
    appCtx.put(UimaAsynchronousEngine.CpcTimeout, 2000);
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueue", 1, CPC_LATCH); // PC_LATCH);
  }

  /**
   * Tests handling of multiple calls to initialize(). A subsequent call to initialize should result
   * in ResourceInitializationException.
   * 
   * @throws Exception
   */
  @Test
  public void testInvalidInitializeCall() throws Exception {
    System.out.println("-------------- testInvalidInitializeCall -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();

    deployService(eeUimaEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "PersonTitleAnnotatorQueue");

    try {
      initialize(eeUimaEngine, appCtx);
      waitUntilInitialized();
      System.out.println("First Initialize Call Completed");
      eeUimaEngine.initialize(appCtx);
      fail("Subsequent call to initialize() did not return expected exception:"
              + UIMA_IllegalStateException.class
              + " Subsequent call to initialize succeeded with no error");
    } catch (ResourceInitializationException e) {
      if (e.getCause() != null && !(e.getCause() instanceof UIMA_IllegalStateException)) {
        fail("Invalid Exception Thrown. Expected:" + UIMA_IllegalStateException.class
                + " Received:" + e.getClass());
      } else {
        System.out.println("Received Expected Exception:" + UIMA_IllegalStateException.class);
      }
    } catch (ServiceShutdownException e) {
      // expected
    } finally {
      eeUimaEngine.stop();
    }
  }

  /**
   * Tests deployment of a primitive Uima-AS Service (PersontTitleAnnotator). Deploys the primitive
   * in the same jvm using Uima-AS Client API and blocks on a monitor until the Uima Client calls
   * initializationComplete() method. Once the primitive service starts it is expected to send its
   * metadata to the Uima client which in turn notifies this object with a call to
   * initializationComplete() where the monitor is signaled to unblock the thread. This code will
   * block if the Uima Client does not call initializationComplete()
   * 
   * @throws Exception
   */
  @Test
  public void testDeployPrimitiveService() throws Exception {
    System.out.println("-------------- testDeployPrimitiveService -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "PersonTitleAnnotatorQueue", 0, EXCEPTION_LATCH);
  }
  @Test
  public void testDeployPrimitiveServiceWithTargeting() throws Exception {
    System.out.println("-------------- testDeployPrimitiveService -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    System.setProperty(UimaAsynchronousEngine.TargetSelectorProperty,"ServiceOne");
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithTargetingSupport.xml");
    
    System.setProperty(UimaAsynchronousEngine.TargetSelectorProperty,"ServiceTwo");
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithTargetingSupport.xml");

    System.setProperty(UimaAsynchronousEngine.TargetSelectorProperty,"ServiceThree");
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithTargetingSupport.xml");

    System.getProperties().remove(UimaAsynchronousEngine.TargetSelectorProperty);
    Map<String, Object> appCtx = buildContext(getMasterConnectorURI(broker),
    "NoOpAnnotatorQueue");
    int toProcess = 10;
    try {
      initialize(eeUimaEngine, appCtx);
      waitUntilInitialized();
      String tsId = "ServiceOne";
      
      for( int i=0; i < 10; i++ ) {
          CAS cas = eeUimaEngine.getCAS();
          //cas.setDocumentText("");
          eeUimaEngine.sendAndReceiveCAS(cas, null, tsId);
          String serviceIdWhereCasWasProcessed = cas.getDocumentText();
          if ( !serviceIdWhereCasWasProcessed.equals(tsId)) {
        	  fail("Received Reply from a Wrong Service - Expected: "+tsId+" Instead Received "+serviceIdWhereCasWasProcessed);
          }
          cas.release();
      }
      
      tsId = "ServiceTwo";
      for( int i=0; i < 10; i++ ) {
          CAS cas = eeUimaEngine.getCAS();
          //cas.setDocumentText("");
          eeUimaEngine.sendAndReceiveCAS(cas, null, tsId);
          String serviceIdWhereCasWasProcessed = cas.getDocumentText();
          if ( !serviceIdWhereCasWasProcessed.equals(tsId)) {
        	  fail("Received Reply from a Wrong Service - Expected: "+tsId+" Instead Received "+serviceIdWhereCasWasProcessed);
          }
          cas.release();
      }
      
      tsId = "ServiceThree";
      
      
      for( int i=0; i < toProcess; i++ ) {
    	  CAS cas = eeUimaEngine.getCAS();
          //cas.setDocumentText("");
          eeUimaEngine.sendCAS(cas, tsId);
          //cas.release();
      }
    } catch( Exception e) {
    	e.printStackTrace();
    } finally {
    	
    	while( getNumberOfCASesProcessed() < toProcess ) {
    		if ( unexpectedException ) {
    			fail("Service Targeting Failed");
    			break;
    		}
    		synchronized(this) {
    			wait(100);
    		}
    	}
    	eeUimaEngine.stop();
    }
//    synchronized(this) {
//    	wait(0);
//    }
    
 //   runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
  //          "PersonTitleAnnotatorQueue", 0, EXCEPTION_LATCH);
  }
  @Test
  public void testDeployPrimitiveServiceWithInitFailure() throws Exception {
	    System.out.println("-------------- testDeployPrimitiveServiceWithInitFailure -------------");
	    // Instantiate Uima-AS Client
	    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
	    // Deploy Uima-AS Primitive Service
	    try {
		    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithInitException.xml");
	    } catch( ResourceInitializationException e) {
	    	System.out.println("Received Expected ResourceInitializationException On Service Deploy");
		    return;
	    }
	    fail("Expected ResourceInitializationException Not Thrown from deployed Service.");
  }
  
  @Test
  public void testTypeSystemMerge() throws Exception {
    System.out.println("-------------- testTypeSystemMerge -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath+ "/Deploy_GovernmentOfficialRecognizer.xml");
    deployService(eeUimaEngine, relativePath+ "/Deploy_NamesAndPersonTitlesRecognizer.xml");
    deployService(eeUimaEngine, relativePath+ "/Deploy_TokenSentenceRecognizer.xml");
    deployService(eeUimaEngine, relativePath+ "/Deploy_AggregateToTestTSMerge.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
    "TopLevelTaeQueue");
    
    try {
      initialize(eeUimaEngine, appCtx);
      waitUntilInitialized();
      //  Check if the type system returned from the service contains
      //  expected types
      CAS cas = eeUimaEngine.getCAS();
      TypeSystem ts = cas.getTypeSystem();
      //  "example.EmailsAddress" type was 'contributed' by the Flow Controller
      if ( ts.getType("example.EmailAddress") == null ) {
        fail("Incomplete Type system. Expected Type 'example.EmailAddress' missing from the CAS type system");
      } else if ( ts.getType("example.GovernmentOfficial") == null) {
        fail("Incomplete Type system. Expected Type 'example.GovernmentOfficial' missing from the CAS type system");
      } else if ( ts.getType("example.Name") == null) {
        fail("Incomplete Type system. Expected Type 'example.Name' missing from the CAS type system");
      } else if ( ts.getType("example.PersonTitle") == null) {
        fail("Incomplete Type system. Expected Type 'example.PersonTitle' missing from the CAS type system");
      } else if ( ts.getType("example.PersonTitleKind") == null) {
        fail("Incomplete Type system. Expected Type 'example.PersonTitleKind' missing from the CAS type system");
      } else if ( ts.getType("org.apache.uima.examples.tokenizer.Sentence") == null) {
        fail("Incomplete Type system. Expected Type 'org.apache.uima.examples.tokenizer.Sentence' missing from the CAS type system");
      } else if ( ts.getType("org.apache.uima.examples.tokenizer.Token") == null) {
        fail("Incomplete Type system. Expected Type 'org.apache.uima.examples.tokenizer.Token' missing from the CAS type system");
      } 
      
    } catch (ResourceInitializationException e) {
        fail("Initialization Exception");
    } catch (Exception e) {
    } finally {
      eeUimaEngine.stop();
    }
  }

  /**
   * Tests detection of misconfiguration between deployement descriptor and AE descriptor.
   * The AE descriptor is configured to allow one instance of the AE while the deployment
   * descriptor attempts to scale out the service containing the AE. ResourceInitializationException
   * is expected.
   * 
   * @throws Exception
   */
  @Test
  public void testMultiInstanceDeployFailureInPrimitiveService() throws Exception {
    System.out.println("-------------- testMultiInstanceDeployFailureInPrimitiveService -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    try {
      // Deploy Uima-AS Primitive Service
      deployService(eeUimaEngine, relativePath + "/Deploy_ScaledPersonTitleAnnotator.xml");
      fail("Expected ResourceInitializationException Due to Misconfiguration but instead the service initialized successfully");
    } catch ( ResourceInitializationException e) {
      // expected
    }
  }
  /**
   * Tests processing of an Exception that a service reports on CPC
   * 
   * @throws Exception
   */
  @Test
  public void testDeployAggregateWithDelegateCpCException() throws Exception {
    System.out.println("-------------- testDeployAggregateWithDelegateCpCException -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithCpCException.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotator.xml");
    addExceptionToignore(org.apache.uima.aae.error.UimaEEServiceException.class);

    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    initialize(eeUimaEngine, appCtx);
    waitUntilInitialized();
    System.out.println("Client Initialized");
    CAS cas = eeUimaEngine.getCAS();
    for (int i = 0; i < 3; i++) {
      eeUimaEngine.sendAndReceiveCAS(cas);
      cas.reset();
    }
    System.out.println("Client Sending CPC");

    // Send CPC. The service should recreate a session and send CPC reply
    eeUimaEngine.collectionProcessingComplete();

    // Now send some CASes and sleep to let the inactivity timer pop again
    for (int i = 0; i < 3; i++) {
      eeUimaEngine.sendAndReceiveCAS(cas); // This will start a timer on reply queue
      cas.reset();
    }
    // Send another CPC
    eeUimaEngine.collectionProcessingComplete();

    eeUimaEngine.stop();

  }

  @Test
  public void testDeployPrimitiveServiceWithCpCException() throws Exception {
    System.out.println("-------------- testDeployPrimitiveServiceWithCpCException -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithCpCException.xml");
    // Add expected exception so that we release CPC Latch
    addExceptionToignore(org.apache.uima.aae.error.UimaEEServiceException.class);

    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueue", 1, PROCESS_LATCH);
  }

  /**
   * Tests sending CPC request from a client that does not send CASes to a service
   * 
   * @throws Exception
   */
  @Test
  public void testCpCWithNoCASesSent() throws Exception {
    System.out.println("-------------- testCpCWithNoCASesSent -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(uimaAsEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "PersonTitleAnnotatorQueue");
    initialize(uimaAsEngine, appCtx);
    waitUntilInitialized();

    for (int i = 0; i < 10; i++) {
      System.out.println("UIMA AS Client Sending CPC Request to a Service");
      uimaAsEngine.collectionProcessingComplete();
    }
    uimaAsEngine.stop();
  }

  /**
   * Tests inactivity timeout on a reply queue. The service stops a session after 5 seconds of
   * sending GetMeta reply. The client than waits for 10 seconds and sends a CPC.
   * 
   * @throws Exception
   */
  @Test
  public void testServiceInactivityTimeoutOnReplyQueue() throws Exception {
    System.out.println("-------------- testServiceInactivityTimeoutOnReplyQueue -------------");
    String sessionTimeoutOverride = System.getProperty("SessionTimeoutOverride");
    System.setProperty("SessionTimeoutOverride", "5000");

    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueue");
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 1000);
    initialize(eeUimaEngine, appCtx);
    waitUntilInitialized();
    System.out.println("Client Initialized");

    CAS cas = eeUimaEngine.getCAS();
    eeUimaEngine.sendAndReceiveCAS(cas); // This will start a timer on reply queue
    cas.reset();
    // Now sleep for 8 seconds to let the service timeout on its reply queue due
    // to a 5 second inactivity timeout
    Thread.currentThread().sleep(8000);
    System.out.println("Client Sending CPC");

    // Send CPC. The service should recreate a session and send CPC reply
    eeUimaEngine.collectionProcessingComplete();
    // Now send some CASes and sleep to let the inactivity timer pop again
    for (int i = 0; i < 5; i++) {
      eeUimaEngine.sendAndReceiveCAS(cas); // This will start a timer on reply queue
      cas.reset();
      if (i == 3) {
        Thread.currentThread().sleep(8000);
      }
    }
    // Send another CPC
    eeUimaEngine.collectionProcessingComplete();

    eeUimaEngine.stop();

    // Reset inactivity to original value or remove if it was not set
    if (sessionTimeoutOverride != null) {
      System.setProperty("SessionTimeoutOverride", sessionTimeoutOverride);
    } else {
      System.clearProperty("SessionTimeoutOverride");
    }
  }

  /**
   * Tests handling of ResourceInitializationException that happens in a collocated primitive
   * 
   * @throws Exception
   */
  @Test
  public void testDeployAggregateServiceWithFailingCollocatedComponent() throws Exception {
    System.out
            .println("-------------- testDeployAggregateServiceWithFailingCollocatedComponent -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    try {
      deployService(eeUimaEngine, relativePath
              + "/Deploy_AggregateWithFailingCollocatedDelegate.xml");
    } catch (ResourceInitializationException e) {
      // This is expected
    } catch (Exception e) {
      e.printStackTrace();
      fail("Expected ResourceInitializationException Instead Caught:" + e.getClass().getName());
    }
  }
  @Test
  public void testClientProcessTimeoutWithAggregateMultiplier() throws Exception {
    System.out.println("-------------- testClientProcessTimeoutWithAggregateMultiplier -------------");
    addExceptionToignore(org.apache.uima.aae.error.UimaASProcessCasTimeout.class);

    BaseUIMAAsynchronousEngine_impl uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(uimaAsEngine, relativePath + "/Deploy_AggregateMultiplierWithDelay.xml");

    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    appCtx.put(UimaAsynchronousEngine.Timeout, 3000);
    appCtx.put(UimaAsynchronousEngine.CasPoolSize, 1);

    // reduce the cas pool size and reply window
    appCtx.remove(UimaAsynchronousEngine.ShadowCasPoolSize);
    appCtx.put(UimaAsynchronousEngine.ShadowCasPoolSize, Integer.valueOf(1));
    
    
    initialize(uimaAsEngine, appCtx);
    waitUntilInitialized();
    
    for( int i=0; i < 2; i++ ) {
      CAS cas = uimaAsEngine.getCAS();
      cas.setDocumentText("Some Text");
      uimaAsEngine.sendCAS(cas);  // will timeout after 5 secs
      uimaAsEngine.collectionProcessingComplete();  // the CPC should not
      // be sent to a service until the timeout occurs.
    }
    uimaAsEngine.stop();
  }

  @Test
  public void testDeployAggregateService() throws Exception {
    System.out.println("-------------- testDeployAggregateService -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    
    
 //   System.setProperty("BrokerURL", "tcp::/localhost:61616");

    
    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotator.xml");
    
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
//   		Map<String, Object> appCtx = buildContext("tcp://localhost:61616",
            "TopLevelTaeQueue");
    appCtx.put(UimaAsynchronousEngine.Timeout, 0);
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 0);
    
    addExceptionToignore(org.apache.uima.aae.error.UimaEEServiceException.class); 
    
//    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
    runTest(appCtx, eeUimaEngine, "tcp://localhost:61616", "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }
  
  @Test
  public void testDeployAggregateServiceWithDoctype() throws Exception {
    System.out.println("-------------- testDeployAggregateServiceWithDoctype -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");

    try {
        deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithDOCTYPE.xml");
    } catch( Exception e ) {
    	if ( e.getMessage() != null && 
    			e.getMessage().indexOf("disallow-doctype-decl") > 0 ) {
    		System.out.println("---- Detected expected error during parsing of a deployment descriptor - SUCCESS ----");
        	return; // success - detected expected error
    	}
    }
    fail("This test should have failed with SAXParseException - Instead no error was detected");
  }
  
  @Test
  public void testAggregateTypePriorities() throws Exception {
    System.out.println("-------------- testAggregateTypePriorities -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotator.xml");
    
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
//   		Map<String, Object> appCtx = buildContext("tcp://localhost:61616",
            "TopLevelTaeQueue");
    appCtx.put(UimaAsynchronousEngine.Timeout, 1000);
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 0);
    initialize(eeUimaEngine, appCtx);

    waitUntilInitialized();
    
    ProcessingResourceMetaData meta = eeUimaEngine.getMetaData();
    TypePriorityList[] pl = meta.getTypePriorities().getPriorityLists();
    for( TypePriorityList tp : pl ) {
    	String[] typeArray = tp.getTypes();
    	Assert.assertEquals(true, typeArray[0].equals("uima.cas.TOP"));
    	Assert.assertEquals(true, typeArray[1].equals("uima.tcas.Annotation"));
     }
    eeUimaEngine.stop();
  }
  /**
   * Sends total of 10 CASes to async aggregate configured to process 2 CASes at a time.
   * The inner NoOp annotator is configured to sleep for 5 seconds. The client should
   * be receiving 2 ACKs simultaneously confirming that the aggregate is processing 2 
   * input CASes at the same time.
   * 
   * @throws Exception
   */
  @Test
  public void testDeployAggregateServiceWithScaledInnerNoOp() throws Exception {
	    System.out.println("-------------- testDeployAggregateServiceWithScaledInnerNoOp -------------");
	    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
	    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");
	    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithScaledInnerNoOp.xml");
	    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
	            "TopLevelTaeQueue");
	    appCtx.put(UimaAsynchronousEngine.Timeout, 0);
	    appCtx.put(UimaAsynchronousEngine.CasPoolSize, 5);
	    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 0);
	    
	    addExceptionToignore(org.apache.uima.aae.error.UimaEEServiceException.class); 
	    
	    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
	            10, PROCESS_LATCH);
	  }
  @Test
  public void testDeployAggregateServiceWithDelegateTimeoutAndContinueOnError() throws Exception {
    System.out.println("-------------- testDeployAggregateServiceWithDelegateTimeoutAndContinueOnError -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithLongDelay.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithDelegateTimeoutAndContinueOnError.xml");

    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),"TopLevelTaeQueue");
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }
  
  @Test
  public void testScaledSyncAggregateProcess() throws Exception {
    System.out.println("-------------- testScaledSyncAggregateProcess -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_ScaledPrimitiveAggregateAnnotator.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            5, PROCESS_LATCH);
    System.out.println(eeUimaEngine.getPerformanceReport());
  }

  @Test
  public void testAggregateWithFailedRemoteDelegate() throws Exception {
    System.out.println("-------------- testAggregateWithFailedRemoteDelegate -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithExceptionOn5thCAS.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithFailedRemoteDelegate.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
  }


  @Test
  public void testCMAggregateClientStopRequest() throws Exception {
    System.out.println("-------------- testCMAggregateClientStopRequest -------------");
    final BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_CMAggregateWithCollocated1MillionDocsCM.xml");
    superRef = this;

    Thread t = new Thread() {
      public void run() {
        try {
          // Wait for some CASes to return from the service
          while (superRef.getNumberOfCASesProcessed() == 0) {
            // No reply received yet so wait for 1 second and
            // check again
            synchronized (this) {
              this.wait(1000); // wait for 1 sec
            }
          }
          // The client received at least one reply, wait
          // at this point the top level service should show a connection error
          synchronized (this) {
            // wait for 3 seconds before stopping
            this.wait(5000);
          }
          eeUimaEngine.stopProducingCases();
        } catch (Exception e) {
          e.printStackTrace();
          fail(e.getMessage());
        }
      }
    };
    t.start();
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testCMAggregateClientStopRequest2() throws Exception {
    System.out.println("-------------- testCMAggregateClientStopRequest2 -------------");
    final BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplierWith1MillionDocs.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplierWith1MillionDocs.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_CMAggregateWithRemote1MillionDocsCM.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_CMAggregateWithRemote1MillionDocsCM.xml");
    superRef = this;
    Thread t = new Thread() {
      public void run() {
        try {
          // Wait for some CASes to return from the service
          while (superRef.getNumberOfCASesProcessed() == 0) {
            // No reply received yet so wait for 1 second and
            // check again
            synchronized (this) {
              this.wait(1000); // wait for 1 sec
            }
          }
          // The client received at least one reply, wait
          // at this point the top level service should show a connection error
          synchronized (this) {
            // wait for 3 seconds before stopping
            this.wait(3000);
          }
          eeUimaEngine.stopProducingCases();
        } catch (Exception e) {
          e.printStackTrace();
          fail(e.getMessage());
        }
      }
    };
    t.start();
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testAggregateCMWithFailedRemoteDelegate() throws Exception {
    System.out.println("-------------- testAggregateCMWithFailedRemoteDelegate -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithExceptionOn5thCAS.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateCMWithFailedRemoteDelegate.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
  }

  @Test
  public void testAggregateCMWithFailedCollocatedDelegate() throws Exception {
    System.out.println("-------------- testAggregateCMWithFailedCollocatedDelegate -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateCMWithFailedCollocatedDelegate.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
  }

  @Test
  public void testComplexAggregateCMWithFailedCollocatedDelegate() throws Exception {
    System.out
            .println("-------------- testComplexAggregateCMWithFailedCollocatedDelegate -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath
            + "/Deploy_ComplexAggregateWithFailingInnerAggregateCM.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
  }

  @Test
  public void testAggregateCMWithRemoteCMAndFailedRemoteDelegate() throws Exception {
    System.out
            .println("-------------- testAggregateCMWithRemoteCMAndFailedRemoteDelegate -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithExceptionOn5thCAS.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplierWith1MillionDocs.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateCMWithRemoteCMAndFailedRemoteDelegate.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
  }

  /**
   * Tests a simple Aggregate with one remote Delegate and collocated Cas Multiplier
   * 
   * @throws Exception
   */
  @Test
  public void testDeployAggregateServiceWithBrokerPlaceholder() throws Exception {
    System.out
            .println("-------------- testDeployAggregateServiceWithBrokerPlaceholder -------------");
    final BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");

    try {
      Thread t = new Thread() {
        public void run() {
          BrokerService bs = null;
          try {
            // at this point the top level service should show a connection error
            synchronized (this) {
              this.wait(5000); // wait for 5 secs
            }
            // Create a new broker that runs a different port that the rest of testcases
            bs = setupSecondaryBroker(false);
            System.setProperty("AggregateBroker", bs.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());
            System.setProperty("NoOpBroker", bs.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());
            deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorUsingPlaceholder.xml");
            deployService(eeUimaEngine, relativePath
                    + "/Deploy_AggregateAnnotatorUsingPlaceholder.xml");
            // Start the uima AS client. It connects to the top level service and sends
            // 10 messages
            runTest(null, eeUimaEngine, System.getProperty("AggregateBroker"), "TopLevelTaeQueue",
                    1, PROCESS_LATCH);
          } catch (InterruptedException e) {
          } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
          } finally {
            if (bs != null) {
              try {
                bs.stop();
                bs.waitUntilStopped();
               
              } catch (Exception e) {
                e.printStackTrace();
              }
            }
          }
        }
      };
      t.start();
      t.join();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /**
   * Tests missing broker on service startup. The service listener on input queue recovers from this
   * by silently attempting reconnect at 5 second intervals. The test first launches the service,
   * than spins a thread where a new broker is created after 5 seconds, and finally the uima as
   * client is started. The test shows initial connection failure and when the broker becomes
   * available the connection is established and messages begin to flow from the client to the
   * service and back.
   * 
   * @throws Exception
   */
  @Test
  public void testDelayedBrokerWithAggregateService() throws Exception {
    System.out.println("-------------- testDelayedBrokerWithAggregateService -------------");
    final BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    BrokerService bs = null;
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    // set up and start secondary broker. It is started only to detect an open port so that we
    // define SecondaryBrokerURL property. This property is used to resolve a placeholder
    // in the aggregate descriptor. Once the property is set we shutdown the secondary broker to
    // test aggregate recovery from missing broker. Hopefully the same port is still open when
    // the test starts the secondary broker for the second time. 
    bs = setupSecondaryBroker(false);
    System.setProperty("SecondaryBrokerURL",bs.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());
    bs.stop();
    
    // wait for the broker to stop
    bs.waitUntilStopped();
    // Deploy aggregate on a secondary broker which was shutdown above. The aggregate should 
    // detect missing broker and silently wait for the broker to come up
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorOnSecondaryBroker.xml");
    try {
      // spin a thread to restart a broker after 5 seconds
      Thread t = new Thread() {
        public void run() {
          BrokerService bs = null;
          try {
            // at this point the top level service should show a connection error
            synchronized (this) {
              this.wait(5000); // wait for 5 secs
            }
            // Create a new broker on port 8119
            bs = setupSecondaryBroker(false);
            bs.waitUntilStarted();
            System.setProperty("SecondaryBrokerURL",bs.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());
            // Start the uima AS client. It connects to the top level service and sends
            // 10 messages
            runTest(null, eeUimaEngine, bs.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "TopLevelTaeQueue", 10,
                    PROCESS_LATCH);
          } catch (InterruptedException e) {
          } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
          } finally {
            if (bs != null) {
              try {
                bs.stop();
                bs.waitUntilStopped();

              } catch (Exception e) {
                e.printStackTrace();
              }
            }
          }
        }
      };
      t.start();
      t.join();
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /**
   * Tests a simple Aggregate with one remote Delegate and collocated Cas Multiplier
   * 
   * @throws Exception
   */
  @Test
  public void testDeployAggregateServiceWithTempReplyQueue() throws Exception {
    System.out.println("-------------- testDeployAggregateServiceWithTempReplyQueue -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateUsingRemoteTempQueue.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);

  }

  /**
   * Tests a simple Aggregate with one remote Delegate and collocated Cas Multiplier
   * 
   * @throws Exception
   */
  @Test
  public void testProcessAggregateServiceWith1000Docs() throws Exception {
    System.out.println("-------------- testProcessAggregateServiceWith1000Docs -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateAnnotatorWithInternalCM1000Docs.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);

  }

  @Test
  public void testProcessAggregateWithInnerAggregateCM() throws Exception {
    System.out.println("-------------- testProcessAggregateWithInnerAggregateCM() -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");
    deployService(eeUimaEngine, relativePath + "/Deploy_ComplexAggregateWithInnerAggregateCM.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }
  @Test
  public void testAggregateWithInnerSynchAggregateCM() throws Exception {
	    System.out.println("-------------- testAggregateWithInnerSynchAggregateCM() -------------");
	    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
	    System.setProperty(JmsConstants.SessionTimeoutOverride, "2500000");
	    deployService(eeUimaEngine, relativePath + "/Deploy_ComplexAggregateWithInnerUimaAggregateCM.xml");
	    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
	            1, PROCESS_LATCH);
	  }

  /**
   * Tests exception thrown in the Uima-AS Client when the Collection Reader is added after the uima
   * ee client is initialized
   * 
   * @throws Exception
   */
  @Test
  public void testExceptionOnPostInitializeCollectionReaderInjection() throws Exception {
    System.out
            .println("-------------- testExceptionOnPostInitializeCollectionReaderInjection -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "PersonTitleAnnotatorQueue");
    initialize(eeUimaEngine, appCtx);
    waitUntilInitialized();
    try {
      // Simulate plugging in a Collection Reader. This should throw
      // ResourceInitializationException since the client code has
      // been already initialized.
      eeUimaEngine.setCollectionReader(null);
    } catch (ResourceInitializationException e) {
      System.out.println("Received Expected Exception:" + ResourceInitializationException.class);
      // Expected
      return;
    } catch (Exception e) {
      fail("Invalid Exception Thrown. Expected:" + ResourceInitializationException.class
              + " Received:" + e.getClass());
    } finally {
      eeUimaEngine.stop();
    }
    fail("Expected" + ResourceInitializationException.class);
  }

  /**
   * Tests the shutdown due to a failure in the Flow Controller while diabling a delegate
   * 
   * @throws Exception
   */
  @Test
  public void testTerminateOnFlowControllerExceptionOnDisable() throws Exception {
    System.out
            .println("-------------- testTerminateOnFlowControllerExceptionOnDisable -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithException.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateWithFlowControllerExceptionOnDisable.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH); 
  }

  /**
   * Tests the shutdown due to a failure in the Flow Controller when initializing
   * 
   * @throws Exception
   */
  @Test
  public void testTerminateOnFlowControllerExceptionOnInitialization() throws Exception {
    System.out
            .println("-------------- testTerminateOnFlowControllerExceptionOnInitialization -------------");

    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    String[] containerIds = new String[2];
    try {
      containerIds[0] = deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      containerIds[1] = deployService(eeUimaEngine, relativePath
              + "/Deploy_AggregateWithFlowControllerExceptionOnInitialization.xml");
      fail("Expected ResourceInitializationException. Instead, the Aggregate Deployed Successfully");
    } catch (ResourceInitializationException e) {
      Exception cause = getCause(e);
      System.out.println("\nExpected Initialization Exception was received:" + cause);
    } catch (Exception e) {
      fail("Expected ResourceInitializationException. Instead Got:" + e.getClass());
    } finally {
      eeUimaEngine.undeploy(containerIds[0]);
      eeUimaEngine.undeploy(containerIds[1]);
    }

  }

  /**
   * Tests the shutdown due to a failure in the Flow Controller when initializing AND have delegates
   * to disable (Jira issue UIMA-1171)
   * 
   * @throws Exception
   */
  @Test
  public void testTerminateOnFlowControllerExceptionOnInitializationWithDisabledDelegates()
          throws Exception {
    System.out
            .println("-------------- testTerminateOnFlowControllerExceptionOnInitializationWithDisabledDelegates -----");

    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    String containerId = null;
    try {
      containerId = deployService(eeUimaEngine, relativePath
              + "/Deploy_AggregateWithFlowControllerExceptionOnInitialization.xml");
      fail("Expected ResourceInitializationException. Instead, the Aggregate Deployed Successfully");
    } catch (ResourceInitializationException e) {
      Exception cause = getCause(e);
      System.out.println("\nExpected Initialization Exception was received - cause: " + cause);
    } catch (Exception e) {
      fail("Expected ResourceInitializationException. Instead Got:" + e.getClass());
    } finally {
      eeUimaEngine.undeploy(containerId);
    }
  }

  /**
   * Deploys a Primitive Uima-AS service and sends 5 CASes to it.
   * 
   * @throws Exception
   */

  @Test
  public void testPrimitiveServiceProcess() throws Exception {
    System.out.println("-------------- testPrimitiveServiceProcess -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "PersonTitleAnnotatorQueue", 5, PROCESS_LATCH);
  }

  /**
   * Deploys a Primitive Uima-AS service and sends 5 CASes to it.
   * 
   * @throws Exception
   */

  @Test
  public void testSyncAggregateProcess() throws Exception {
    System.out.println("-------------- testSyncAggregateProcess -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_MeetingDetectorAggregate.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "MeetingDetectorQueue", 5, PROCESS_LATCH);
  }

  /**
   * Deploys a Primitive Uima-AS service and sends 5 CASes to it.
   * 
   * @throws Exception
   */

  @Test
  public void testPrimitiveServiceProcessPingFailure() throws Exception {
    System.out.println("-------------- testPrimitiveServiceProcessPingFailure -------------");
    // Instantiate Uima-AS Client
    final BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    final String containerID = deployService(eeUimaEngine, relativePath
            + "/Deploy_PersonTitleAnnotator.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "PersonTitleAnnotatorQueue");
    // Set an explicit getMeta (Ping)timeout
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 2000);
    // Set an explicit process timeout so to test the ping on timeout
    appCtx.put(UimaAsynchronousEngine.Timeout, 1000);
    // Spin a thread and wait for awhile before killing the remote service.
    // This will cause the client to timeout waiting for a CAS reply and
    // to send a Ping message to test service availability. The Ping times
    // out and causes the client API to stop.
    new Thread() {
      public void run() {
        Object mux = new Object();
        synchronized (mux) {
          try {
            mux.wait(1000);
            // Undeploy service container
            System.out.println("** About to undeploy PersonTitle service");
            eeUimaEngine.undeploy(containerID);
          } catch (Exception e) {
          }
        }
      }
    }.start();

    super.countPingRetries=true;
    
    try {
      runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
              "PersonTitleAnnotatorQueue", 1000, EXCEPTION_LATCH);
    } catch (RuntimeException e) {
      System.out.println(">>> runtest generated exception: " + e);
      e.printStackTrace(System.out);
    }
    super.countPingRetries=false;
 //   eeUimaEngine.stop();
  }

  /**
   * Tests error handling on delegate timeout. The Delegate is started as remote, the aggregate
   * initializes and the client starts sending CASes. After a short while the client kills the
   * remote delegate. The aggregate receives a CAS timeout and disables the delegate. A timed out
   * CAS is sent to the next delegate in the pipeline. ALL 1000 CASes are returned to the client.
   * 
   * @throws Exception
   */
  @Test
  public void testDelegateTimeoutAndDisable() throws Exception {
    System.out.println("-------------- testDelegateTimeoutAndDisable -------------");
    // Instantiate Uima-AS Client
    final BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    final String containerID = deployService(eeUimaEngine, relativePath
            + "/Deploy_RoomNumberAnnotator.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_MeetingDetectorTAE_RemoteRoomNumberDisable.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "MeetingDetectorTaeQueue");
    // Set an explicit getMeta (Ping)timeout
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 2000);
    // Set an explicit process timeout so to test the ping on timeout
    appCtx.put(UimaAsynchronousEngine.Timeout, 1000);
    // Spin a thread and wait for awhile before killing the remote service.
    // This will cause the client to timeout waiting for a CAS reply and
    // to send a Ping message to test service availability. The Ping times
    // out and causes the client API to stop.

    new Thread() {
      public void run() {
        Object mux = new Object();
        synchronized (mux) {
          try {
            mux.wait(500);
            // Undeploy service container
            eeUimaEngine.undeploy(containerID);
          } catch (Exception e) {
          }
        }
      }
    }.start();

    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "MeetingDetectorTaeQueue", 1000, PROCESS_LATCH);

  }

  /**
   * This test kills a remote Delegate while in the middle of processing 1000 CASes. The CAS timeout
   * error handling disables the delegate and forces ALL CASes from the Pending Reply List to go
   * through Error Handler. The Flow Controller is configured to continueOnError and CASes that
   * timed out are allowed to continue to the next delegate. ALL 1000 CASes are accounted for in the
   * NoOp Annotator that is last in the flow.
   * 
   * @throws Exception
   */
  @Test
  public void testDisableDelegateOnTimeoutWithCM() throws Exception {
    System.out.println("-------------- testDisableDelegateOnTimeoutWithCM -------------");
    // Instantiate Uima-AS Client
    final BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    final String containerID = deployService(eeUimaEngine, relativePath
            + "/Deploy_RoomNumberAnnotator.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_MeetingDetectorTAEWithCM_RemoteRoomNumberDisable.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "MeetingDetectorTaeQueue");
    // Set an explicit getMeta (Ping)timeout
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 2000);
    // Set an explicit process timeout so to test the ping on timeout
    appCtx.put(UimaAsynchronousEngine.Timeout, 1000);
    // Spin a thread and wait for awhile before killing the remote service.
    // This will cause the client to timeout waiting for a CAS reply and
    // to send a Ping message to test service availability. The Ping times
    // out and causes the client API to stop.

    new Thread() {
      public void run() {
        Object mux = new Object();
        synchronized (mux) {
          try {
            mux.wait(300);
            // Undeploy service container
            eeUimaEngine.undeploy(containerID);
          } catch (Exception e) {
          }
        }
      }
    }.start();

    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "MeetingDetectorTaeQueue", 1, PROCESS_LATCH);

  }

  /**
   * Tests Uima-AS client ability to test sendAndReceive in multiple/concurrent threads It spawns 4
   * thread each sending 100 CASes to a Primitive Uima-AS service
   * 
   * @throws Exception
   */
  @Test
  public void testSynchCallProcessWithMultipleThreads() throws Exception {
    System.out.println("-------------- testSynchCallProcessWithMultipleThreads -------------");
    int howManyCASesPerRunningThread = 100;
    int howManyRunningThreads = 4;
    runTestWithMultipleThreads(relativePath + "/Deploy_PersonTitleAnnotator.xml",
            "PersonTitleAnnotatorQueue", howManyCASesPerRunningThread, howManyRunningThreads, 0, 0);
  }

  /**
   * 
   * @throws Exception
   */
  @Test
  public void testPrimitiveProcessCallWithLongDelay() throws Exception {
    System.out.println("-------------- testPrimitiveProcessCallWithLongDelay -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithLongDelay.xml");
    // We expect 18000ms to be spent in process method
    super.setExpectedProcessTime(6000);

    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueueLongDelay");
    appCtx.remove(UimaAsynchronousEngine.ReplyWindow);
    appCtx.put(UimaAsynchronousEngine.ReplyWindow, 1);
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueueLongDelay", 4, PROCESS_LATCH, true);
  }

  /**
   * Tests time spent in process CAS. The CAS is sent to three remote delegates each with a delay of
   * 6000ms in the process method. The aggregate is expected to sum up the time spent in each
   * annotator process method. The final sum is returned to the client (the test) and compared
   * against expected 18000ms. The test actually allows for 20ms margin to account for any overhead
   * (garbage collecting, slow cpu, etc)
   * 
   * @throws Exception
   */
  @Test
  public void testAggregateProcessCallWithLongDelay() throws Exception {

    System.out.println("-------------- testAggregateProcessCallWithLongDelay -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Services each with 6000ms delay in process()
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorAWithLongDelay.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorBWithLongDelay.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorCWithLongDelay.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithLongDelay.xml");
    // We expect 18000ms to be spent in process method
    super.setExpectedProcessTime(18000);
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    appCtx.remove(UimaAsynchronousEngine.ReplyWindow);
    // make sure we only send 1 CAS at a time
    appCtx.put(UimaAsynchronousEngine.ReplyWindow, 1);
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue", 1, PROCESS_LATCH, true);
  }

  /**
   * Tests Aggregate configuration where the Cas Multiplier delegate is the last delegate in the
   * Aggregate's pipeline
   * 
   * @throws Exception
   */
  @Test
  public void testAggregateProcessCallWithLastCM() throws Exception {
    System.out.println("-------------- testAggregateProcessCallWithLastCM -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy Uima-AS Primitive Services each with 6000ms delay in process()
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithLastCM.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH, true);
  }

  /**
   * Tests shutdown while running with multiple/concurrent threads The Annotator throws an exception
   * and the Aggregate error handling is setup to terminate on the first error.
   * 
   * @throws Exception
   */

  @Test
  public void testTimeoutInSynchCallProcessWithMultipleThreads() throws Exception {
    System.out
            .println("-------------- testTimeoutInSynchCallProcessWithMultipleThreads -------------");
    int howManyCASesPerRunningThread = 2;
    int howManyRunningThreads = 4;
    int processTimeout = 2000;
    int getMetaTimeout = 500;
    runTestWithMultipleThreads(relativePath + "/Deploy_NoOpAnnotatorWithLongDelay.xml",
            "NoOpAnnotatorQueueLongDelay", howManyCASesPerRunningThread, howManyRunningThreads,
            processTimeout, getMetaTimeout);
  }

  /**
   * Tests shutdown while running with multiple/concurrent threads The Annotator throws an exception
   * and the Aggregate error handling is setup to terminate on the first error.
   * 
   * @throws Exception
   */

  @Test
  public void testTimeoutFailureInSynchCallProcessWithMultipleThreads() throws Exception {
    System.out
            .println("-------------- testTimeoutFailureInSynchCallProcessWithMultipleThreads -------------");
    int howManyCASesPerRunningThread = 1000;
    int howManyRunningThreads = 4;
    int processTimeout = 2000;
    int getMetaTimeout = 500;
    runTestWithMultipleThreads(relativePath + "/Deploy_NoOpAnnotator.xml", "NoOpAnnotatorQueue",
            howManyCASesPerRunningThread, howManyRunningThreads, 2000, 1000, true);

  }

  /**
   * Tests a parallel flow in the Uima-AS aggregate.
   * 
   * @throws Exception
   */
  @Test
  public void testProcessWithParallelFlow() throws Exception {
    System.out.println("-------------- testProcessWithParallelFlow -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator2.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithParallelFlow.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  /**
   * Tests ability to disable one delegate in parallel flow and continue
   * 
   * @throws Exception
   */
  @Test
  public void testDisableDelegateInParallelFlow() throws Exception {
    System.out.println("-------------- testDisableDelegateInParallelFlow -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator2.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithParallelFlow.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }
  /**
   * This test launches four remote delegates and an aggregate. Three of the delegates:
   * SlowNoOp1, SlowNoOp2, and NoOp are processing CASes in parallel. SlowNoOp1 and SlowNoOp2
   * are configured with a very long delay that exceeds timeout values specified in the aggregate 
   * deployment descriptor. The SlowNoOp1 times out 3 times and is disabled,
   * SlowNoOp2 times out 4 times and is disabled. The test tests ability
   * of an aggregate to recover from the timeouts and subsequent disable and carry
   * on processing with remaining delegates.  
   * 
   * @throws Exception
   */
  @Test
  public void testMutlipleDelegateTimeoutsInParallelFlows() throws Exception {
    System.out.println("-------------- testTimeoutDelegateInParallelFlows -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator2.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_SlowNoOpAnnotator1.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_SlowNoOpAnnotator2.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithSlowParallelDelegates.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    appCtx.put(UimaAsynchronousEngine.Timeout, 30000);

    runTest(appCtx, eeUimaEngine, null, null, 1, PROCESS_LATCH);
  }

  /**
   * 
   * @throws Exception
   */
  @Test
  public void testTimeoutDelegateInParallelFlows() throws Exception {
    System.out.println("-------------- testTimeoutDelegateInParallelFlows -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithDelay.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithLongDelay.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator2.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator3.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithParallelFlows.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    // Set an explicit process timeout so one of the 1st parallels is disabled but 2nd parallel flow
    // continues.
    appCtx.put(UimaAsynchronousEngine.Timeout, 200000);

    runTest(appCtx, eeUimaEngine, null, null, 1, PROCESS_LATCH);
  }

  /**
   * Tests Timeout logic
   * 
   * @throws Exception
   */
  @Test
  public void testRemoteDelegateTimeout() throws Exception {
    System.out.println("-------------- testRemoteDelegateTimeout -------------");
    System.out.println("The Aggregate sends 2 CASes to the NoOp Annotator which");
    System.out.println("delays each CAS for 6000ms. The timeout is set to 4000ms");
    System.out.println("Two CAS retries are expected");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithLongDelay.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateAnnotatorWithLongDelayDelegate.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    // The Remote NoOp delays each CAS for 6000ms. The Aggregate sends two CASes so adjust
    // client timeout to be just over 12000ms.
    appCtx.put(UimaAsynchronousEngine.Timeout, 19000);

    runTest(appCtx, eeUimaEngine, null, null, 1, PROCESS_LATCH);
  }

  /**
   * Tests Timeout logic
   * 
   * @throws Exception
   */
  @Test
  public void testDisableOnRemoteDelegatePingTimeout() throws Exception {
    System.out.println("-------------- testDisableOnRemoteDelegatePingTimeout -------------");
    System.out.println("The Aggregate sends 2 CASes to the NoOp Annotator which");
    System.out.println("delays each CAS for 6000ms. The timeout is set to 4000ms");
    System.out.println("Two CAS retries are expected");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    String delegateContainerId = deployService(eeUimaEngine, relativePath
            + "/Deploy_NoOpAnnotatorWithLongDelay.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateAnnotatorWithLongDelayDelegate.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    // The Remote NoOp delays each CAS for 6000ms. The Aggregate sends two CASes so adjust
    // client timeout to be just over 12000ms.
    appCtx.put(UimaAsynchronousEngine.Timeout, 13000);
    // Remove container with the remote NoOp delegate so that we can test
    // the CAS Process and Ping timeout.
    eeUimaEngine.undeploy(delegateContainerId);
    // Send the CAS and handle exception
    runTest(appCtx, eeUimaEngine, null, null, 1, EXCEPTION_LATCH);
  }

  @Test
  public void testDeployAggregateWithCollocatedAggregateService() throws Exception {
    System.out
            .println("-------------- testDeployAggregateWithCollocatedAggregateService -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_ComplexAggregate.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            10, PROCESS_LATCH);
  }

  @Test
  public void testProcessWithAggregateUsingCollocatedMultiplier() throws Exception {
    System.out
            .println("-------------- testProcessWithAggregateUsingCollocatedMultiplier -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotator.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testExistanceOfParentCasReferenceIdOnChildFailure() throws Exception {
	  System.out
            .println("-------------- testExistanceOfParentCasReferenceIdOnChildFailure -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithException.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithDelegateFailure.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
    // When a callback is received to handle the exception on the child CAS, the message should
    // contain an CAS id of the parent. If it does the callback handler will set
    // receivedExpectedParentReferenceId = true
    Assert.assertTrue(super.receivedExpectedParentReferenceId);
  }

  @Test
  public void testProcessWithAggregateUsingRemoteMultiplier() throws Exception {
    System.out
            .println("-------------- testProcessWithAggregateUsingRemoteMultiplier -------------");
    System.setProperty("BrokerURL", getMasterConnectorURI(broker));

    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplier.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithRemoteMultiplier.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testParentProcessLast() throws Exception {
    System.out
            .println("-------------- testParentProcessLast -------------");
    System.setProperty("BrokerURL", getMasterConnectorURI(broker));

    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplierWith10Docs_1.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithProcessParentLastCMs.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  /**
   * Starts two remote delegates on one broker and a top level client aggregate on 
   * another. Tests sending Free Cas requests to the appropriate broker. 
   * 
   * @throws Exception
   */
  @Test
  public void testProcessWithAggregateUsingRemoteMultiplierOnSeparateBroker() throws Exception {
    System.out
            .println("-------------- testProcessWithAggregateUsingRemoteMultiplierOnSeparateBroker -------------");
    System.setProperty("activemq.broker.jmx.domain","org.apache.activemq.test");
    BrokerService broker2 = setupSecondaryBroker(true);
    System.setProperty("BrokerURL", broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());

    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplier.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithRemoteMultiplier.xml");
    
    Map<String, Object> appCtx = new HashMap();
    appCtx.put(UimaAsynchronousEngine.ServerUri, broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());
    appCtx.put(UimaAsynchronousEngine.ENDPOINT, "TopLevelTaeQueue");
    appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 0);
    runTest(appCtx, eeUimaEngine, broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(),
            "TopLevelTaeQueue", 1, PROCESS_LATCH);    
    super.cleanBroker(broker2);

    broker2.stop();
    broker2.waitUntilStopped();

  }
  
  
  /**
   * First CM feeds 100 CASes to a "merger" CM that generates one output CAS for every 5 input.
   * Second CM creates unique document text that is checked by the last component. The default FC
   * should let 4 childless CASes through, replacing every 5th by its child.
   * 
   * @throws Exception
   */
  @Test
  public void testProcessWithAggregateUsingCollocatedMerger() throws Exception {
    System.out.println("-------------- testProcessWithAggregateUsingRemoteMerger -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithCollocatedMerger.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testProcessWithAggregateUsingRemoteMerger() throws Exception {
    System.out.println("-------------- testProcessWithAggregateUsingRemoteMerger -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMerger.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithRemoteMerger.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  
  @Test
  public void testClientWithAggregateMultiplier() throws Exception {
    System.out.println("-------------- testClientWithAggregateMultiplier -------------");
    System.setProperty("BrokerURL", broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString());

    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplier.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateMultiplier.xml");

    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    appCtx.remove(UimaAsynchronousEngine.ShadowCasPoolSize);
    appCtx.put(UimaAsynchronousEngine.ShadowCasPoolSize, Integer.valueOf(1));
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue", 1, PROCESS_LATCH);
    
    /*
    broker.stop();
    broker.waitUntilStopped();

    //System.setProperty("activemq.broker.jmx.domain","org.apache.activemq.test");
    //broker2 = setupSecondaryBroker(true);
    broker = createBroker();
    broker.start();
    broker.waitUntilStarted();
*/

    
/*    
    
    appCtx.remove(UimaAsynchronousEngine.ShadowCasPoolSize);
    appCtx.put(UimaAsynchronousEngine.ShadowCasPoolSize, Integer.valueOf(1));
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "TestMultiplierQueue", 1, PROCESS_LATCH);

    
    String burl = broker.getConnectorByName(DEFAULT_BROKER_URL_KEY).getUri().toString();
    Map<String, Object> appCtx = 
    buildContext(burl, "TopLevelTaeQueue");
*/
//    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
  //          "TopLevelTaeQueue");
/*
broker.stop();
broker.waitUntilStopped();

//System.setProperty("activemq.broker.jmx.domain","org.apache.activemq.test");
//broker2 = setupSecondaryBroker(true);
broker = createBroker();
broker.start();
broker.waitUntilStarted();
*/
/*
    // reduce the cas pool size and reply window
    appCtx.remove(UimaAsynchronousEngine.ShadowCasPoolSize);
    appCtx.put(UimaAsynchronousEngine.ShadowCasPoolSize, Integer.valueOf(2));
//    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
    runTest(appCtx, eeUimaEngine,burl,
            "TopLevelTaeQueue", 1, PROCESS_LATCH);
    
    eeUimaEngine.stop();
  */
  }
  @Test
  public void testClientProcessWithRemoteMultiplier() throws Exception {
    System.out.println("-------------- testClientProcessWithRemoteMultiplier -------------");
   
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplier.xml");

    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TestMultiplierQueue");
    
    appCtx.remove(UimaAsynchronousEngine.ShadowCasPoolSize);
    appCtx.put(UimaAsynchronousEngine.ShadowCasPoolSize, Integer.valueOf(1));
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "TestMultiplierQueue", 1, PROCESS_LATCH);
  }

  @Test
  public void testClientProcessWithComplexAggregateRemoteMultiplier() throws Exception {

    System.out
            .println("-------------- testClientProcessWithComplexAggregateRemoteMultiplier -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplierWith10Docs_1.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_CasMultiplierAggregateWithRemoteCasMultiplier.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testProcessWithAggregateUsing2RemoteMultipliers() throws Exception {
    System.out
            .println("-------------- testProcessWithAggregateUsing2RemoteMultipliers -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplierWith10Docs_1.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplierWith10Docs_2.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWith2RemoteMultipliers.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testProcessWithAggregateUsing2CollocatedMultipliers() throws Exception {
    System.out
            .println("-------------- testProcessWithAggregateUsing2CollocatedMultipliers -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWith2Multipliers.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testProcessAggregateWithInnerCMAggregate() throws Exception {
    System.out.println("-------------- testProcessAggregateWithInnerCMAggregate -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_TopAggregateWithInnerAggregateCM.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testProcessAggregateWithInnerAggregateDelegateInitFailure() throws Exception {
    System.out.println("-------------- testProcessAggregateWithInnerAggregateDelegateInitFailure -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    try {
      deployService(eeUimaEngine, relativePath + "/Deploy_TopAggregateWithInnerAggregateDelegateInitFailure.xml");
      fail("Expected ResourceInitializationException But The Deployment Succeeded Instead");
    } catch( ResourceInitializationException e) {
      eeUimaEngine.stop();
    }
  }

  @Test
  public void testComplexDeployment() throws Exception {
    System.out.println("-------------- testComplexDeployment -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy replicated services for the inner remote aggregate CM
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    // Deploy an instance of a remote aggregate CM containing a collocated Cas Multiplier
    // CM --> Replicated Remote Primitive --> NoOp CC
    deployService(eeUimaEngine, relativePath + "/Deploy_CMAggregateWithCollocatedCM.xml");
    // Deploy top level Aggregate Cas Multiplier with 2 collocated Cas Multipliers
    // CM1 --> CM2 --> Remote AggregateCM --> Candidate Answer --> CC
    deployService(eeUimaEngine, relativePath + "/Deploy_TopLevelComplexAggregateCM.xml");

    runTest2(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue", 10, PROCESS_LATCH);
  }

  @Test
  public void testTypesystemMergeWithMultiplier() throws Exception {
    System.out.println("-------------- testTypesystemMergeWithMultiplier -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithMergedTypes.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testStopAggregateWithRemoteMultiplier() throws Exception {
    System.out.println("-------------- testStopAggregateWithRemoteMultiplier -------------");
    
    System.setProperty("BrokerURL", getMasterConnectorURI(broker));
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplier.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithExceptionOn5thCAS.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithRemoteMultiplier.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
  }

  @Test
  public void testCancelProcessAggregateWithCollocatedMultiplier() throws Exception {
    System.out
            .println("-------------- testCancelProcessAggregateWithCollocatedMultiplier -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_ComplexAggregateWith1MillionDocs.xml");
    // Spin a thread to cancel Process after 20 seconds
    spinShutdownThread(eeUimaEngine, 20000);
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);
  }

  @Test
  public void testCancelProcessAggregateWithRemoteMultiplier() throws Exception {
    System.out.println("-------------- testStopAggregateWithRemoteMultiplier -------------");
    System.setProperty("BrokerURL", getMasterConnectorURI(broker));
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_RemoteCasMultiplierWith1MillionDocs.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithRemoteMultiplier.xml");
    // Spin a thread to cancel Process after 20 seconds
    spinShutdownThread(eeUimaEngine, 20000);
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH);// EXCEPTION_LATCH);
  }

  /**
   * Test correct reply from the service when its process method fails. Deploys the Primitive
   * Service ( NoOp Annotator) that is configured to throw an exception on every CAS. The expected
   * behavior is for the Primitive Service to return a reply with an Exception. This code blocks on
   * a Count Down Latch, until the exception is returned from the service. When the exception is
   * received the latch is opened indicating success.
   * 
   * @throws Exception
   */
  @Test
  public void testPrimitiveServiceResponseOnException() throws Exception {
    System.out.println("-------------- testPrimitiveServiceResponseOnException -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy remote service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithException.xml");
    // Deploy Uima-AS Primitive Service
    // Initialize and run the Test. Wait for a completion and cleanup resources.
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueue", 1, EXCEPTION_LATCH);
  }

  @Test
  public void testProcessParallelFlowWithDelegateFailure() throws Exception {
    System.out.println("-------------- testProcessParallelFlowWithDelegateFailure -------------");
    // Create Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy remote service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithException.xml");
    // Deploy remote service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator2.xml");
    // Deploy top level aggregate service
    deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateWithParallelFlowTerminateOnDelegateFailure.xml");

    // Initialize and run the Test. Wait for a completion and cleanup resources.
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH); // PC_LATCH);
  }

  /**
   * Tests that the thresholdAction is taken when thresholdCount errors occur in the last
   * thresholdWindow CASes. Aggregate has two annotators, first fails with increasing frequency (on
   * CASes 10 19 27 34 40 45 49 52 54) and is disabled after 3 errors in a window of 7 (49,52,54)
   * Second annotator counts the CASes that reach it and verifies that it sees all but the 9
   * failures. It throws an exception if the first is disabled after too many or too few errors.
   * 
   * @throws Exception
   */
  @Test
  public void testErrorThresholdWindow() throws Exception {
    System.out.println("-------------- testErrorThresholdWindow -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithThresholdWindow.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue");
    // Set an explicit CPC timeout as exceptions thrown in the 2nd annotator's CPC don't reach the
    // client.
    appCtx.put(UimaAsynchronousEngine.CpcTimeout, 20000);
    runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
            "TopLevelTaeQueue", 1, PROCESS_LATCH); // PC_LATCH);
  }

  @Test
  public void testProcessParallelFlowWithDelegateDisable() throws Exception {
    System.out.println("-------------- testProcessParallelFlowWithDelegateDisable -------------");
    // Create Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithException.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_SimpleAnnotator.xml");
    deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateWithParallelFlowDisableOnDelegateFailure.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, PROCESS_LATCH); // PC_LATCH);
  }

  @Test
  public void testPrimitiveShutdownOnTooManyErrors() throws Exception {
    System.out.println("-------------- testPrimitiveShutdownOnTooManyErrors -------------");
    // Create Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Deploy remote service
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithException.xml");
    // Deploy top level aggregate service
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotator.xml");
    // Initialize and run the Test. Wait for a completion and cleanup resources.
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
  }


  /**
   * Tests exception thrown in the Uima-AS Client when the Collection Reader is added after the uima
   * ee client is initialized
   * 
   * @throws Exception
   */

  @Test
  public void testCollectionReader() throws Exception {
    System.out.println("-------------- testCollectionReader -------------");
    // Instantiate Uima-AS Client
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_PersonTitleAnnotator.xml");
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "PersonTitleAnnotatorQueue");
    // reduce the cas pool size and reply window
    appCtx.remove(UimaAsynchronousEngine.CasPoolSize);
    appCtx.put(UimaAsynchronousEngine.CasPoolSize, Integer.valueOf(2));
    appCtx.remove(UimaAsynchronousEngine.ReplyWindow);
    appCtx.put(UimaAsynchronousEngine.ReplyWindow, 1);

    // set the collection reader
    String filename = super
            .getFilepathFromClassloader("descriptors/collection_reader/ExtendedTestFileSystemCollectionReader.xml");
    if (filename == null) {
      fail("Unable to find file:" + "descriptors/collection_reader/ExtendedTestFileSystemCollectionReader.xml"
              + "in classloader");
    }
    File collectionReaderDescriptor = new File(filename);
    CollectionReaderDescription collectionReaderDescription = UIMAFramework.getXMLParser()
            .parseCollectionReaderDescription(new XMLInputSource(collectionReaderDescriptor));
    CollectionReader collectionReader = UIMAFramework
            .produceCollectionReader(collectionReaderDescription);
    eeUimaEngine.setCollectionReader(collectionReader);
    initialize(eeUimaEngine, appCtx);
    waitUntilInitialized();
    runCrTest(eeUimaEngine, 7);
    synchronized (this) {
      wait(50);
    }
    eeUimaEngine.stop();
  }

  @Test
  public void testAsynchronousTerminate() throws Exception {
    System.out.println("-------------- testAsynchronousTerminate -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
    "TopLevelTaeQueue");
    try {
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithDelay.xml");
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator2.xml");
      deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithParallelFlow.xml");
      initialize(eeUimaEngine, appCtx);
      // Wait until the top level service returns its metadata
      waitUntilInitialized();
    } catch( Exception e) {
			throw e;
    }


    CAS cas = eeUimaEngine.getCAS();
    System.out.println(" Sending CAS to kick off aggregate w/colocated CasMultiplier");
    eeUimaEngine.sendCAS(cas);

    System.out.println(" Waiting 1 seconds");
    Thread.sleep(1000);

    System.out.println(" Trying to stop service");
    eeUimaEngine.stop();
    System.out.println(" stop() returned!");
    
  }

  @Test
  public void testCallbackListenerOnFailure() throws Exception {
    System.out.println("-------------- testCallbackListenerOnFailure -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithException.xml");

    Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
            "NoOpAnnotatorQueue");
    // Register special callback listener. This listener will receive
    // an exception with the Cas Reference id.
    TestListener listener = new TestListener(this);
    eeUimaEngine.addStatusCallbackListener(listener);
    initialize(eeUimaEngine, appCtx);
    // Wait until the top level service returns its metadata
    waitUntilInitialized();
    CAS cas = eeUimaEngine.getCAS();

    // Send request out and save Cas Reference id
    String casReferenceId = eeUimaEngine.sendCAS(cas);
    // Spin a callback listener thread
    Thread t = new Thread(listener);
    t.start();
    // Wait for reply CAS. This method blocks
    String cRefId = listener.getCasReferenceId();

    try {
      // Test if received Cas Reference Id matches the id of the CAS sent out
      if (!cRefId.equals(casReferenceId)) {
        fail("Received Invalid Cas Reference Id. Expected:" + casReferenceId + " Received: "
                + cRefId);
      } else {
        System.out.println("Received Expected Cas Identifier:" + casReferenceId);
      }
    } finally {
      // Stop callback listener thread
      listener.doStop();
      eeUimaEngine.stop();
    }
  }
  @Test
  public void testCauseOfInitializationFailure() throws Exception {
	    System.out.println("-------------- testCauseOfInitializationFailure -------------");
	    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
	    try {
	      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotatorWithInitException.xml");
	      Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
	              "NoOpAnnotatorQueue");
	      exceptionCountLatch = new CountDownLatch(1);
	      initialize(eeUimaEngine, appCtx);
	      fail("Expected ResourceInitializationException. Instead, the Aggregate Reports Successfull Initialization");
	    } catch (ResourceInitializationException e) {
	      Exception cause = getCause(e);
	      if ( cause != null && (cause instanceof FileNotFoundException) ) {
		      System.out.println("Expected FileNotFoundException was received");
	      } else {
		      fail("Expected FileNotFoundException NOT received as a cause of failure. Instead Got:" + cause);
	      }
	    } catch (Exception e) {
	      fail("Expected ResourceInitializationException. Instead Got:" + e.getClass());
	    } finally {
	      eeUimaEngine.stop();
	    }
	  }

  @Test
  public void testTerminateOnInitializationFailure() throws Exception {
    System.out.println("-------------- testTerminateOnInitializationFailure -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    try {
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithParallelFlow.xml");
      Map<String, Object> appCtx = buildContext(String.valueOf(getMasterConnectorURI(broker)),
              "TopLevelTaeQueue");
      exceptionCountLatch = new CountDownLatch(1);
      initialize(eeUimaEngine, appCtx);
      fail("Expected ResourceInitializationException. Instead, the Aggregate Reports Successfull Initialization");
    } catch (ResourceInitializationException e) {
      Exception cause = getCause(e);
      System.out.println("Expected Initialization Exception was received:" + cause);
    } catch (Exception e) {
      fail("Expected ResourceInitializationException. Instead Got:" + e.getClass());
    } finally {
      eeUimaEngine.stop();
    }
  }

  /**
   * Tests shutdown due to delegate broker missing. The Aggregate is configured to retry getMeta 3
   * times and continue. The client times out after 20 seconds and forces the shutdown. NOTE: The
   * Spring listener tries to recover JMS connection on failure. In this test a Listener to remote
   * delegate cannot be established due to a missing broker. The Listener is setup to retry every 60
   * seconds. After failure, the listener goes to sleep for 60 seconds and tries again. This results
   * in a 60 second delay at the end of this test.
   * 
   * @throws Exception
   */
  @Test
  public void testTerminateOnInitializationFailureWithDelegateBrokerMissing() throws Exception {
    System.out
            .println("-------------- testTerminateOnInitializationFailureWithDelegateBrokerMissing -------------");
    System.out
            .println("---------------------- The Uima Client Times Out After 20 seconds --------------------------");
    System.out
            .println("-- The test requires 1 minute to complete due to 60 second delay in Spring Listener ----");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    try {
      // Deploy remote service
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      // Deploy top level aggregate that communicates with the remote via Http Tunnelling
      deployService(eeUimaEngine, relativePath
              + "/Deploy_AggregateAnnotatorTerminateOnDelegateBadBrokerURL.xml");
      // Initialize and run the Test. Wait for a completion and cleanup resources.
      Map<String, Object> appCtx = new HashMap();
      appCtx.put(UimaAsynchronousEngine.ServerUri, String.valueOf(getMasterConnectorURI(broker)));
      appCtx.put(UimaAsynchronousEngine.ENDPOINT, "TopLevelTaeQueue");
      appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 20000);
      runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
              "TopLevelTaeQueue", 1, EXCEPTION_LATCH);
      fail("Expected ResourceInitializationException. Instead, the Aggregate Reports Successfull Initialization");
    } catch (ResourceInitializationException e) {
      Exception cause = getCause(e);
      System.out.println("Expected Initialization Exception was received:" + cause);
    } catch (Exception e) {
      fail("Expected ResourceInitializationException. Instead Got:" + e.getClass());
    } finally {
      eeUimaEngine.stop();
    }
  }

  /**
   * Tests shutdown due to delegate broker missing. The Aggregate is configured to terminate on
   * getMeta timeout.
   * 
   * @throws Exception
   */
  @Test
  public void testTerminateOnInitializationFailureWithAggregateForcedShutdown() throws Exception {
    System.out
            .println("-------------- testTerminateOnInitializationFailureWithAggregateForcedShutdown -------------");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    // Initialize and run the Test. Wait for a completion and cleanup resources.
    try {
      // Deploy remote service
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      // Deploy top level aggregate that communicates with the remote via Http Tunnelling
      deployService(eeUimaEngine, relativePath
              + "/Deploy_AggregateAnnotatorWithHttpDelegateNoRetries.xml");
      runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
              "TopLevelTaeQueue", 10, EXCEPTION_LATCH);
      fail("Expected ResourceInitializationException. Instead, the Aggregate Reports Successfull Initialization");
    } catch (ResourceInitializationException e) {
      Exception cause = getCause(e);
      System.out.println("Expected Initialization Exception was received:" + cause);
    } catch (Exception e) {
      fail("Expected ResourceInitializationException. Instead Got:" + e.getClass());
    } finally {
      eeUimaEngine.stop();
    }

  }


  /**
   * Tests shutdown due to delegate broker missing. The Aggregate is configured to retry getMeta 3
   * times and continue. The client times out after 20 seconds and forces the shutdown. NOTE: The
   * Spring listener tries to recover JMS connection on failure. In this test a Listener to remote
   * delegate cannot be established due to a missing broker. The Listener is setup to retry every 60
   * seconds. After failure, the listener goes to sleep for 60 seconds and tries again. This results
   * in a 60 second delay at the end of this test.
   * 
   * @throws Exception
   */
  @Test
  public void testDisableOnInitializationFailureWithDelegateBrokerMissing() throws Exception {
    System.out
            .println("-------------- testDisableOnInitializationFailureWithDelegateBrokerMissing() -------------");
    System.out
            .println("---------------------- The Uima Client Times Out After 20 seconds --------------------------");
    System.out
            .println("-- The test requires 1 minute to complete due to 60 second delay in Spring Listener ----");
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    try {
      // Deploy remote service
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      // Deploy top level aggregate that communicates with the remote via Http Tunnelling
      deployService(eeUimaEngine, relativePath + "/Deploy_AggregateAnnotatorWithHttpDelegate.xml");
      // Initialize and run the Test. Wait for a completion and cleanup resources.
      Map<String, Object> appCtx = new HashMap();
      appCtx.put(UimaAsynchronousEngine.ServerUri, String.valueOf(getMasterConnectorURI(broker)));
      appCtx.put(UimaAsynchronousEngine.ENDPOINT, "TopLevelTaeQueue");
      appCtx.put(UimaAsynchronousEngine.GetMetaTimeout, 20000);
      runTest(appCtx, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
              "TopLevelTaeQueue", 1, PROCESS_LATCH);
    } catch (Exception e) {
      fail("Expected Success. Instead Received Exception:" + e.getClass());
    } finally {
      eeUimaEngine.stop();
    }
  }

 
  /**
   * This tests some of the error handling. Each annotator writes a file and throws an exception.
   * After the CAS is processed the presence/absence of certain files indicates success or failure.
   * The first annotator fails and lets the CAS proceed, so should write only one file. The second
   * annotator fails and is retried 2 times, and doesn't let the CAS proceed, so should write 3
   * files. The third annotator should not see the CAS, so should not write any files
   * 
   * @throws Exception
   */
  @Test
  public void testContinueOnRetryFailure() throws Exception {
    System.out.println("-------------- testContinueOnRetryFailure -------------");
    File tempDir = new File("target/temp");
    deleteAllFiles(tempDir);
    try {
        tempDir.mkdir();
    } catch( Exception e) {
    	e.printStackTrace();
    	throw e;
    }
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    deployService(eeUimaEngine, relativePath + "/Deploy_WriterAnnotatorA.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_WriterAnnotatorB.xml");
    deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithContinueOnRetryFailures.xml");
    runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)), "TopLevelTaeQueue",
            1, EXCEPTION_LATCH);
    if (!(new File(tempDir, "WriterAnnotatorB.3")).exists()
            || (new File(tempDir, "WriterAnnotatorB.4")).exists()) {
      fail("Second annotator should have run 3 times");
    }
    if ((new File(tempDir, "WriterAnnotatorC.1")).exists()) {
      fail("Third annotator should not have seen CAS");
    }
  }

//  public void testCatchExtraThreads() throws Exception {
//    Thread.sleep(24 * 60 * 60 * 1000);  // sleep for one day
//  }
  
  @Test
  public void testDeployAgainAndAgain() throws Exception {
    System.out.println("-------------- testDeployAgainAndAgain -------------");
                                                                                          // in the
                                                                                          // loop,
                                                                                          // no
                                                                                          // change.

    for (int num = 1; num <= 50; num++) {
      BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl(); // here or
      System.out.println("\nRunning iteration " + num);
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator.xml");
      deployService(eeUimaEngine, relativePath + "/Deploy_NoOpAnnotator2.xml");
      deployService(eeUimaEngine, relativePath + "/Deploy_AggregateWithParallelFlow.xml");
      runTest(null, eeUimaEngine, String.valueOf(getMasterConnectorURI(broker)),
              "TopLevelTaeQueue", 1, PROCESS_LATCH);
    }
  }
  
  public void testMultipleASClients() throws Exception  {
	    System.out.println("-------------- testMultipleSyncClientsWithMultipleBrokers -------------");
	    
	    class RunnableClient implements Runnable {
	    	String brokerURL;
	    	BaseTestSupport testSupport;
          BaseUIMAAsynchronousEngine_impl uimaAsEngine;
          String serviceEndpoint;
          
          
          RunnableClient(BaseTestSupport testSupport, String brokerURL,String serviceEndpoint) {
	    		this.brokerURL = brokerURL;
	    		this.testSupport = testSupport;
	    		this.serviceEndpoint = serviceEndpoint;
	    		uimaAsEngine = new BaseUIMAAsynchronousEngine_impl();
	    	}
	    	public BaseUIMAAsynchronousEngine_impl getUimaAsClient() {
	    		return uimaAsEngine;
	    	}
	    	public void initialize() throws Exception {
	    		@SuppressWarnings("unchecked")
			  Map<String, Object> appCtx = buildContext(brokerURL, serviceEndpoint);
	    		appCtx.put(UimaAsynchronousEngine.Timeout, 1100);
		  	  appCtx.put(UimaAsynchronousEngine.CpcTimeout, 1100);
		  	  testSupport.initialize(getUimaAsClient(), appCtx);
		  	  waitUntilInitialized();
		  	  
	    	}
			public void run() {
				try {
					initialize();
					System.out.println("Thread:"+Thread.currentThread().getId()+" Completed GetMeta() broker:"+brokerURL);
				} catch( Exception e) {
					e.printStackTrace();
				} finally {
					try {
				        uimaAsEngine.stop();
					} catch( Exception e) {
						e.printStackTrace();
					}
				}

			}
	    	
	    }
	    
	    ExecutorService executor = Executors.newCachedThreadPool();
      String serviceId1;
      String serviceId2;

	    //	change broker URl in system properties
	    System.setProperty("BrokerURL", getMasterConnectorURI(broker).toString());
	    
	    RunnableClient client1 = 
	    		new RunnableClient(this, getMasterConnectorURI(broker), "NoOpAnnotatorQueue");
	    BaseUIMAAsynchronousEngine_impl engine = client1.getUimaAsClient();
	    serviceId1 = deployService(engine, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");

	    final BrokerService broker2 = setupSecondaryBroker(true);
	    //	change broker URl in system properties
	    System.setProperty("BrokerURL", broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString());
	    
	    RunnableClient client2 = 
	    		new RunnableClient(this, "failover:tcp://f5n633:51514,tcp://f12n1133:51514","NoOpAnnotatorQueue");//broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");
	    		//new RunnableClient(this, "failover:ssl://f5n6:51514,ssl://f12n11:51514","NoOpAnnotatorQueue");//broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString(), "NoOpAnnotatorQueue");
	    BaseUIMAAsynchronousEngine_impl engine2 = client2.getUimaAsClient();
//	    serviceId2 = deployService(engine2, relativePath + "/Deploy_NoOpAnnotatorWithPlaceholder.xml");

	    
	    for( int x = 0; x < 100; x++) {
		    List<Future<?>> list1 = new ArrayList<Future<?>>();
		    List<Future<?>> list2 = new ArrayList<Future<?>>();
		    String b;
		  /*
		    if ( x % 2 == 0 ) {
		    	b = getMasterConnectorURI(broker);
		    } else {
		    	b = "failover:ssl://f5n6:51514,ssl://f12n11:51514";//broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString();
		    }
		    */
		    List<Future<?>> list = new ArrayList<Future<?>>();
		    for (int i = 0; i < 20; i++) {
		    	  if ( i % 2 == 0 ) {
				    	b = getMasterConnectorURI(broker);
				    	list = list1;
				    } else {
				    	b = "failover:tcp://f5n633:51514,tcp://f12n1133:51514?maxReconnectAttempts=2&timeout=300&transport.maxReconnectAttempts=2&transport.timeout=300&startupMaxReconnectAttempts=1";//broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString();
//				    	b = "failover:ssl://f5n6:51514,ssl://f12n11:51514?maxReconnectAttempts=2&timeout=300&transport.maxReconnectAttempts=2&transport.timeout=300&startupMaxReconnectAttempts=1";//broker2.getConnectorByName(DEFAULT_BROKER_URL_KEY_2).getUri().toString();
				    	list = list2;
				    }
			    RunnableClient client = 
			    		new RunnableClient(this, b, "NoOpAnnotatorQueue");
			    list.add(executor.submit(client));;
		    }
		    /*
		    
			for (int i = 0; i < 10; i++) {
		    	try {
			    	list.get(i).get();//1, TimeUnit.SECONDS);
		    	} catch( Exception e) {
		    		e.printStackTrace();
			    	list.get(i).cancel(true);
		    	}
		    }
		    */

	    Worker worker1 = new Worker(list1);
	    Worker worker2 = new Worker(list1);
	    Thread t1 = new Thread(worker1);
	    Thread t2 = new Thread(worker2);
	    t1.start();
	    t2.start();
	    
	    t1.join();
	    t2.join();
	    
	    list.clear();
	    
	    }
	//    engine2.undeploy(serviceId2);
	    engine.undeploy(serviceId1);
	    
	    //engine2.stop();
	    executor.shutdownNow();
	    while( !executor.isShutdown() ) {
	    	synchronized(broker) {
	    		broker.wait(100);
	    	}
	    }
	    broker2.stop();
	    broker2.waitUntilStopped();
	    //broker.stop();
	    //broker.waitUntilStopped();
	    //System.out.println("Done");
	}

  private Exception getCause(Throwable e) {
    Exception cause = (Exception) e;
    while (cause.getCause() != null) {
      cause = (Exception) cause.getCause();
    }
    return cause;
  }

  /**
   * This tests GetMeta retries. It deploys a simple Aggregate service that contains a collocated
   * Primitive service and a Primitive remote. The Primitive remote is simulated in this code. The
   * code starts a listener where the Aggregate sends GetMeta requests. The listener responds to the
   * Aggregate with its metadata only when an expected number of GetMeta retries is met. If the
   * Aggregate fails to send expected number of GetMeta requests, the listener will not adjust its
   * CountDownLatch and will cause this test to hang.
   * 
   * @throws Exception
   */

  public void getMetaRetry() throws Exception {
    getMetaCountLatch = new CountDownLatch(MaxGetMetaRetryCount);
    Connection connection = getConnection();
    connection.start();

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    ActiveMQDestination destination = (ActiveMQDestination) session
            .createQueue(primitiveServiceQueue1);
    ActiveMQMessageConsumer consumer = (ActiveMQMessageConsumer) session
            .createConsumer(destination);
    consumer.setMessageListener(new MessageListener() {
      public void onMessage(Message aMessage) {
        try {
          if (isMetaRequest(aMessage)) {
            // Reply with metadata when retry count reaches defined threshold
            if (getMetaRequestCount > 0 && getMetaRequestCount % MaxGetMetaRetryCount == 0) {
              JmsMessageContext msgContext = new JmsMessageContext(aMessage, primitiveServiceQueue1);
              JmsOutputChannel outputChannel = new JmsOutputChannel();
              outputChannel.setServiceInputEndpoint(primitiveServiceQueue1);
              outputChannel.setServerURI(getBrokerUri());
              Endpoint endpoint = msgContext.getEndpoint();
              outputChannel.sendReply(getPrimitiveMetadata1(PrimitiveDescriptor1), endpoint, true);
            }
            getMetaRequestCount++;
            getMetaCountLatch.countDown(); // Count down to unblock the thread
          }
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
    });

    consumer.start();
    BaseUIMAAsynchronousEngine_impl eeUimaEngine = new BaseUIMAAsynchronousEngine_impl();
    String containerId = deployService(eeUimaEngine, relativePath
            + "/Deploy_AggregateAnnotator.xml");

    Map<String, Object> appCtx = new HashMap();
    appCtx.put(UimaAsynchronousEngine.ServerUri, String.valueOf(getMasterConnectorURI(broker)));
    appCtx.put(UimaAsynchronousEngine.ENDPOINT, "TopLevelTaeQueue");
    appCtx.put(UimaAsynchronousEngine.CasPoolSize, Integer.valueOf(4));
    appCtx.put(UimaAsynchronousEngine.ReplyWindow, 15);
    appCtx.put(UimaAsynchronousEngine.Timeout, 0);
    initialize(eeUimaEngine, appCtx);

    System.out
            .println("TestBroker.testGetMetaRetry()-Blocking On GetMeta Latch. Awaiting GetMeta Requests");

    /*********************************************************************************/
    /**** This Code Will Block Until Expected Number Of GetMeta Requests Arrive ******/
    getMetaCountLatch.await();
    /*********************************************************************************/

    consumer.stop();
    connection.stop();
    eeUimaEngine.undeploy(containerId);
    eeUimaEngine.stop();
  }

  public ProcessingResourceMetaData getPrimitiveMetadata1(String aDescriptor) throws Exception {
    ResourceSpecifier resourceSpecifier = UimaClassFactory.produceResourceSpecifier(aDescriptor);
    return ((AnalysisEngineDescription) resourceSpecifier).getAnalysisEngineMetaData();
  }

  private static boolean deleteAllFiles(File directory) {
    if (directory.isDirectory()) {
      String[] files = directory.list();
      for (int i = 0; i < files.length; i++) {
        deleteAllFiles(new File(directory, files[i]));
      }
    }
    // Now have an empty directory or simple file
    return directory.delete();
  }

  private static class TestListener extends UimaAsBaseCallbackListener implements Runnable {
    private String casReferenceId = null;

    private Object monitor = new Object();

    public TestListener(TestUimaASExtended aTester) {
    }

    public void collectionProcessComplete(EntityProcessStatus arg0) {
      // TODO Auto-generated method stub

    }

    public void onBeforeMessageSend(UimaASProcessStatus status) {
      System.out.println("TestListener Received onBeforeMessageSend Notification with Cas:"
              + status.getCasReferenceId());
    }

    public void entityProcessComplete(CAS aCAS, EntityProcessStatus aProcessStatus) {
      if (aProcessStatus.isException()) {
        if (aProcessStatus instanceof UimaASProcessStatus) {
          casReferenceId = ((UimaASProcessStatus) aProcessStatus).getCasReferenceId();
          if (casReferenceId != null) {
            synchronized (monitor) {
              monitor.notifyAll();
            }
          }
        }
      }
    }

    public void initializationComplete(EntityProcessStatus arg0) {
      // TODO Auto-generated method stub

    }

    public String getCasReferenceId() {
      synchronized (monitor) {
        while (casReferenceId == null) {
          try {
            monitor.wait();
          } catch (InterruptedException e) {
          }
        }
      }
      return casReferenceId;
    }

    public void doStop() {
    }

    public void run() {
      System.out.println("Stopping TestListener Callback Listener Thread");
    }
  }
  
  private class Worker implements Runnable {

	  
	List<Future<?>> list = new ArrayList<Future<?>>();
	
	public Worker(List<Future<?>> list ) {
		this.list = list;
	}
	@Override
	public void run() {
		for (int i = 0; i < list.size(); i++) {
	    	try {
		    	list.get(i).get();//1, TimeUnit.SECONDS);
	    	} catch( Exception e) {
	    		e.printStackTrace();
		    	list.get(i).cancel(true);
	    	}
	    }

	}
	  
  }
  class StartBrokerTask extends TimerTask {
	  BrokerService broker = null;
	  TestUimaASExtended testSuiteRef = null;
	  StartBrokerTask(BrokerService broker, TestUimaASExtended testSuiteRef) {
		this.broker = broker;  
		this.testSuiteRef = testSuiteRef;
	  }
	@Override
	public void run() {
		try {
			broker = testSuiteRef.setupSecondaryBroker(true);
	        broker.waitUntilStarted();
	      	System.out.println("..... Restarted broker ............................");

		} catch( Exception e ) {
			throw new RuntimeException(e);
			
		}
		
	}

	  
  }
}
