/*
 * 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.geode.cache.query.cq.dunit;

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.logging.log4j.Logger;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.cache.AttributesFactory;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionFactory;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqAttributesFactory;
import org.apache.geode.cache.query.CqAttributesMutator;
import org.apache.geode.cache.query.CqClosedException;
import org.apache.geode.cache.query.CqExistsException;
import org.apache.geode.cache.query.CqListener;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.RegionNotFoundException;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.data.Portfolio;
import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.cache30.CertifiableTestCacheListener;
import org.apache.geode.cache30.ClientServerTestCase;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.DistributedTombstoneOperation;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

/**
 * This class tests the ContinuousQuery mechanism in GemFire. It does so by creating a cache server
 * with a cache and a pre-defined region and a data loader. The client creates the same region and
 * attaches the connection pool.
 */
@Category({ClientSubscriptionTest.class})
@SuppressWarnings({"serial", "Convert2MethodRef"})
public class CqQueryDUnitTest extends JUnit4CacheTestCase {
  private static final Logger logger = LogService.getLogger();
  /**
   * The port on which the cache server was started in this VM
   */
  private static int bridgeServerPort;

  private static final int port = 0;
  protected static int port2 = 0;

  public static final int noTest = -1;

  public final String[] regions = new String[] {"regionA", "regionB"};

  private static final int CREATE = 0;
  private static final int UPDATE = 1;
  private static final int DESTROY = 2;
  private static final int INVALIDATE = 3;
  private static final int CLOSE = 4;
  private static final int REGION_CLEAR = 5;
  private static final int REGION_INVALIDATE = 6;

  public static final String KEY = "key-";

  private static final String WAIT_PROPERTY = "CqQueryTest.maxWaitTime";

  private static final int WAIT_DEFAULT = (20 * 1000);

  private static final long MAX_TIME = Integer.getInteger(WAIT_PROPERTY, WAIT_DEFAULT);

  public final String[] cqs = new String[] {
      // 0 - Test for ">"
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + " p where p.ID > 0",

      // 1 - Test for "=" and "and".
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0]
          + " p where p.ID = 2 and p.status='active'",

      // 2 - Test for "<" and "and".
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[1]
          + " p where p.ID < 5 and p.status='active'",

      // FOLLOWING CQS ARE NOT TESTED WITH VALUES; THEY ARE USED TO TEST PARSING LOGIC WITHIN CQ.
      // 3
      "SELECT * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + " ;",
      // 4
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0],
      // 5
      "import org.apache.geode.cache.\"query\".data.Portfolio; " + "SELECT ALL * FROM " + SEPARATOR
          + "root" + SEPARATOR
          + regions[0] + " TYPE Portfolio",
      // 6
      "import org.apache.geode.cache.\"query\".data.Portfolio; " + "SELECT ALL * FROM " + SEPARATOR
          + "root" + SEPARATOR
          + regions[0] + " p TYPE Portfolio",
      // 7
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[1]
          + " p where p.ID < 5 and p.status='active';",
      // 8
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + "  ;",
      // 9
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0]
          + " p where p.description = NULL",
      // 10
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0]
          + " p where p.ID > 0 and p.status='active'",
      // 11 test for region 0
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0] + " p where p.ID > 0",
      // 12 test for region 1
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[1] + " p where p.ID > 0",};

  public final String[] prCqs = new String[] {
      // 0 - Test for ">"
      "SELECT ALL * FROM " + SEPARATOR + regions[0] + " p where p.ID > 0",

      // 1 - Test for "=" and "and".
      "SELECT ALL * FROM " + SEPARATOR + regions[0] + " p where p.ID = 2 and p.status='active'"};

  private final String[] invalidCQs = new String[] {
      // Test for ">"
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + "invalidRegion p where p.ID > 0"};


  private final String[] shortTypeCQs = new String[] {
      // 11 - Test for "short" number type
      "SELECT ALL * FROM " + SEPARATOR + "root" + SEPARATOR + regions[0]
          + " p where p.shortID IN SET(1,2,3,4,5)"};

  @Override
  public final void postSetUp() throws Exception {
    // avoid IllegalStateException from HandShake by connecting all vms tor
    // system before creating connection pools
    getSystem();
    Invoke.invokeInEveryVM(new SerializableRunnable("getSystem") {
      @Override
      public void run() {
        getSystem();
      }
    });
    postSetUpCqQueryDUnitTest();
  }

  protected void postSetUpCqQueryDUnitTest() throws Exception {}

  /* Returns Cache Server Port */
  public static int getCacheServerPort() {
    return bridgeServerPort;
  }

  /* Create Cache Server */
  public void createServer(VM server) {
    createServer(server, AvailablePortHelper.getRandomAvailableTCPPort());
  }

  public void createServer(VM server, final int p) {
    createServer(server, p, false);
  }

  public void createServer(VM server, final int thePort, final boolean eviction) {
    createServer(server, thePort, eviction, DataPolicy.REPLICATE);
  }

  public void createServer(VM server, final int thePort, final boolean eviction,
      final DataPolicy dataPolicy) {
    server.invoke(() -> {
      logger.info("### Create Cache Server. ###");
      AttributesFactory factory = new AttributesFactory();
      factory.setScope(Scope.DISTRIBUTED_ACK);
      factory.setDataPolicy(dataPolicy);

      // setting the eviction attributes.
      if (eviction) {
        EvictionAttributes evictAttrs =
            EvictionAttributes.createLRUEntryAttributes(100000, EvictionAction.OVERFLOW_TO_DISK);
        factory.setEvictionAttributes(evictAttrs);
      }

      for (String region : regions) {
        createRegion(region, factory.createRegionAttributes());
        if (getRootRegion("root").getSubregion(region).isEmpty()) {
          logger.info("### CreateServer: Region is empty ###");
        }
      }

      startBridgeServer(thePort, true);
    });
  }

  public void createServerOnly(VM server, final int thePort) {
    server.invoke(() -> {
      logger.info("### Create Cache Server. ###");
      startBridgeServer(thePort, true);
    });
  }

  public void createPartitionRegion(final VM server, final String[] regionNames) {
    SerializableRunnable createRegion = new CacheSerializableRunnable("Create Region") {

      @Override
      public void run2() throws CacheException {
        RegionFactory rf = getCache().createRegionFactory(RegionShortcut.PARTITION);
        for (String regionName : regionNames) {
          rf.create(regionName);
        }
      }
    };

    server.invoke(createRegion);
  }

  private void createReplicateRegionWithLocalDestroy(final VM server, final String[] regionNames) {
    SerializableRunnable createRegion = new CacheSerializableRunnable("Create Region") {

      @Override
      public void run2() throws CacheException {
        RegionFactory rf =
            getCache().createRegionFactory(RegionShortcut.REPLICATE).setEvictionAttributes(
                EvictionAttributes.createLIFOEntryAttributes(10, EvictionAction.LOCAL_DESTROY));
        for (String regionName : regionNames) {
          rf.create(regionName);
        }
      }
    };

    server.invoke(createRegion);
  }


  /* Close Cache Server */
  public void closeServer(VM server) {
    server.invoke(new SerializableRunnable("Close CacheServer") {
      @Override
      public void run() {
        logger.info("### Close CacheServer. ###");
        stopBridgeServer(getCache());
      }
    });

  }

  private void crashServer(VM server) {
    server.invoke(new SerializableRunnable("Crash CacheServer") {
      @Override
      public void run() {
        org.apache.geode.cache.client.internal.ConnectionImpl.setTEST_DURABLE_CLIENT_CRASH(true);
        logger.info("### Crashing CacheServer. ###");
        stopBridgeServer(getCache());
      }
    });
    Wait.pause(2 * 1000);
  }

  private void closeCrashServer(VM server) {
    server.invoke(new SerializableRunnable("Close CacheServer") {
      @Override
      public void run() {
        org.apache.geode.cache.client.internal.ConnectionImpl.setTEST_DURABLE_CLIENT_CRASH(false);
        logger.info("### Crashing CacheServer. ###");
        stopBridgeServer(getCache());
      }
    });
    Wait.pause(2 * 1000);
  }

  /* Create Client */
  public void createClient(VM client, final int serverPort, final String serverHost) {
    int[] serverPorts = new int[] {serverPort};
    createClient(client, serverPorts, serverHost, null);
  }

  /* Create Client */
  public void createClient(VM client, final int[] serverPorts, final String serverHost,
      final String redundancyLevel) {
    client.invoke(() -> {
      logger.info("### Create Client. ###");

      // Initialize CQ Service.
      getCache().getQueryService();

      AttributesFactory regionFactory = new AttributesFactory();
      regionFactory.setScope(Scope.LOCAL);
      if (redundancyLevel != null) {
        ClientServerTestCase.configureConnectionPool(regionFactory, serverHost, serverPorts, true,
            Integer.parseInt(redundancyLevel), -1, null);
      } else {
        ClientServerTestCase.configureConnectionPool(regionFactory, serverHost, serverPorts, true,
            -1, -1, null);
      }
      for (String region : regions) {
        createRegion(region, regionFactory.createRegionAttributes());
        logger.info("### Successfully Created Region on Client :" + region);
      }
    });
  }

  /* Create Local Region */
  public void createLocalRegion(VM client, final int[] serverPorts, final String serverHost,
      final String redundancyLevel, final String[] regionNames) {
    SerializableRunnable createQService = new CacheSerializableRunnable("Create Local Region") {
      @Override
      public void run2() throws CacheException {
        logger.info("### Create Local Region. ###");
        AttributesFactory af = new AttributesFactory();
        af.setScope(Scope.LOCAL);

        if (redundancyLevel != null) {
          ClientServerTestCase.configureConnectionPool(af, serverHost, serverPorts, true,
              Integer.parseInt(redundancyLevel), -1, null);
        } else {
          ClientServerTestCase.configureConnectionPool(af, serverHost, serverPorts, true, -1, -1,
              null);
        }

        RegionFactory rf = getCache().createRegionFactory(af.create());
        for (int i = 0; i < regionNames.length; i++) {
          rf.create(regionNames[i]);
          logger.info("### Successfully Created Region on Client :" + regions[i]);
        }
      }
    };

    client.invoke(createQService);
  }

  public void createClientWith2Pools(VM client, final int[] serverPorts1, final int[] serverPorts2,
      final String serverHost, final String redundancyLevel) {
    client.invoke(() -> {
      logger.info("### Create Client. ###");

      // Initialize CQ Service.
      getCache().getQueryService();

      AttributesFactory regionFactory0 = new AttributesFactory();
      AttributesFactory regionFactory1 = new AttributesFactory();
      regionFactory0.setScope(Scope.LOCAL);
      regionFactory1.setScope(Scope.LOCAL);
      if (redundancyLevel != null) {
        ClientServerTestCase.configureConnectionPoolWithNameAndFactory(regionFactory0, serverHost,
            serverPorts1, true,
            Integer.parseInt(redundancyLevel),
            -1, null, "testPoolA", PoolManager.createFactory(), -1, -1, -2,
            -1);
        ClientServerTestCase.configureConnectionPoolWithNameAndFactory(regionFactory1, serverHost,
            serverPorts2, true,
            Integer.parseInt(redundancyLevel),
            -1, null, "testPoolB", PoolManager.createFactory(), -1, -1, -2,
            -1);
      } else {
        ClientServerTestCase.configureConnectionPoolWithNameAndFactory(regionFactory0, serverHost,
            serverPorts1, true,
            -1,
            -1, null, "testPoolA", PoolManager.createFactory(), -1, -1, -2,
            -1);
        ClientServerTestCase.configureConnectionPoolWithNameAndFactory(regionFactory1, serverHost,
            serverPorts2, true,
            -1,
            -1, null, "testPoolB", PoolManager.createFactory(), -1, -1, -2,
            -1);
      }
      createRegion(regions[0], regionFactory0.create());
      createRegion(regions[1], regionFactory1.create());
      logger.info("### Successfully Created Region on Client :" + regions[0]);
      logger.info("### Successfully Created Region on Client :" + regions[1]);

    });
  }


  /* Close Client */
  public void closeClient(VM client) {
    SerializableRunnable closeCQService = new CacheSerializableRunnable("Close Client") {
      @Override
      public void run2() throws CacheException {
        logger.info("### Close Client. ###");
        try {
          ((DefaultQueryService) getCache().getQueryService()).closeCqService();
        } catch (Exception ex) {
          logger.info("### Failed to get CqService during ClientClose() ###");
        }

      }
    };

    client.invoke(closeCQService);
    Wait.pause(2 * 1000);
  }


  /* Create/Init values */
  public void createValues(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          region1.put(KEY + i, new Portfolio(i));
        }
        logger.info("### Number of Entries in Region :" + region1.keySet().size());
      }
    });
  }

  /* Create/Init values */
  public void createValuesWithTime(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          Portfolio portfolio = new Portfolio(i);
          portfolio.createTime = System.currentTimeMillis();
          region1.put(KEY + i, portfolio);
        }
        logger.info("### Number of Entries in Region :" + region1.keySet().size());
      }
    });
  }

  private void createValuesWithShort(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          Portfolio portfolio = new Portfolio(i);
          portfolio.shortID = new Short("" + i);
          region1.put(KEY + i, portfolio);
        }
        logger.info("### Number of Entries in Region :" + region1.keySet().size());
      }
    });
  }

  private void createValuesAsPrimitives(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          switch (i % 5) {
            case 0:
              region1.put("key" + i, "seeded");
              break;
            case 1:
              region1.put("key" + i, "seeding");
              break;
            case 2:
              region1.put("key" + i, (double) i);
              break;
            case 3:
              region1.put("key" + i, i);
              break;
            case 4:
              region1.put("key" + i, new Portfolio(i));
              break;
            default:
              region1.put("key" + i, i);
              break;

          }
        }
        logger.info("### Number of Entries in Region :" + region1.keySet().size());
      }
    });
  }

  public void updateValuesAsPrimitives(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          switch (i % 5) {
            case 0:
              region1.put("key" + i, "seeding");
              break;
            case 1:
              region1.put("key" + i, "seeded");
              break;
            case 2:
              region1.put("key" + i, i);
              break;
            case 3:
              region1.put("key" + i, new Portfolio(i));
              break;
            case 4:
              region1.put("key" + i, (double) i);
              break;
            default:
              region1.put("key" + i, i);
              break;

          }
        }
        logger.info("### Number of Entries in Region :" + region1.keySet().size());
      }
    });
  }

  public void createValuesAsPortfolios(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          region1.put("key" + i, new Portfolio(i));
        }
        logger.info("### Number of Entries in Region :" + region1.keySet().size());
      }
    });
  }

  private void createIndex(VM vm, final String indexName, final String indexedExpression,
      final String regionPath) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        try {
          QueryService qs = getCache().getQueryService();
          qs.createIndex(indexName, indexedExpression, regionPath);
        } catch (Exception e) {
          throw new CacheException(e) {};
        }
      }
    });
  }


  /* delete values */
  public void deleteValues(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Delete values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          region1.destroy(KEY + i);
        }
        logger.info("### Number of Entries In Region after Delete :" + region1.keySet().size());
      }

    });
  }

  /**
   * support for invalidating values.
   */
  public void invalidateValues(VM vm, final String regionName, final int size) {
    vm.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(regionName);
        for (int i = 1; i <= size; i++) {
          region1.invalidate(KEY + i);
        }
        logger.info("### Number of Entries In Region after Delete :" + region1.keySet().size());
      }

    });
  }

  /* Register CQs */
  public void createCQ(VM vm, final String cqName, final String queryStr) {
    createCQ(vm, cqName, queryStr, false);
  }

  public void createCQ(VM vm, final String cqName, final String queryStr,
      final boolean isBridgeMemberTest) {
    vm.invoke(() -> {

      logger.info("### Create CQ. ###" + cqName);
      // Get CQ Service.
      QueryService cqService = getCache().getQueryService();

      // Create CQ Attributes.
      CqAttributesFactory cqf = new CqAttributesFactory();
      CqListener[] cqListeners = {new CqQueryTestListener(LogWriterUtils.getLogWriter())};

      cqf.initCqListeners(cqListeners);
      CqAttributes cqa = cqf.create();

      // Create CQ.
      CqQuery cq1 = cqService.newCq(cqName, queryStr, cqa);
      assertThat(cq1.getState().isStopped()).describedAs("newCq() state mismatch").isTrue();

    });
  }

  /* Register CQs with no name, execute, and close */
  public void createAndExecCQNoName(VM vm, final String queryStr) {
    vm.invoke(() -> {
      logger.info("### DEBUG CREATE CQ START ####");
      logger.info("### Create CQ with no name. ###");
      String cqName = null;
      QueryService cqService = getCache().getQueryService();

      SelectResults cqResults;
      for (int i = 0; i < 20; ++i) {
        // Create CQ Attributes.
        CqAttributesFactory cqf = new CqAttributesFactory();
        CqListener[] cqListeners = {new CqQueryTestListener(LogWriterUtils.getLogWriter())};

        cqf.initCqListeners(cqListeners);
        CqAttributes cqa = cqf.create();

        // Create CQ with no name and execute with initial results.

        CqQuery cq1 = cqService.newCq(queryStr, cqa);
        ((CqQueryTestListener) cqListeners[0]).cqName = cq1.getName();

        if (cq1 == null) {
          logger.info("Failed to get CqQuery object for CQ with no name.");
        } else {
          cqName = cq1.getName();
          logger.info("Created CQ with no name, generated CQ name: "
              + cqName + " CQ state:" + cq1.getState());
          assertThat(cq1.getState().isStopped()).describedAs("Create CQ with no name illegal state")
              .isTrue();
        }
        if (i % 2 == 0) {
          cqResults = cq1.executeWithInitialResults();

          logger.info("initial result size = " + cqResults.size());
          logger.info("CQ state after execute with initial results = " + cq1.getState());
          assertThat(cq1.getState().isRunning())
              .describedAs("executeWithInitialResults() state mismatch").isTrue();
        } else {
          try {
            cq1.execute();
          } catch (Exception ex) {
            logger.info("CQService is :" + cqService);
            ex.printStackTrace();
            fail("Failed to execute CQ " + cqName + " . " + ex.getMessage());
          }
          logger.info("CQ state after execute = " + cq1.getState());
          assertThat(cq1.getState().isRunning()).describedAs("execute() state mismatch").isTrue();
        }

        // Close the CQ
        try {
          cq1.close();
        } catch (Exception ex) {
          logger.info("CqService is :" + cqService, ex);
          fail("Failed to close CQ " + cqName + " . " + ex.getMessage());
        }
        assertThat(cq1.getState().isClosed()).describedAs("closeCq() state mismatch").isTrue();
      }
    });
  }

  public void executeCQ(VM vm, final String cqName, final boolean initialResults,
      String expectedErr) {
    executeCQ(vm, cqName, initialResults, noTest, expectedErr);
  }

  /**
   * Execute/register CQ as running.
   *
   * @param initialResults true if initialResults are requested
   * @param expectedResultsSize if >= 0, validate results against this size
   * @param expectedErr if not null, an error we expect
   */
  private void executeCQ(VM vm, final String cqName, final boolean initialResults,
      final int expectedResultsSize, final String expectedErr) {
    vm.invoke(() -> {
      if (expectedErr != null) {
        getCache().getLogger()
            .info("<ExpectedException action=add>" + expectedErr + "</ExpectedException>");
      }
      try {
        QueryService cqService = getCache().getQueryService();
        CqQuery cq1 = cqService.getCq(cqName);
        assertThat(cq1).isNotNull();
        assertThat(cq1.getState().isStopped()).describedAs("newCq() state mismatch").isTrue();

        if (initialResults) {
          SelectResults cqResults;
          cqResults = cq1.executeWithInitialResults();
          assertThat(cq1.getState().isRunning())
              .describedAs("executeWithInitialResults() state mismatch").isTrue();
          if (expectedResultsSize >= 0) {
            assertThat(cqResults.size()).describedAs("unexpected results size")
                .isEqualTo(expectedResultsSize);
          }
        } else {
          cq1.execute();
          assertThat(cq1.getState().isRunning()).describedAs("execute() state mismatch").isTrue();
        }
      } finally {
        if (expectedErr != null) {
          getCache().getLogger()
              .info("<ExpectedException action=remove>" + expectedErr + "</ExpectedException>");
        }
      }
    });
  }

  /* Stop/pause CQ */
  public void stopCQ(VM vm, final String cqName) {
    vm.invoke(() -> {
      logger.info("### Stop CQ. ###" + cqName);
      QueryService cqService = getCache().getQueryService();

      CqQuery cq1 = cqService.getCq(cqName);
      cq1.stop();

      assertThat(cq1.getState().isStopped()).describedAs("Stop CQ state mismatch").isTrue();
    });
  }

  // Stop and execute CQ repeatedly
  /* Stop/pause CQ */
  private void stopExecCQ(VM vm) {
    vm.invoke(() -> {
      logger.info("### Stop and Exec CQ. ###" + "testCQStopExecute_0");
      QueryService cqService = getCache().getQueryService();
      CqQuery cq1 = cqService.getCq("testCQStopExecute_0");

      for (int i = 0; i < 20; ++i) {

        cq1.stop();

        assertThat(cq1.getState().isStopped()).describedAs("Stop CQ state mismatch, count = " + i)
            .isTrue();
        logger.info("After stop in Stop and Execute loop, ran successfully, loop count: " + i);
        logger.info("CQ state: " + cq1.getState());

        cq1.execute();

        assertThat(cq1.getState().isRunning())
            .describedAs("Execute CQ state mismatch, count = " + i).isTrue();
        logger.info("After execute in Stop and Execute loop, ran successfully, loop count: " + i);
        logger.info("CQ state: " + cq1.getState());
      }
    });
  }


  /* UnRegister CQs */
  public void closeCQ(VM vm, final String cqName) {
    vm.invoke(() -> {
      // Get CQ Service.
      QueryService cqService = getCache().getQueryService();

      // Close CQ.
      CqQuery cq1 = cqService.getCq(cqName);
      cq1.close();

      assertThat(cq1.getState().isClosed()).describedAs("Close CQ state mismatch").isTrue();
    });
  }

  /* Register CQs */
  private void registerInterestListCQ(VM vm, final String regionName) {
    vm.invoke(() -> {
      Region region = getRootRegion().getSubregion(regionName);
      region.getAttributesMutator()
          .addCacheListener(new CertifiableTestCacheListener());

      List list = new ArrayList();
      for (int i = 1; i <= 10; i++) {
        list.add(KEY + i);
      }
      region.registerInterest(list);
    });
  }

  // helps test case where executeIR is called multiple times as well as after close
  private void executeAndCloseAndExecuteIRMultipleTimes(VM vm,
      final String queryStr) {
    vm.invoke(() -> {
      QueryService cqService = getCache().getQueryService();
      CqAttributesFactory cqf = new CqAttributesFactory();
      CqListener[] cqListeners = {new CqQueryTestListener(LogWriterUtils.getLogWriter())};

      cqf.initCqListeners(cqListeners);
      CqAttributes cqa = cqf.create();

      CqQuery cq1 = cqService.newCq("testCQResultSet_0", queryStr, cqa);
      assertThat(cq1.getState().isStopped()).describedAs("newCq() state mismatch").isTrue();

      cq1.executeWithInitialResults();
      try {
        cq1.executeWithInitialResults();
      } catch (IllegalStateException e) {
        // expected
      }
      cq1.close();

      try {
        cq1.executeWithInitialResults();
      } catch (CqClosedException e) {
        // expected
        return;
      }
      fail("should have received cqClosedException");
    });
  }


  /* Validate CQ Count */
  public void validateCQCount(VM vm, final int cqCnt) {
    vm.invoke(() -> {
      // Get CQ Service.
      QueryService cqService = getCache().getQueryService();

      int numCqs = cqService.getCqs().length;
      assertThat(numCqs).describedAs("Number of cqs mismatch.").isEqualTo(cqCnt);
    });
  }


  /**
   * Throws AssertionError if the CQ can be found or if any other error occurs
   */
  private void failIfCQExists(VM vm, final String cqName) {
    vm.invoke(() -> {
      // Get CQ Service.
      QueryService cqService = getCache().getQueryService();

      CqQuery cQuery = cqService.getCq(cqName);
      assertThat(cQuery).describedAs("Unexpectedly found CqQuery for CQ : " + cqName).isNull();
    });
  }

  public void validateCQ(VM vm, final String cqName, final int resultSize, final int creates,
      final int updates, final int deletes) {
    validateCQ(vm, cqName, resultSize, creates, updates, deletes, noTest, noTest, noTest, noTest);
  }

  public void validateCQ(VM vm, final String cqName, final int resultSize, final int creates,
      final int updates, final int deletes, final int queryInserts, final int queryUpdates,
      final int queryDeletes, final int totalEvents) {
    vm.invoke(() -> {
      QueryService cqService = getCache().getQueryService();
      CqQuery cQuery = cqService.getCq(cqName);
      assertThat(cQuery).describedAs("Failed to get CqQuery for CQ : " + cqName).isNotNull();

      CqAttributes cqAttr = cQuery.getCqAttributes();
      CqListener[] cqListeners = cqAttr.getCqListeners();
      CqQueryTestListener listener = (CqQueryTestListener) cqListeners[0];
      listener.printInfo(false);

      // Check for totalEvents count.
      if (totalEvents != noTest) {
        // Result size validation.
        listener.printInfo(true);
