/*
 * 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.junit.Assert.fail;

import java.util.HashSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

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

import org.apache.geode.cache.AttributesFactory;
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.Scope;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.SelectResults;
import org.apache.geode.cache.query.Struct;
import org.apache.geode.cache.query.cq.internal.CqQueryImpl;
import org.apache.geode.cache.query.data.Portfolio;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.cache30.CertifiableTestCacheListener;
import org.apache.geode.cache30.ClientServerTestCase;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.test.dunit.Assert;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.IgnoredException;
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.ThreadUtils;
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 ContiunousQuery mechanism in GemFire. This includes the test with different
 * data activities.
 *
 */
@Category({ClientSubscriptionTest.class})
public class CqDataDUnitTest extends JUnit4CacheTestCase {
  private static final Logger logger = LogService.getLogger();
  protected CqQueryDUnitTest cqDUnitTest = new CqQueryDUnitTest();

  public CqDataDUnitTest() {
    super();
  }

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

  protected void postSetUpCqDataDUnitTest() throws Exception {}

  /**
   * Tests with client acting as feeder/publisher and registering cq. Added wrt bug 37161. In case
   * of InterestList the events are not sent back to the client if its the originator, this is not
   * true for cq.
   *
   */
  @Test
  public void testClientWithFeederAndCQ() throws Exception {
    final Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(1);

    cqDUnitTest.createServer(server);

    final int port = server.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server.getHost());

    // Create client.
    cqDUnitTest.createClient(client, port, host0);


    cqDUnitTest.createCQ(client, "testClientWithFeederAndCQ_0", cqDUnitTest.cqs[0]);
    cqDUnitTest.executeCQ(client, "testClientWithFeederAndCQ_0", false, null);

    final int size = 10;
    cqDUnitTest.createValues(client, cqDUnitTest.regions[0], size);
    cqDUnitTest.waitForCreated(client, "testClientWithFeederAndCQ_0", CqQueryDUnitTest.KEY + size);

    cqDUnitTest.validateCQ(client, "testClientWithFeederAndCQ_0",
        /* resultSize: */ CqQueryDUnitTest.noTest, /* creates: */ size, /* updates: */ 0,
        /* deletes; */ 0, /* queryInserts: */ size, /* queryUpdates: */ 0, /* queryDeletes: */ 0,
        /* totalEvents: */ size);

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server);

  }

  /**
   * Test for CQ Fail over/HA with redundancy level set.
   *
   */
  @Test
  public void testCQHAWithState() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);
    VM server3 = host.getVM(2);

    VM client = host.getVM(3);

    // Killing servers can cause this message on the client side.
    IgnoredException.addIgnoredException("Could not find any server");
    cqDUnitTest.createServer(server1);

    final int port1 = server1.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    final int[] ports = AvailablePortHelper.getRandomAvailableTCPPorts(2);

    cqDUnitTest.createServer(server2, ports[0]);
    final int port2 = server2.invoke(() -> CqQueryDUnitTest.getCacheServerPort());

    // Create client - With 3 server endpoints and redundancy level set to 2.

    // Create client with redundancyLevel 1
    cqDUnitTest.createClient(client, new int[] {port1, port2, ports[1]}, host0, "1");

    // Create CQs.
    int numCQs = 1;
    for (int i = 0; i < numCQs; i++) {
      // Create CQs.
      cqDUnitTest.createCQ(client, "testCQHAWithState_" + i, cqDUnitTest.cqs[i]);
      cqDUnitTest.executeCQ(client, "testCQHAWithState_" + i, false, null);
    }

    Wait.pause(1 * 1000);

    int size = 10;

    // CREATE.
    cqDUnitTest.createValues(server1, cqDUnitTest.regions[0], size);
    cqDUnitTest.createValues(server1, cqDUnitTest.regions[1], size);

    for (int i = 1; i <= size; i++) {
      cqDUnitTest.waitForCreated(client, "testCQHAWithState_0", CqQueryDUnitTest.KEY + i);
    }

    // Clients expected initial result.
    int[] resultsCnt = new int[] {10, 1, 2};

    for (int i = 0; i < numCQs; i++) {
      cqDUnitTest.validateCQ(client, "testCQHAWithState_" + i, CqQueryDUnitTest.noTest,
          resultsCnt[i], 0, 0);
    }

    // Close server1.
    // To maintain the redundancy; it will make connection to endpoint-3.
    cqDUnitTest.closeServer(server1);
    Wait.pause(3 * 1000);


    // UPDATE-1.
    cqDUnitTest.createValues(server2, cqDUnitTest.regions[0], 10);
    cqDUnitTest.createValues(server2, cqDUnitTest.regions[1], 10);

    for (int i = 1; i <= size; i++) {
      cqDUnitTest.waitForUpdated(client, "testCQHAWithState_0", CqQueryDUnitTest.KEY + size);
    }

    for (int i = 0; i < numCQs; i++) {
      cqDUnitTest.validateCQ(client, "testCQHAWithState_" + i, CqQueryDUnitTest.noTest,
          resultsCnt[i], resultsCnt[i], CqQueryDUnitTest.noTest);
    }

    // Stop cq.
    cqDUnitTest.stopCQ(client, "testCQHAWithState_0");

    Wait.pause(2 * 1000);

    // UPDATE with stop.
    cqDUnitTest.createServer(server3, ports[1]);
    server3.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    Wait.pause(2 * 1000);

    cqDUnitTest.clearCQListenerEvents(client, "testCQHAWithState_0");

    cqDUnitTest.createValues(server2, cqDUnitTest.regions[0], 10);
    cqDUnitTest.createValues(server2, cqDUnitTest.regions[1], 10);

    // Wait for events at client.
    try {
      cqDUnitTest.waitForUpdated(client, "testCQHAWithState_0", CqQueryDUnitTest.KEY + 1);
      fail("Events not expected since CQ is in stop state.");
    } catch (Exception ex) {
      // Success.
    }

    cqDUnitTest.executeCQ(client, "testCQHAWithState_0", false, null);
    Wait.pause(2 * 1000);

    // Update - 2
    cqDUnitTest.createValues(server3, cqDUnitTest.regions[0], 10);
    cqDUnitTest.createValues(server3, cqDUnitTest.regions[1], 10);

    for (int i = 1; i <= size; i++) {
      cqDUnitTest.waitForUpdated(client, "testCQHAWithState_0", CqQueryDUnitTest.KEY + size);
    }

    for (int i = 0; i < numCQs; i++) {
      cqDUnitTest.validateCQ(client, "testCQHAWithState_" + i, CqQueryDUnitTest.noTest,
          resultsCnt[i], resultsCnt[i] * 2, CqQueryDUnitTest.noTest);
    }

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server2);
    cqDUnitTest.closeServer(server3);
  }


  /**
   * Tests propogation of invalidates and destorys to the clients. Bug 37242.
   *
   */
  @Test
  public void testCQWithDestroysAndInvalidates() throws Exception {
    final Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(1);
    VM producer = host.getVM(2);
    cqDUnitTest.createServer(server, 0, true);
    final int port = server.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server.getHost());

    // Create client.
    cqDUnitTest.createClient(client, port, host0);
    // producer is not doing any thing.
    cqDUnitTest.createClient(producer, port, host0);

    final int size = 10;
    final String name = "testQuery_4";
    cqDUnitTest.createValues(server, cqDUnitTest.regions[0], size);

    cqDUnitTest.createCQ(client, name, cqDUnitTest.cqs[4]);
    cqDUnitTest.executeCQ(client, name, true, null);

    // do destroys and invalidates.
    server.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(cqDUnitTest.regions[0]);
        for (int i = 1; i <= 5; i++) {
          region1.destroy(CqQueryDUnitTest.KEY + i);
        }
      }
    });
    for (int i = 1; i <= 5; i++) {
      cqDUnitTest.waitForDestroyed(client, name, CqQueryDUnitTest.KEY + i);
    }
    // recreate the key values from 1 - 5
    cqDUnitTest.createValues(server, cqDUnitTest.regions[0], 5);
    // wait for all creates to arrive.
    for (int i = 1; i <= 5; i++) {
      cqDUnitTest.waitForCreated(client, name, CqQueryDUnitTest.KEY + i);
    }

    // do more puts to push first five key-value to disk.
    cqDUnitTest.createValues(server, cqDUnitTest.regions[0], 10);
    // do invalidates on fisrt five keys.
    server.invoke(new CacheSerializableRunnable("Create values") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getRootRegion().getSubregion(cqDUnitTest.regions[0]);
        for (int i = 1; i <= 5; i++) {
          region1.invalidate(CqQueryDUnitTest.KEY + i);
        }
      }
    });
    // wait for invalidates now.
    for (int i = 1; i <= 5; i++) {
      cqDUnitTest.waitForInvalidated(client, name, CqQueryDUnitTest.KEY + i);
    }

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server);

  }

  /**
   * Tests make sure that the second client doesnt get more events then there should be. This will
   * test the fix for bug 37295.
   *
   */
  @Test
  public void testCQWithMultipleClients() throws Exception {

    final Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client1 = host.getVM(1);
    VM client2 = host.getVM(2);
    VM client3 = host.getVM(3);

    /* Create Server and Client */
    cqDUnitTest.createServer(server);
    final int port = server.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server.getHost());
    cqDUnitTest.createClient(client1, port, host0);
    cqDUnitTest.createClient(client2, port, host0);

    /* Create CQs. and initialize the region */
    // this should statisfy every thing since id is always greater than
    // zero.
    cqDUnitTest.createCQ(client1, "testCQWithMultipleClients_0", cqDUnitTest.cqs[0]);
    cqDUnitTest.executeCQ(client1, "testCQWithMultipleClients_0", false, null);
    // should only satisfy one key-value pair in the region.
    cqDUnitTest.createCQ(client2, "testCQWithMultipleClients_0", cqDUnitTest.cqs[1]);
    cqDUnitTest.executeCQ(client2, "testCQWithMultipleClients_0", false, null);

    int size = 10;

    // Create Values on Server.
    cqDUnitTest.createValues(server, cqDUnitTest.regions[0], size);


    cqDUnitTest.waitForCreated(client1, "testCQWithMultipleClients_0", CqQueryDUnitTest.KEY + 10);


    /* Validate the CQs */
    cqDUnitTest.validateCQ(client1, "testCQWithMultipleClients_0",
        /* resultSize: */ CqQueryDUnitTest.noTest, /* creates: */ size, /* updates: */ 0,
        /* deletes; */ 0, /* queryInserts: */ size, /* queryUpdates: */ 0, /* queryDeletes: */ 0,
        /* totalEvents: */ size);


    cqDUnitTest.waitForCreated(client2, "testCQWithMultipleClients_0", CqQueryDUnitTest.KEY + 2);


    cqDUnitTest.validateCQ(client2, "testCQWithMultipleClients_0",
        /* resultSize: */ CqQueryDUnitTest.noTest, /* creates: */ 1, /* updates: */ 0,
        /* deletes; */ 0, /* queryInserts: */ 1, /* queryUpdates: */ 0, /* queryDeletes: */ 0,
        /* totalEvents: */ 1);


    /* Close Server and Client */
    cqDUnitTest.closeClient(client2);
    cqDUnitTest.closeClient(client3);
    cqDUnitTest.closeServer(server);
  }

  /**
   * Test for CQ when region is populated with net load.
   *
   */
  @Test
  public void testCQWithLoad() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM server2 = host.getVM(1);

    VM client = host.getVM(2);

    cqDUnitTest.createServer(server1, 0, false, DataPolicy.REPLICATE);
    cqDUnitTest.createServer(server2, 0, false, DataPolicy.REPLICATE);

    final int port1 = server1.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    cqDUnitTest.createClient(client, port1, host0);

    // Create CQs.
    cqDUnitTest.createCQ(client, "testCQWithLoad_0", cqDUnitTest.cqs[0]);
    cqDUnitTest.executeCQ(client, "testCQWithLoad_0", false, null);

    Wait.pause(2 * 1000);

    final int size = 10;

    // CREATE VALUES.
    cqDUnitTest.createValues(server2, cqDUnitTest.regions[0], size);

    server1.invoke(new CacheSerializableRunnable("Load from second server") {
      @Override
      public void run2() throws CacheException {
        Region region1 = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
        for (int i = 1; i <= size; i++) {
          region1.get(CqQueryDUnitTest.KEY + i);
        }
      }
    });

    for (int i = 1; i <= size; i++) {
      cqDUnitTest.waitForCreated(client, "testCQWithLoad_0", CqQueryDUnitTest.KEY + i);
    }

    cqDUnitTest.validateCQ(client, "testCQWithLoad_0", CqQueryDUnitTest.noTest, size, 0, 0);

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server1);
    cqDUnitTest.closeServer(server2);
  }

  /**
   * Test for CQ when entries are evicted from region.
   *
   */
  @Test
  public void testCQWithEviction() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM client = host.getVM(2);

    final int evictionThreshold = 1;
    server1.invoke(new CacheSerializableRunnable("Create Cache Server") {
      @Override
      public void run2() throws CacheException {
        logger.info("### Create Cache Server. ###");
        AttributesFactory factory = new AttributesFactory();
        factory.setScope(Scope.DISTRIBUTED_ACK);
        factory.setDataPolicy(DataPolicy.REPLICATE);
        // factory.setMirrorType(MirrorType.NONE);
        // setting the eviction attributes.
        EvictionAttributes evictAttrs = EvictionAttributes
            .createLRUEntryAttributes(evictionThreshold, EvictionAction.OVERFLOW_TO_DISK);
        factory.setEvictionAttributes(evictAttrs);

        for (int i = 0; i < cqDUnitTest.regions.length; i++) {
          Region region = createRegion(cqDUnitTest.regions[i], factory.createRegionAttributes());
          // Set CacheListener.
          region.getAttributesMutator()
              .addCacheListener(new CertifiableTestCacheListener(LogWriterUtils.getLogWriter()));
        }
        Wait.pause(2000);

        try {
          cqDUnitTest.startBridgeServer(0, true);
        } catch (Exception ex) {
          Assert.fail("While starting CacheServer", ex);
        }
        Wait.pause(2000);

      }
    });

    final int port1 = server1.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server1.getHost());

    cqDUnitTest.createClient(client, port1, host0);

    // Create CQs.
    cqDUnitTest.createCQ(client, "testCQWithEviction_0", cqDUnitTest.cqs[0]);

    final int size = 10;

    // CREATE VALUES.
    cqDUnitTest.createValues(server1, cqDUnitTest.regions[0], size);

    cqDUnitTest.executeCQ(client, "testCQWithEviction_0", false, "CqException");

    Wait.pause(1 * 1000);

    // Update VALUES.
    cqDUnitTest.createValues(server1, cqDUnitTest.regions[0], size);

    for (int i = 1; i <= size; i++) {
      cqDUnitTest.waitForUpdated(client, "testCQWithEviction_0", cqDUnitTest.KEY + i);
    }

    cqDUnitTest.validateCQ(client, "testCQWithEviction_0", cqDUnitTest.noTest, 0, 10, 0);

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server1);
  }

  /**
   * Test for CQ with ConnectionPool.
   *
   */
  @Test
  public void testCQWithConnectionPool() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM client = host.getVM(1);

    cqDUnitTest.createServer(server1, 0, false, DataPolicy.REPLICATE);

    final int port1 = server1.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String serverHost = NetworkUtils.getServerHostName(server1.getHost());

    final String[] regions = cqDUnitTest.regions;
    final int[] serverPorts = new int[] {port1};

    // createClientWithConnectionPool
    SerializableRunnable createClientWithPool =
        new CacheSerializableRunnable("createClientWithPool") {
          @Override
          public void run2() throws CacheException {
            logger.info("### Create Client. ###");
            // Initialize CQ Service.
            try {
              getCache().getQueryService();
            } catch (Exception cqe) {
              cqe.printStackTrace();
              fail("Failed to getCQService.");
            }

            AttributesFactory regionFactory = new AttributesFactory();
            regionFactory.setScope(Scope.LOCAL);
            ClientServerTestCase.configureConnectionPool(regionFactory, serverHost, serverPorts[0],
                -1, false, -1, -1, null);
            for (int i = 0; i < regions.length; i++) {
              createRegion(regions[i], regionFactory.create());
              LogWriterUtils.getLogWriter()
                  .info("### Successfully Created Region on Client :" + regions[i]);
            }
          }
        };

    client.invoke(createClientWithPool);

    // Create CQs.
    cqDUnitTest.createCQ(client, "testCQWithPool_0", cqDUnitTest.cqs[0]);

    // This should fail as Region doesn't have ConnectionPool
    try {
      cqDUnitTest.executeCQ(client, "testCQWithPool_0", false, "CqException");
      fail("CQ Execution should have failed with BridgeClient/Writer not found.");
    } catch (Exception ex) {
      // Expected.
    }

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server1);
  }

  /**
   * Test for CQ with BridgeClient.
   *
   */
  @Test
  public void testCQWithBridgeClient() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM client = host.getVM(1);

    cqDUnitTest.createServer(server1, 0, false, DataPolicy.REPLICATE);

    final int port1 = server1.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String serverHost = NetworkUtils.getServerHostName(server1.getHost());

    final String[] regions = cqDUnitTest.regions;
    final int[] serverPorts = new int[] {port1};

    // createClientWithBridgeClient
    SerializableRunnable createClientWithPool =
        new CacheSerializableRunnable("createClientWithPool") {
          @Override
          public void run2() throws CacheException {
            logger.info("### Create Client. ###");
            // Region region1 = null;
            // Initialize CQ Service.
            try {
              getCache().getQueryService();
            } catch (Exception cqe) {
              cqe.printStackTrace();
              fail("Failed to getCQService.");
            }

            AttributesFactory regionFactory = new AttributesFactory();
            regionFactory.setScope(Scope.LOCAL);
            ClientServerTestCase.configureConnectionPool(regionFactory, serverHost, serverPorts[0],
                -1, true, -1, -1, null);

            for (int i = 0; i < regions.length; i++) {
              createRegion(regions[i], regionFactory.createRegionAttributes());
              LogWriterUtils.getLogWriter()
                  .info("### Successfully Created Region on Client :" + regions[i]);
            }
          }
        };

    client.invoke(createClientWithPool);

    // Create CQs.
    cqDUnitTest.createCQ(client, "testCQWithPool_1", cqDUnitTest.cqs[0]);

    // This should pass.
    cqDUnitTest.executeCQ(client, "testCQWithPool_1", false, null);

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server1);
  }

  /**
   * Test for CQ with ConnectionPool.
   *
   */
  @Test
  public void testCQWithPool() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM client = host.getVM(1);

    cqDUnitTest.createServer(server1, 0, false, DataPolicy.REPLICATE);

    final int port1 = server1.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String serverHost = NetworkUtils.getServerHostName(server1.getHost());

    final String[] regions = cqDUnitTest.regions;
    final int[] serverPorts = new int[] {port1};

    // createClientWithConnectionPool
    SerializableRunnable createClientWithConnectionPool =
        new CacheSerializableRunnable("createClientWithConnectionPool") {
          @Override
          public void run2() throws CacheException {
            logger.info("### Create Client. ###");
            // Region region1 = null;
            // Initialize CQ Service.
            try {
              getCache().getQueryService();
            } catch (Exception cqe) {
              cqe.printStackTrace();
              fail("Failed to getCQService.");
            }

            AttributesFactory regionFactory = new AttributesFactory();
            regionFactory.setScope(Scope.LOCAL);
            ClientServerTestCase.configureConnectionPool(regionFactory, serverHost, serverPorts[0],
                -1, true, -1, -1, null);
            for (int i = 0; i < regions.length; i++) {
              createRegion(regions[i], regionFactory.createRegionAttributes());
              LogWriterUtils.getLogWriter()
                  .info("### Successfully Created Region on Client :" + regions[i]);
            }
          }
        };

    client.invoke(createClientWithConnectionPool);

    // Create CQs.
    cqDUnitTest.createCQ(client, "testCQWithPool_2", cqDUnitTest.cqs[0]);

    // This should pass.
    cqDUnitTest.executeCQ(client, "testCQWithPool_2", false, null);

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server1);
  }

  /**
   * Test for CQ with establishCallBackConnection.
   *
   */
  @Test
  public void testCQWithEstablishCallBackConnection() throws Exception {
    final Host host = Host.getHost(0);
    VM server1 = host.getVM(0);
    VM client = host.getVM(1);

    cqDUnitTest.createServer(server1, 0, false, DataPolicy.REPLICATE);

    final int port1 = server1.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String serverHost = NetworkUtils.getServerHostName(server1.getHost());

    final String[] regions = cqDUnitTest.regions;
    final int[] serverPorts = new int[] {port1};

    // createClientWithPool
    SerializableRunnable createClientWithPool =
        new CacheSerializableRunnable("createClientWithPool") {
          @Override
          public void run2() throws CacheException {
            logger.info("### Create Client. ###");
            // Region region1 = null;
            // Initialize CQ Service.
            try {
              getCache().getQueryService();
            } catch (Exception cqe) {
              cqe.printStackTrace();
              fail("Failed to getCQService.");
            }

            AttributesFactory regionFactory = new AttributesFactory();
            regionFactory.setScope(Scope.LOCAL);

            ClientServerTestCase.configureConnectionPool(regionFactory, serverHost, serverPorts[0],
                -1, false, -1, -1, null);

            for (int i = 0; i < regions.length; i++) {
              createRegion(regions[i], regionFactory.createRegionAttributes());
              LogWriterUtils.getLogWriter()
                  .info("### Successfully Created Region on Client :" + regions[i]);
            }
          }
        };


    client.invoke(createClientWithPool);

    // Create CQs.
    cqDUnitTest.createCQ(client, "testCQWithEstablishCallBackConnection_0", cqDUnitTest.cqs[0]);

    // This should fail.
    try {
      cqDUnitTest.executeCQ(client, "testCQWithEstablishCallBackConnection_0", false,
          "CqException");
      fail("Test should have failed with connection with establishCallBackConnection not found.");
    } catch (Exception ex) {
      // Expected.
    }

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server1);
  }

  /**
   * Test for: Region destroy, calls close on the server. Region clear triggers cqEvent with query
   * op region clear. Region invalidate triggers cqEvent with query op region invalidate.
   *
   */
  @Test
  public void testRegionEvents() throws Exception {

    final Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(1);

    cqDUnitTest.createServer(server);
    final int port = server.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server.getHost());

    cqDUnitTest.createClient(client, port, host0);

    // Create CQ on regionA
    cqDUnitTest.createCQ(client, "testRegionEvents_0", cqDUnitTest.cqs[0]);
    cqDUnitTest.executeCQ(client, "testRegionEvents_0", false, null);

    // Create CQ on regionB
    cqDUnitTest.createCQ(client, "testRegionEvents_1", cqDUnitTest.cqs[2]);
    cqDUnitTest.executeCQ(client, "testRegionEvents_1", false, null);

    // Test for Event on Region Clear.
    server.invoke(new CacheSerializableRunnable("testRegionEvents") {
      @Override
      public void run2() throws CacheException {
        logger.info("### Clearing the region on the server ###");
        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
        for (int i = 1; i <= 5; i++) {
          region.put(CqQueryDUnitTest.KEY + i, new Portfolio(i));
        }
        region.clear();
      }
    });

    cqDUnitTest.waitForRegionClear(client, "testRegionEvents_0");

    // Test for Event on Region invalidate.
    server.invoke(new CacheSerializableRunnable("testRegionEvents") {
      @Override
      public void run2() throws CacheException {
        logger.info("### Invalidate the region on the server ###");
        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
        for (int i = 1; i <= 5; i++) {
          region.put(CqQueryDUnitTest.KEY + i, new Portfolio(i));
        }
        region.invalidateRegion();
      }
    });

    cqDUnitTest.waitForRegionInvalidate(client, "testRegionEvents_0");

    // Test for Event on Region destroy.
    server.invoke(new CacheSerializableRunnable("testRegionEvents") {
      @Override
      public void run2() throws CacheException {
        logger.info("### Destroying the region on the server ###");
        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[1]);
        for (int i = 1; i <= 5; i++) {
          region.put(CqQueryDUnitTest.KEY + i, new Portfolio(i));
        }
        // this should close one cq on client.
        region.destroyRegion();
      }
    });

    Wait.pause(1000); // wait for cq to close becuse of region destroy on server.
    // cqDUnitTest.waitForClose(client,"testRegionEvents_1");
    cqDUnitTest.validateCQCount(client, 1);


    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server);

  }


  /**
   * Test for events created during the CQ query execution. When CQs are executed using
   * executeWithInitialResults there may be possibility that the region changes during that time may
   * not be reflected in the query result set thus making the query data and region data
   * inconsistent.
   *
   */
  @Test
  public void testEventsDuringQueryExecution() throws Exception {

    final Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(1);
    final String cqName = "testEventsDuringQueryExecution_0";
    cqDUnitTest.createServer(server);
    final int port = server.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server.getHost());

    // Initialize Client.
    cqDUnitTest.createClient(client, port, host0);

    // create CQ.
    cqDUnitTest.createCQ(client, cqName, cqDUnitTest.cqs[0]);

    final int numObjects = 200;
    final int totalObjects = 500;

    // initialize Region.
    server.invoke(new CacheSerializableRunnable("Update Region") {
      @Override
      public void run2() throws CacheException {
        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
        for (int i = 1; i <= numObjects; i++) {
          Portfolio p = new Portfolio(i);
          region.put("" + i, p);
        }
      }
    });

    // Execute CQ while update is in progress.
    AsyncInvocation processCqs = client.invokeAsync(new CacheSerializableRunnable("Execute CQ") {
      @Override
      public void run2() throws CacheException {
        QueryService cqService = getCache().getQueryService();
        // Get CqQuery object.
        CqQuery cq1 = cqService.getCq(cqName);
        if (cq1 == null) {
          fail("Failed to get CQ " + cqName);
        }

        SelectResults cqResults = null;

        try {
          cqResults = cq1.executeWithInitialResults();
        } catch (Exception ex) {
          AssertionError err = new AssertionError("Failed to execute  CQ " + cqName);
          err.initCause(ex);
          throw err;
        }

        // getLogWriter().info("initial result size = " + cqResults.size());

        CqQueryTestListener cqListener =
            (CqQueryTestListener) cq1.getCqAttributes().getCqListener();
        // Wait for the last key to arrive.
        cqListener.waitForCreated("" + totalObjects);

        // Check if the events from CqListener are in order.
        int oldId = 0;
        for (Object cqEvent : cqListener.events.toArray()) {
          int newId = new Integer(cqEvent.toString()).intValue();
          if (oldId > newId) {
            fail("Queued events for CQ Listener during execution with "
                + "Initial results is not in the order in which they are created.");
          }
          oldId = newId;
        }

        // Check if all the IDs are present as part of Select Results and CQ Events.
        HashSet ids = new HashSet(cqListener.events);
        for (Object o : cqResults.asList()) {
          Struct s = (Struct) o;
          ids.add(s.get("key"));
        }

        // Iterator iter = cqResults.asSet().iterator();
        // while (iter.hasNext()) {
        // Portfolio p = (Portfolio)iter.next();
        // ids.add(p.getPk());
        // //getLogWriter().info("Result set value : " + p.getPk());
        // }

        HashSet missingIds = new HashSet();
        String key = "";
        for (int i = 1; i <= totalObjects; i++) {
          key = "" + i;
          if (!(ids.contains(key))) {
            missingIds.add(key);
          }
        }

        if (!missingIds.isEmpty()) {
          fail("Missing Keys in either ResultSet or the Cq Event list. "
              + " Missing keys : [size : " + missingIds.size() + "]" + missingIds
              + " Ids in ResultSet and CQ Events :" + ids);
        }

      }
    });

    // Keep updating region (async invocation).
    server.invokeAsync(new CacheSerializableRunnable("Update Region") {
      @Override
      public void run2() throws CacheException {
        // Wait to allow client a chance to register the cq
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
        for (int i = numObjects + 1; i <= totalObjects; i++) {
          Portfolio p = new Portfolio(i);
          region.put("" + i, p);
        }
      }
    });

    // wait for 60 seconds for test to complete
    ThreadUtils.join(processCqs, 60 * 1000);

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server);
  }

  /**
   * This test was created to test executeWithInitialResults being called multiple times.
   * Previously, the queueEvents would be overwritten and we would lose data. This test will execute
   * the method twice. The first time, the first execution will block it's own child thread (TC1).
   * The second execution will block until TC1 is completed (based on how executeWithInitialResults
   * is implemented) A third thread will be awaken and release the latch in the testhook for TC1 to
   * complete.
   *
   */
  @Test
  @Ignore("GEODE-5863 - The test fails with an Awaitility timeout after increasing the timeout. It previously ignored the timeout")
  public void testMultipleExecuteWithInitialResults() throws Exception {
    final int numObjects = 200;
    final int totalObjects = 500;
    final Host host = Host.getHost(0);
    VM server = host.getVM(0);
    VM client = host.getVM(1);
    client.invoke(setTestHook());
    final String cqName = "testMultiExecuteWithInitialResults";

    // initialize server and retreive host and port values
    cqDUnitTest.createServer(server);
    final int port = server.invoke(() -> CqQueryDUnitTest.getCacheServerPort());
    final String host0 = NetworkUtils.getServerHostName(server.getHost());

    // Initialize Client.
    cqDUnitTest.createClient(client, port, host0);

    // create CQ.
    cqDUnitTest.createCQ(client, cqName, cqDUnitTest.cqs[0]);

    // initialize Region.
    server.invoke(new CacheSerializableRunnable("Update Region") {
      @Override
      public void run2() throws CacheException {
        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
        for (int i = 1; i <= numObjects; i++) {
          Portfolio p = new Portfolio(i);
          region.put("" + i, p);
        }
      }
    });

    // Keep updating region (async invocation).
    server.invokeAsync(new CacheSerializableRunnable("Update Region") {
      @Override
      public void run2() throws CacheException {
        // Wait to give client a chance to register the cq
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
        for (int i = numObjects + 1; i <= totalObjects; i++) {
          Portfolio p = new Portfolio(i);
          region.put("" + i, p);
        }
      }
    });

    // the thread that validates all results and executes first
    AsyncInvocation processCqs =
        client.invokeAsync(new CacheSerializableRunnable("Execute CQ first") {
          @Override
          public void run2() throws CacheException {
            SelectResults cqResults = null;
            QueryService cqService = getCache().getQueryService();
            // Get CqQuery object.
            CqQuery cq1 = cqService.getCq(cqName);
            if (cq1 == null) {
              fail("Failed to get CQ " + cqName);
            }
            try {
              cqResults = cq1.executeWithInitialResults();

            } catch (Exception e) {
              AssertionError err = new AssertionError("Failed to execute  CQ " + cqName);
              err.initCause(e);
              throw err;
            }

            CqQueryTestListener cqListener =
                (CqQueryTestListener) cq1.getCqAttributes().getCqListener();
            // Wait for the last key to arrive.
            cqListener.waitForCreated("" + totalObjects);
            // Check if the events from CqListener are in order.
            int oldId = 0;
            for (Object cqEvent : cqListener.events.toArray()) {
              int newId = new Integer(cqEvent.toString()).intValue();
              if (oldId > newId) {
                fail("Queued events for CQ Listener during execution with "
                    + "Initial results is not in the order in which they are created.");
              }
              oldId = newId;
            }

            // Check if all the IDs are present as part of Select Results and CQ
            // Events.
            HashSet ids = new HashSet(cqListener.events);
            for (Object o : cqResults.asList()) {
              Struct s = (Struct) o;
              ids.add(s.get("key"));
            }

            HashSet missingIds = new HashSet();
            String key = "";
            for (int i = 1; i <= totalObjects; i++) {
              key = "" + i;
              if (!(ids.contains(key))) {
                missingIds.add(key);
              }
            }

            if (!missingIds.isEmpty()) {
              fail("Missing Keys in either ResultSet or the Cq Event list. "
                  + " Missing keys : [size : " + missingIds.size() + "]" + missingIds
                  + " Ids in ResultSet and CQ Events :" + ids);
            }
          }
        });

    // the second call to executeWithInitialResults. Goes to sleep hopefully
    // long enough
    // for the first call to executeWithInitialResults first
    client.invokeAsync(new CacheSerializableRunnable("Execute CQ second") {
      @Override
      public void run2() throws CacheException {
        try {
          Thread.sleep(2000);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
        QueryService cqService = getCache().getQueryService();
        // Get CqQuery object.
        CqQuery cq1 = cqService.getCq(cqName);
        if (cq1 == null) {
          fail("Failed to get CQ " + cqName);
        }
        try {
          cq1.executeWithInitialResults();
        } catch (IllegalStateException e) {
          // we expect an error due to the cq having already being in run state
        } catch (Exception e) {
          AssertionError err = new AssertionError("test hook lock interrupted" + cqName);
          err.initCause(e);
          throw err;
        }
      }
    });

    // thread that unlatches the test hook, sleeping long enough for both
    // the other two threads to execute first
    client.invokeAsync(new CacheSerializableRunnable("Release latch") {
      @Override
      public void run2() throws CacheException {
        // we wait to release the testHook and hope the other two threads have
        // had a chance to invoke executeWithInitialResults
        try {
          Thread.sleep(5000);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          AssertionError err = new AssertionError("test hook lock interrupted" + cqName);
          err.initCause(e);
          throw err;
        }
        CqQueryImpl.testHook.ready();
      }
    });

    // wait for 60 seconds for test to complete
    processCqs.get(1, TimeUnit.MINUTES);

    // Close.
    cqDUnitTest.closeClient(client);
    cqDUnitTest.closeServer(server);
  }

  public CacheSerializableRunnable setTestHook() {
    SerializableRunnable sr = new CacheSerializableRunnable("TestHook") {
      @Override
      public void run2() {
        class CqQueryTestHook implements CqQueryImpl.TestHook {

          CountDownLatch latch = new CountDownLatch(1);

          @Override
          public void pauseUntilReady() {
            try {
              latch.await();
            } catch (Exception e) {
              e.printStackTrace();
              Thread.currentThread().interrupt();
            }
          }

          @Override
          public void ready() {
            latch.countDown();
          }

          @Override
          public int numQueuedEvents() {
            // TODO Auto-generated method stub
            return 0;
          }

          @Override
          public void setEventCount(int count) {
            // TODO Auto-generated method stub

          }

        };
        CqQueryImpl.testHook = new CqQueryTestHook();
      }
    };
    return (CacheSerializableRunnable) sr;
  }

}