| /* |
| * 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.distributed.ConfigurationProperties.DURABLE_CLIENT_ID; |
| import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_TIMEOUT; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.test.dunit.Assert.assertEquals; |
| import static org.apache.geode.test.dunit.Assert.assertFalse; |
| import static org.apache.geode.test.dunit.Assert.assertNotNull; |
| import static org.apache.geode.test.dunit.Assert.assertNull; |
| import static org.apache.geode.test.dunit.Assert.assertTrue; |
| import static org.apache.geode.test.dunit.Assert.fail; |
| |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.apache.logging.log4j.Logger; |
| 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.EvictionAttributes; |
| import org.apache.geode.cache.MirrorType; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.client.Pool; |
| import org.apache.geode.cache.client.PoolFactory; |
| 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.CqEvent; |
| import org.apache.geode.cache.query.CqException; |
| 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.QueryService; |
| import org.apache.geode.cache.query.RegionNotFoundException; |
| 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.cq.internal.CqQueryImpl.TestHook; |
| import org.apache.geode.cache.query.data.Portfolio; |
| import org.apache.geode.cache30.CacheSerializableRunnable; |
| import org.apache.geode.cache30.CertifiableTestCacheListener; |
| import org.apache.geode.internal.AvailablePortHelper; |
| import org.apache.geode.internal.cache.PoolFactoryImpl; |
| import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| import org.apache.geode.test.awaitility.GeodeAwaitility; |
| 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.VM; |
| import org.apache.geode.test.dunit.Wait; |
| import org.apache.geode.test.dunit.WaitCriterion; |
| 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. This includes the test with different |
| * data activities. |
| */ |
| @Category({ClientSubscriptionTest.class}) |
| public class CqDataUsingPoolDUnitTest extends JUnit4CacheTestCase { |
| |
| protected CqQueryUsingPoolDUnitTest cqDUnitTest = new CqQueryUsingPoolDUnitTest(); // TODO: don't |
| // do this! |
| private static final Logger logger = LogService.getLogger(); |
| |
| @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(); |
| } |
| }); |
| postSetUpCqDataUsingPoolDUnitTest(); |
| } |
| |
| protected void postSetUpCqDataUsingPoolDUnitTest() 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(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server.getHost()); |
| |
| String poolName = "testClientWithFeederAndCQ"; |
| cqDUnitTest.createPool(client, poolName, host0, port); |
| |
| // Create client. |
| cqDUnitTest.createClient(client, port, host0); |
| |
| |
| cqDUnitTest.createCQ(client, poolName, "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", |
| CqQueryUsingPoolDUnitTest.KEY + size); |
| |
| cqDUnitTest.validateCQ(client, "testClientWithFeederAndCQ_0", |
| /* resultSize: */ CqQueryUsingPoolDUnitTest.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); |
| |
| cqDUnitTest.createServer(server1); |
| |
| final int port1 = server1.invoke(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server1.getHost()); |
| |
| final int[] ports = AvailablePortHelper.getRandomAvailableTCPPorts(2); |
| |
| cqDUnitTest.createServer(server2, ports[0]); |
| final int port2 = server2.invoke(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| |
| // Create client - With 3 server endpoints and redundancy level set to 2. |
| |
| // Create client with redundancyLevel 1 |
| |
| String poolName = "testCQHAWithState"; |
| cqDUnitTest.createPool(client, poolName, new String[] {host0, host0, host0}, |
| new int[] {port1, port2, ports[1]}, "1"); |
| |
| // Create CQs. |
| int numCQs = 1; |
| for (int i = 0; i < numCQs; i++) { |
| // Create CQs. |
| cqDUnitTest.createCQ(client, poolName, "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", CqQueryUsingPoolDUnitTest.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, CqQueryUsingPoolDUnitTest.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", |
| CqQueryUsingPoolDUnitTest.KEY + size); |
| } |
| |
| for (int i = 0; i < numCQs; i++) { |
| cqDUnitTest.validateCQ(client, "testCQHAWithState_" + i, CqQueryUsingPoolDUnitTest.noTest, |
| resultsCnt[i], resultsCnt[i], CqQueryUsingPoolDUnitTest.noTest); |
| } |
| |
| // Stop cq. |
| cqDUnitTest.stopCQ(client, "testCQHAWithState_0"); |
| |
| Wait.pause(2 * 1000); |
| |
| // UPDATE with stop. |
| cqDUnitTest.createServer(server3, ports[1]); |
| server3.invoke(() -> CqQueryUsingPoolDUnitTest.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", CqQueryUsingPoolDUnitTest.KEY + 1); |
| fail("Events not expected since CQ is in stop state."); |
| } catch (Exception expected) { |
| // Success. |
| } |
| |
| cqDUnitTest.executeCQ(client, "testCQHAWithState_0", false, null); |
| |
| // 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", |
| CqQueryUsingPoolDUnitTest.KEY + size); |
| } |
| |
| for (int i = 0; i < numCQs; i++) { |
| cqDUnitTest.validateCQ(client, "testCQHAWithState_" + i, CqQueryUsingPoolDUnitTest.noTest, |
| resultsCnt[i], resultsCnt[i] * 2, CqQueryUsingPoolDUnitTest.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(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server.getHost()); |
| |
| String poolName = "testCQWithDestroysAndInvalidates"; |
| cqDUnitTest.createPool(client, poolName, host0, port); |
| |
| // 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, poolName, 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(CqQueryUsingPoolDUnitTest.KEY + i); |
| } |
| } |
| }); |
| for (int i = 1; i <= 5; i++) { |
| cqDUnitTest.waitForDestroyed(client, name, CqQueryUsingPoolDUnitTest.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, CqQueryUsingPoolDUnitTest.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(CqQueryUsingPoolDUnitTest.KEY + i); |
| } |
| } |
| }); |
| // wait for invalidates now. |
| for (int i = 1; i <= 5; i++) { |
| cqDUnitTest.waitForInvalidated(client, name, CqQueryUsingPoolDUnitTest.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(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server.getHost()); |
| |
| String poolName1 = "testCQWithMultipleClients1"; |
| String poolName2 = "testCQWithMultipleClients2"; |
| |
| cqDUnitTest.createPool(client1, poolName1, host0, port); |
| cqDUnitTest.createPool(client2, poolName2, host0, port); |
| |
| /* Create CQs. and initialize the region */ |
| // this should stasify every thing since id is always greater than |
| // zero. |
| cqDUnitTest.createCQ(client1, poolName1, "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, poolName2, "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", |
| CqQueryUsingPoolDUnitTest.KEY + 10); |
| |
| /* Validate the CQs */ |
| cqDUnitTest.validateCQ(client1, "testCQWithMultipleClients_0", |
| /* resultSize: */ CqQueryUsingPoolDUnitTest.noTest, /* creates: */ size, /* updates: */ 0, |
| /* deletes; */ 0, /* queryInserts: */ size, /* queryUpdates: */ 0, /* queryDeletes: */ 0, |
| /* totalEvents: */ size); |
| |
| cqDUnitTest.waitForCreated(client2, "testCQWithMultipleClients_0", |
| CqQueryUsingPoolDUnitTest.KEY + 2); |
| |
| cqDUnitTest.validateCQ(client2, "testCQWithMultipleClients_0", |
| /* resultSize: */ CqQueryUsingPoolDUnitTest.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, MirrorType.KEYS_VALUES); |
| cqDUnitTest.createServer(server2, 0, false, MirrorType.KEYS); |
| |
| final int port1 = server1.invoke(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server1.getHost()); |
| |
| String poolName = "testCQWithLoad"; |
| cqDUnitTest.createPool(client, poolName, host0, port1); |
| |
| // cqDUnitTest.createClient(client, port1, host0); |
| |
| // Create CQs. |
| cqDUnitTest.createCQ(client, poolName, "testCQWithLoad_0", cqDUnitTest.cqs[0]); |
| cqDUnitTest.executeCQ(client, "testCQWithLoad_0", false, null); |
| |
| Wait.pause(1 * 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(CqQueryUsingPoolDUnitTest.KEY + i); |
| } |
| } |
| }); |
| |
| for (int i = 1; i <= size; i++) { |
| cqDUnitTest.waitForCreated(client, "testCQWithLoad_0", CqQueryUsingPoolDUnitTest.KEY + i); |
| } |
| |
| cqDUnitTest.validateCQ(client, "testCQWithLoad_0", CqQueryUsingPoolDUnitTest.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 server2 = host.getVM(1); |
| |
| VM client = host.getVM(2); |
| |
| final int evictionThreshold = 5; |
| server1.invoke(new CacheSerializableRunnable("Create Cache Server") { |
| @Override |
| public void run2() throws CacheException { |
| LogWriterUtils.getLogWriter().info("### Create Cache Server. ###"); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| factory.setMirrorType(MirrorType.NONE); |
| |
| // setting the eviction attributes. |
| factory |
| .setEvictionAttributes(EvictionAttributes.createLRUEntryAttributes(evictionThreshold)); |
| 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); |
| } |
| }); |
| |
| cqDUnitTest.createServer(server2, 0, false, MirrorType.NONE); |
| |
| final int port1 = server1.invoke(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server1.getHost()); |
| |
| String poolName = "testCQWithEviction"; |
| cqDUnitTest.createPool(client, poolName, host0, port1); |
| |
| // cqDUnitTest.createClient(client, port1, host0); |
| |
| // Create CQs. |
| cqDUnitTest.createCQ(client, poolName, "testCQWithEviction_0", cqDUnitTest.cqs[0]); |
| |
| // This should fail as Region is not replicated. |
| // There is a bug37966 filed on this. |
| try { |
| cqDUnitTest.executeCQ(client, "testCQWithEviction_0", false, "CqException"); |
| fail("Should have thrown exception, cq not supported on Non-replicated region."); |
| } catch (Exception expected) { |
| // Ignore expected. |
| } |
| |
| Wait.pause(1 * 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(CqQueryUsingPoolDUnitTest.KEY + i); |
| } |
| } |
| }); |
| |
| Wait.pause(2 * 1000); |
| |
| server1.invoke(new CacheSerializableRunnable("validate destroy") { |
| @Override |
| public void run2() throws CacheException { |
| Region region = getRootRegion().getSubregion(cqDUnitTest.regions[0]); |
| assertNotNull(region); |
| |
| Set keys = region.entrySet(); |
| int keyCnt = size - evictionThreshold; |
| assertEquals("Mismatch, number of keys in local region is not equal to the expected size", |
| keyCnt, keys.size()); |
| |
| CertifiableTestCacheListener ctl = |
| (CertifiableTestCacheListener) region.getAttributes().getCacheListener(); |
| for (int i = 1; i <= keyCnt; i++) { |
| ctl.waitForDestroyed(CqQueryUsingPoolDUnitTest.KEY + i); |
| assertNull(region.getEntry(CqQueryUsingPoolDUnitTest.KEY + i)); |
| } |
| } |
| }); |
| |
| // Close. |
| cqDUnitTest.closeClient(client); |
| cqDUnitTest.closeServer(server1); |
| cqDUnitTest.closeServer(server2); |
| } |
| |
| /** |
| * 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, MirrorType.KEYS_VALUES); |
| |
| final int port1 = server1.invoke(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String serverHost = NetworkUtils.getServerHostName(server1.getHost()); |
| |
| final String poolName = "testCQWithEstablishCallBackConnection"; |
| |
| client.invoke(new CacheSerializableRunnable("createPool :" + poolName) { |
| @Override |
| public void run2() throws CacheException { |
| // Create Cache. |
| getCache(); |
| |
| PoolFactory cpf = PoolManager.createFactory(); |
| cpf.setSubscriptionEnabled(false); |
| cpf.addServer(serverHost, port1); |
| cpf.create(poolName); |
| } |
| }); |
| |
| // Create CQs. |
| cqDUnitTest.createCQ(client, poolName, "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 expected) { |
| // 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(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server.getHost()); |
| |
| String poolName = "testRegionEvents"; |
| cqDUnitTest.createPool(client, poolName, host0, port); |
| |
| // cqDUnitTest.createClient(client, port, host0); |
| |
| // Create CQ on regionA |
| cqDUnitTest.createCQ(client, poolName, "testRegionEvents_0", cqDUnitTest.cqs[0]); |
| cqDUnitTest.executeCQ(client, "testRegionEvents_0", false, null); |
| |
| // Create CQ on regionB |
| cqDUnitTest.createCQ(client, poolName, "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 { |
| LogWriterUtils.getLogWriter().info("### Clearing the region on the server ###"); |
| Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]); |
| for (int i = 1; i <= 5; i++) { |
| region.put(CqQueryUsingPoolDUnitTest.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 { |
| LogWriterUtils.getLogWriter().info("### Invalidate the region on the server ###"); |
| Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]); |
| for (int i = 1; i <= 5; i++) { |
| region.put(CqQueryUsingPoolDUnitTest.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 { |
| LogWriterUtils.getLogWriter().info("### Destroying the region on the server ###"); |
| Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[1]); |
| for (int i = 1; i <= 5; i++) { |
| region.put(CqQueryUsingPoolDUnitTest.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); |
| final VM client = host.getVM(1); |
| final String cqName = "testEventsDuringQueryExecution_0"; |
| cqDUnitTest.createServer(server); |
| |
| final int port = server.invoke(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server.getHost()); |
| |
| String poolName = "testEventsDuringQueryExecution"; |
| cqDUnitTest.createPool(client, poolName, host0, port); |
| |
| // create CQ. |
| cqDUnitTest.createCQ(client, poolName, 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); |
| } |
| } |
| }); |
| |
| // First set testhook in executeWithInitialResults so that queued events |
| // are not drained before we verify there number. |
| client.invoke(setTestHook()); |
| |
| // Execute CQ while update is in progress. |
| AsyncInvocation executeCq = |
| client.invokeAsync(new CacheSerializableRunnable("Execute CQ AsyncInvoke") { |
| @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) { |
| fail("CQ execution failed", ex); |
| } |
| |
| // Check num of events received during executeWithInitialResults. |
| final TestHook testHook = CqQueryImpl.testHook; |
| GeodeAwaitility.await().untilAsserted(new WaitCriterion() { |
| |
| @Override |
| public boolean done() { |
| return testHook.numQueuedEvents() > 0; |
| } |
| |
| @Override |
| public String description() { |
| return "No queued events found."; |
| } |
| }); |
| |
| getCache().getLogger().fine("Queued Events Size" + testHook.numQueuedEvents()); |
| // Make sure CQEvents are queued during execute with initial results. |
| |
| 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); |
| } |
| } |
| }); |
| |
| // Keep updating region (async invocation). |
| server.invoke(new CacheSerializableRunnable("Update Region") { |
| @Override |
| public void run2() throws CacheException { |
| Wait.pause(200); |
| client.invoke(new CacheSerializableRunnable("Releasing the latch") { |
| @Override |
| public void run2() throws CacheException { |
| // Now release the testhook so that CQListener can proceed. |
| final TestHook testHook = CqQueryImpl.testHook; |
| testHook.ready(); |
| } |
| }); |
| 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); |
| } |
| } |
| }); |
| |
| // Close. |
| cqDUnitTest.closeClient(client); |
| cqDUnitTest.closeServer(server); |
| } |
| |
| @Test |
| public void testCqStatInitializationTimingIssue() { |
| disconnectAllFromDS(); |
| |
| // The async close can cause this exception on the server |
| IgnoredException.addIgnoredException("java.net.SocketException: Broken pipe"); |
| final String regionName = "testCqStatInitializationTimingIssue"; |
| final String cq1Name = "testCq1"; |
| final Host host = Host.getHost(0); |
| VM server = host.getVM(0); |
| VM client = host.getVM(1); |
| VM client2 = host.getVM(2); |
| |
| // Start server 1 |
| final int server1Port = ((Integer) server |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| |
| // Start a client |
| client.invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(client.getHost()), server1Port), regionName)); |
| |
| // Start a pub client |
| client2.invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(client2.getHost()), server1Port), regionName)); |
| |
| // client has thread that invokes new and remove cq over and over |
| client.invokeAsync(new CacheSerializableRunnable("Register cq") { |
| @Override |
| public void run2() throws CacheException { |
| for (int i = 0; i < 10000; i++) { |
| CqQuery query = createCq(regionName, cq1Name); |
| if (query != null) { |
| try { |
| query.close(); |
| } catch (Exception e) { |
| fail("exception while closing cq:", e); |
| } |
| } |
| } |
| } |
| }); |
| |
| client2.invokeAsync(new CacheSerializableRunnable("pub updates") { |
| @Override |
| public void run2() throws CacheException { |
| Region region = CacheServerTestUtil.getCache().getRegion(regionName); |
| while (true) { |
| for (int i = 0; i < 50000; i++) { |
| region.put("" + i, "" + Math.random()); |
| } |
| } |
| } |
| }); |
| |
| server.invokeAsync(new CacheSerializableRunnable("pub updates") { |
| @Override |
| public void run2() throws CacheException { |
| Region region = CacheServerTestUtil.getCache().getRegion(regionName); |
| while (true) { |
| for (int i = 0; i < 50000; i++) { |
| region.put("" + i, "" + Math.random()); |
| } |
| } |
| } |
| }); |
| |
| // client has another thread that retrieves cq map and checks stat over and over |
| client.invoke(new CacheSerializableRunnable("Check Stats") { |
| @Override |
| public void run2() throws CacheException { |
| for (int i = 0; i < 10000; i++) { |
| checkCqStats(cq1Name); |
| } |
| } |
| }); |
| |
| client.invoke(() -> CacheServerTestUtil.closeCache()); |
| client2.invoke(() -> CacheServerTestUtil.closeCache()); |
| server.invoke(() -> CacheServerTestUtil.closeCache()); |
| } |
| |
| @Test |
| public void testGetDurableCQsFromPoolOnly() throws Exception { |
| final String regionName = "regionA"; |
| final Host host = Host.getHost(0); |
| VM server = host.getVM(0); |
| VM client1 = host.getVM(1); |
| VM client2 = host.getVM(2); |
| |
| /* Create Server and Client */ |
| cqDUnitTest.createServer(server); |
| final int port = server.invoke(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server.getHost()); |
| |
| final String poolName1 = "pool1"; |
| final String poolName2 = "pool2"; |
| |
| cqDUnitTest.createPool(client1, poolName1, host0, port); |
| cqDUnitTest.createPool(client2, poolName2, host0, port); |
| |
| client1.invoke(new CacheSerializableRunnable("Register cq for client 1") { |
| @Override |
| public void run2() throws CacheException { |
| |
| QueryService queryService = null; |
| try { |
| queryService = (PoolManager.find(poolName1)).getQueryService(); |
| } catch (Exception cqe) { |
| Assert.fail("Failed to getCQService.", cqe); |
| } |
| try { |
| CqAttributesFactory cqAf = new CqAttributesFactory(); |
| CqAttributes attributes = cqAf.create(); |
| queryService.newCq("client1DCQ1", "Select * From /root/" + regionName + " where id = 1", |
| attributes, true).execute(); |
| queryService.newCq("client1DCQ2", "Select * From /root/" + regionName + " where id = 10", |
| attributes, true).execute(); |
| queryService.newCq("client1NoDC1", "Select * From /root/" + regionName, attributes, false) |
| .execute(); |
| queryService.newCq("client1NoDC2", "Select * From /root/" + regionName + " where id = 3", |
| attributes, false).execute(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } catch (CqExistsException e) { |
| fail("failed", e); |
| } catch (RegionNotFoundException e) { |
| fail("failed", e); |
| } |
| } |
| }); |
| |
| client2.invoke(new CacheSerializableRunnable("Register cq for client 2") { |
| @Override |
| public void run2() throws CacheException { |
| |
| QueryService queryService = null; |
| try { |
| queryService = (PoolManager.find(poolName2)).getQueryService(); |
| } catch (Exception cqe) { |
| Assert.fail("Failed to getCQService.", cqe); |
| } |
| try { |
| CqAttributesFactory cqAf = new CqAttributesFactory(); |
| CqAttributes attributes = cqAf.create(); |
| queryService.newCq("client2DCQ1", "Select * From /root/" + regionName + " where id = 1", |
| attributes, true).execute(); |
| queryService.newCq("client2DCQ2", "Select * From /root/" + regionName + " where id = 10", |
| attributes, true).execute(); |
| queryService.newCq("client2DCQ3", "Select * From /root/" + regionName, attributes, true) |
| .execute(); |
| queryService.newCq("client2DCQ4", "Select * From /root/" + regionName + " where id = 3", |
| attributes, true).execute(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } catch (CqExistsException e) { |
| fail("failed", e); |
| } catch (RegionNotFoundException e) { |
| fail("failed", e); |
| } |
| } |
| }); |
| |
| client2.invoke(new CacheSerializableRunnable("test getDurableCQsFromServer for client2") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = null; |
| try { |
| queryService = (PoolManager.find(poolName2)).getQueryService(); |
| } catch (Exception cqe) { |
| Assert.fail("Failed to getCQService.", cqe); |
| } |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(4, list.size()); |
| assertTrue(list.contains("client2DCQ1")); |
| assertTrue(list.contains("client2DCQ2")); |
| assertTrue(list.contains("client2DCQ3")); |
| assertTrue(list.contains("client2DCQ4")); |
| } |
| }); |
| |
| client1.invoke(new CacheSerializableRunnable("test getDurableCQsFromServer for client1") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = null; |
| try { |
| queryService = (PoolManager.find(poolName1)).getQueryService(); |
| } catch (Exception cqe) { |
| Assert.fail("Failed to getCQService.", cqe); |
| } |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(2, list.size()); |
| assertTrue(list.contains("client1DCQ1")); |
| assertTrue(list.contains("client1DCQ2")); |
| } |
| }); |
| |
| cqDUnitTest.closeClient(client2); |
| cqDUnitTest.closeClient(client1); |
| cqDUnitTest.closeServer(server); |
| } |
| |
| @Test |
| public void testGetDurableCQsFromServerWithEmptyList() throws Exception { |
| final Host host = Host.getHost(0); |
| VM server = host.getVM(0); |
| VM client1 = host.getVM(1); |
| |
| /* Create Server and Client */ |
| cqDUnitTest.createServer(server); |
| final int port = server.invoke(() -> CqQueryUsingPoolDUnitTest.getCacheServerPort()); |
| final String host0 = NetworkUtils.getServerHostName(server.getHost()); |
| |
| final String poolName1 = "pool1"; |
| |
| cqDUnitTest.createPool(client1, poolName1, host0, port); |
| |
| client1.invoke(new CacheSerializableRunnable("test getDurableCQsFromServer for client1") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = null; |
| try { |
| queryService = (PoolManager.find(poolName1)).getQueryService(); |
| } catch (Exception cqe) { |
| Assert.fail("Failed to getCQService.", cqe); |
| } |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(0, list.size()); |
| assertFalse(list.contains("client1DCQ1")); |
| assertFalse(list.contains("client1DCQ2")); |
| } |
| }); |
| |
| cqDUnitTest.closeClient(client1); |
| cqDUnitTest.closeServer(server); |
| } |
| |
| @Test |
| public void testGetDurableCqsFromServer() { |
| disconnectAllFromDS(); |
| |
| final String regionName = "testGetAllDurableCqsFromServer"; |
| final String cq1Name = "testCq1"; |
| final Host host = Host.getHost(0); |
| VM server = host.getVM(0); |
| VM client1 = host.getVM(1); |
| VM client2 = host.getVM(2); |
| |
| // Start server 1 |
| final int server1Port = ((Integer) server |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| |
| // Start client 1 |
| client1.invoke(() -> CacheServerTestUtil.createClientCache( |
| getClientPool(NetworkUtils.getServerHostName(client1.getHost()), server1Port), regionName)); |
| |
| // Start client 2 |
| client2.invoke(() -> CacheServerTestUtil.createClientCache( |
| getClientPool(NetworkUtils.getServerHostName(client2.getHost()), server1Port), regionName)); |
| |
| createClient1CqsAndDurableCqs(client1, regionName); |
| createClient2CqsAndDurableCqs(client2, regionName); |
| |
| client2.invoke(new CacheSerializableRunnable("check durable cqs for client 2") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(4, list.size()); |
| assertTrue(list.contains("client2DCQ1")); |
| assertTrue(list.contains("client2DCQ2")); |
| assertTrue(list.contains("client2DCQ3")); |
| assertTrue(list.contains("client2DCQ4")); |
| } |
| }); |
| |
| client1.invoke(new CacheSerializableRunnable("check durable cqs for client 1") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(2, list.size()); |
| assertTrue(list.contains("client1DCQ1")); |
| assertTrue(list.contains("client1DCQ2")); |
| } |
| }); |
| |
| client1.invoke(() -> CacheServerTestUtil.closeCache()); |
| client2.invoke(() -> CacheServerTestUtil.closeCache()); |
| server.invoke(() -> CacheServerTestUtil.closeCache()); |
| } |
| |
| @Test |
| public void testGetDurableCqsFromServerCycleClients() { |
| disconnectAllFromDS(); |
| |
| final String regionName = "testGetAllDurableCqsFromServerCycleClients"; |
| final Host host = Host.getHost(0); |
| VM server = host.getVM(0); |
| VM client1 = host.getVM(1); |
| VM client2 = host.getVM(2); |
| int timeout = 60000; |
| // Start server 1 |
| final int server1Port = ((Integer) server |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| |
| // Start client 1 |
| client1.invoke(() -> CacheServerTestUtil.createClientCache( |
| getClientPool(NetworkUtils.getServerHostName(client1.getHost()), server1Port), regionName, |
| getDurableClientProperties("client1_dc", timeout))); |
| |
| // Start client 2 |
| client2.invoke(() -> CacheServerTestUtil.createClientCache( |
| getClientPool(NetworkUtils.getServerHostName(client2.getHost()), server1Port), regionName, |
| getDurableClientProperties("client2_dc", timeout))); |
| |
| createClient1CqsAndDurableCqs(client1, regionName); |
| createClient2CqsAndDurableCqs(client2, regionName); |
| |
| cycleDurableClient(client1, "client1_dc", server1Port, regionName, timeout); |
| cycleDurableClient(client2, "client2_dc", server1Port, regionName, timeout); |
| |
| client2.invoke(new CacheSerializableRunnable("check durable cqs for client 2") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(4, list.size()); |
| assertTrue(list.contains("client2DCQ1")); |
| assertTrue(list.contains("client2DCQ2")); |
| assertTrue(list.contains("client2DCQ3")); |
| assertTrue(list.contains("client2DCQ4")); |
| } |
| }); |
| |
| client1.invoke(new CacheSerializableRunnable("check durable cqs for client 1") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(2, list.size()); |
| assertTrue(list.contains("client1DCQ1")); |
| assertTrue(list.contains("client1DCQ2")); |
| } |
| }); |
| |
| client1.invoke(() -> CacheServerTestUtil.closeCache()); |
| client2.invoke(() -> CacheServerTestUtil.closeCache()); |
| server.invoke(() -> CacheServerTestUtil.closeCache()); |
| } |
| |
| @Test |
| public void testGetDurableCqsFromServerCycleClientsAndMoreCqs() { |
| final String regionName = "testGetAllDurableCqsFromServerCycleClients"; |
| final Host host = Host.getHost(0); |
| VM server = host.getVM(0); |
| VM client1 = host.getVM(1); |
| VM client2 = host.getVM(2); |
| int timeout = 60000; |
| // Start server 1 |
| final int server1Port = ((Integer) server |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| |
| // Start client 1 |
| client1.invoke(() -> CacheServerTestUtil.createClientCache( |
| getClientPool(NetworkUtils.getServerHostName(client1.getHost()), server1Port), regionName, |
| getDurableClientProperties("client1_dc", timeout))); |
| |
| // Start client 2 |
| client2.invoke(() -> CacheServerTestUtil.createClientCache( |
| getClientPool(NetworkUtils.getServerHostName(client2.getHost()), server1Port), regionName, |
| getDurableClientProperties("client2_dc", timeout))); |
| |
| // create the test cqs |
| createClient1CqsAndDurableCqs(client1, regionName); |
| createClient2CqsAndDurableCqs(client2, regionName); |
| |
| cycleDurableClient(client1, "client1_dc", server1Port, regionName, timeout); |
| cycleDurableClient(client2, "client2_dc", server1Port, regionName, timeout); |
| |
| client1.invoke(new CacheSerializableRunnable("Register more cq for client 1") { |
| @Override |
| public void run2() throws CacheException { |
| // register the cq's |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| CqAttributesFactory cqAf = new CqAttributesFactory(); |
| CqAttributes attributes = cqAf.create(); |
| try { |
| queryService.newCq("client1MoreDCQ1", "Select * From /" + regionName + " where id = 1", |
| attributes, true).execute(); |
| queryService.newCq("client1MoreDCQ2", "Select * From /" + regionName + " where id = 10", |
| attributes, true).execute(); |
| queryService.newCq("client1MoreNoDC1", "Select * From /" + regionName, attributes, false) |
| .execute(); |
| queryService.newCq("client1MoreNoDC2", "Select * From /" + regionName + " where id = 3", |
| attributes, false).execute(); |
| } catch (RegionNotFoundException e) { |
| fail("failed", e); |
| } catch (CqException e) { |
| fail("failed", e); |
| } catch (CqExistsException e) { |
| fail("failed", e); |
| } |
| } |
| }); |
| |
| client2.invoke(new CacheSerializableRunnable("Register more cq for client 2") { |
| @Override |
| public void run2() throws CacheException { |
| // register the cq's |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| CqAttributesFactory cqAf = new CqAttributesFactory(); |
| CqAttributes attributes = cqAf.create(); |
| try { |
| queryService.newCq("client2MoreDCQ1", "Select * From /" + regionName + " where id = 1", |
| attributes, true).execute(); |
| queryService.newCq("client2MoreDCQ2", "Select * From /" + regionName + " where id = 10", |
| attributes, true).execute(); |
| queryService.newCq("client2MoreDCQ3", "Select * From /" + regionName, attributes, true) |
| .execute(); |
| queryService.newCq("client2MoreDCQ4", "Select * From /" + regionName + " where id = 3", |
| attributes, true).execute(); |
| } catch (RegionNotFoundException e) { |
| fail("failed", e); |
| } catch (CqException e) { |
| fail("failed", e); |
| } catch (CqExistsException e) { |
| fail("failed", e); |
| } |
| } |
| }); |
| |
| // Cycle clients a second time |
| cycleDurableClient(client1, "client1_dc", server1Port, regionName, timeout); |
| cycleDurableClient(client2, "client2_dc", server1Port, regionName, timeout); |
| |
| client2.invoke(new CacheSerializableRunnable("check durable cqs for client 2") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(8, list.size()); |
| assertTrue(list.contains("client2DCQ1")); |
| assertTrue(list.contains("client2DCQ2")); |
| assertTrue(list.contains("client2DCQ3")); |
| assertTrue(list.contains("client2DCQ4")); |
| assertTrue(list.contains("client2MoreDCQ1")); |
| assertTrue(list.contains("client2MoreDCQ2")); |
| assertTrue(list.contains("client2MoreDCQ3")); |
| assertTrue(list.contains("client2MoreDCQ4")); |
| } |
| }); |
| |
| client1.invoke(new CacheSerializableRunnable("check durable cqs for client 1") { |
| @Override |
| public void run2() throws CacheException { |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| List<String> list = null; |
| try { |
| list = queryService.getAllDurableCqsFromServer(); |
| } catch (CqException e) { |
| fail("failed", e); |
| } |
| assertEquals(4, list.size()); |
| assertTrue(list.contains("client1DCQ1")); |
| assertTrue(list.contains("client1DCQ2")); |
| assertTrue(list.contains("client1MoreDCQ1")); |
| assertTrue(list.contains("client1MoreDCQ2")); |
| } |
| }); |
| |
| client1.invoke(() -> CacheServerTestUtil.closeCache()); |
| client2.invoke(() -> CacheServerTestUtil.closeCache()); |
| server.invoke(() -> CacheServerTestUtil.closeCache()); |
| } |
| |
| private Properties getDurableClientProperties(String durableClientId, int durableClientTimeout) { |
| Properties properties = new Properties(); |
| properties.setProperty(MCAST_PORT, "0"); |
| properties.setProperty(LOCATORS, ""); |
| properties.setProperty(DURABLE_CLIENT_ID, durableClientId); |
| properties.setProperty(DURABLE_CLIENT_TIMEOUT, String.valueOf(durableClientTimeout)); |
| return properties; |
| } |
| |
| // helper to create durable cqs to test out getAllDurableCqs functionality |
| private void createClient1CqsAndDurableCqs(VM client, final String regionName) { |
| client.invoke(new CacheSerializableRunnable("Register cq for client 1") { |
| @Override |
| public void run2() throws CacheException { |
| // register the cq's |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| CqAttributesFactory cqAf = new CqAttributesFactory(); |
| CqAttributes attributes = cqAf.create(); |
| try { |
| queryService.newCq("client1DCQ1", "Select * From /" + regionName + " where id = 1", |
| attributes, true).execute(); |
| queryService.newCq("client1DCQ2", "Select * From /" + regionName + " where id = 10", |
| attributes, true).execute(); |
| queryService.newCq("client1NoDC1", "Select * From /" + regionName, attributes, false) |
| .execute(); |
| queryService.newCq("client1NoDC2", "Select * From /" + regionName + " where id = 3", |
| attributes, false).execute(); |
| } catch (RegionNotFoundException e) { |
| fail("failed", e); |
| } catch (CqException e) { |
| fail("failed", e); |
| } catch (CqExistsException e) { |
| fail("failed", e); |
| } |
| } |
| }); |
| } |
| |
| private void createClient2CqsAndDurableCqs(VM client, final String regionName) { |
| client.invoke(new CacheSerializableRunnable("Register cq for client 2") { |
| @Override |
| public void run2() throws CacheException { |
| // register the cq's |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| CqAttributesFactory cqAf = new CqAttributesFactory(); |
| CqAttributes attributes = cqAf.create(); |
| try { |
| queryService.newCq("client2DCQ1", "Select * From /" + regionName + " where id = 1", |
| attributes, true).execute(); |
| queryService.newCq("client2DCQ2", "Select * From /" + regionName + " where id = 10", |
| attributes, true).execute(); |
| queryService.newCq("client2DCQ3", "Select * From /" + regionName, attributes, true) |
| .execute(); |
| queryService.newCq("client2DCQ4", "Select * From /" + regionName + " where id = 3", |
| attributes, true).execute(); |
| } catch (RegionNotFoundException e) { |
| fail("failed", e); |
| } catch (CqException e) { |
| fail("failed", e); |
| } catch (CqExistsException e) { |
| fail("failed", e); |
| } |
| |
| } |
| }); |
| } |
| |
| private void cycleDurableClient(VM client, final String dcName, final int serverPort, |
| final String regionName, final int durableClientTimeout) { |
| client.invoke(new CacheSerializableRunnable("cycle client") { |
| @Override |
| public void run2() throws CacheException { |
| CacheServerTestUtil.closeCache(true); |
| } |
| }); |
| |
| client.invoke(() -> CacheServerTestUtil.createClientCache( |
| getClientPool(NetworkUtils.getServerHostName(client.getHost()), serverPort), regionName, |
| getDurableClientProperties(dcName, durableClientTimeout))); |
| } |
| |
| private CqQuery createCq(String regionName, String cqName) { |
| // Create CQ Attributes. |
| CqAttributesFactory cqAf = new CqAttributesFactory(); |
| |
| // Initialize and set CqListener. |
| CqListener[] cqListeners = {new CqListener() { |
| @Override |
| public void close() {} |
| |
| @Override |
| public void onEvent(CqEvent aCqEvent) {} |
| |
| @Override |
| public void onError(CqEvent aCqEvent) {} |
| }}; |
| cqAf.initCqListeners(cqListeners); |
| CqAttributes cqa = cqAf.create(); |
| |
| // Create cq's |
| // Get the query service for the Pool |
| QueryService queryService = CacheServerTestUtil.getCache().getQueryService(); |
| CqQuery query = null; |
| try { |
| query = queryService.newCq(cqName, "Select * from /" + regionName, cqa); |
| query.execute(); |
| } catch (CqExistsException e) { |
| fail("Could not find specified region:" + regionName + ":", e); |
| } catch (CqException e) { |
| fail("Could not find specified region:" + regionName + ":", e); |
| |
| } catch (RegionNotFoundException e) { |
| fail("Could not find specified region:" + regionName + ":", e); |
| } |
| return query; |
| } |
| |
| private void checkCqStats(String cqName) { |
| QueryService queryService = CacheServerTestUtil.getPool().getQueryService(); |
| CqQueryImpl query = null; |
| query = (CqQueryImpl) queryService.getCq(cqName); |
| if (query != null) { |
| query.getVsdStats(); |
| query.getVsdStats().getNumEvents(); |
| } |
| } |
| |
| private Pool getClientPool(String host, int serverPort) { |
| PoolFactory pf = PoolManager.createFactory(); |
| pf.addServer(host, serverPort).setSubscriptionAckInterval(1).setSubscriptionEnabled(true); |
| return ((PoolFactoryImpl) pf).getPoolAttributes(); |
| } |
| |
| public CacheSerializableRunnable setTestHook() { |
| SerializableRunnable sr = new CacheSerializableRunnable("TestHook") { |
| @Override |
| public void run2() { |
| class CqQueryTestHook implements CqQueryImpl.TestHook { |
| |
| CountDownLatch latch = new CountDownLatch(1); |
| private int numEvents = 0; |
| |
| @Override |
| public void pauseUntilReady() { |
| try { |
| logger.debug("CqQueryTestHook: Going to wait on latch until ready is called."); |
| if (!latch.await(10, TimeUnit.SECONDS)) { |
| fail("query was never unlatched"); |
| } |
| } catch (Exception e) { |
| fail("interrupted", e); |
| } |
| } |
| |
| @Override |
| public void ready() { |
| latch.countDown(); |
| logger.debug("CqQueryTestHook: The latch has been released."); |
| } |
| |
| @Override |
| public int numQueuedEvents() { |
| return numEvents; |
| } |
| |
| @Override |
| public void setEventCount(int count) { |
| logger.debug("CqQueryTestHook: Setting numEVents to: " + count); |
| numEvents = count; |
| } |
| }; |
| CqQueryImpl.testHook = new CqQueryTestHook(); |
| } |
| }; |
| return (CacheSerializableRunnable) sr; |
| } |
| |
| } |
