| /* |
| * 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.internal.cache.wan; |
| |
| import static org.apache.geode.cache.Region.SEPARATOR; |
| import static org.apache.geode.distributed.ConfigurationProperties.CONSERVE_SOCKETS; |
| import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_CIPHERS; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_ENABLED; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE_PASSWORD; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE_TYPE; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_PROTOCOLS; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_REQUIRE_AUTHENTICATION; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_TRUSTSTORE; |
| import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_TRUSTSTORE_PASSWORD; |
| import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER; |
| import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_HTTP_PORT; |
| import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_PORT; |
| import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_START; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.distributed.ConfigurationProperties.OFF_HEAP_MEMORY_SIZE; |
| import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR; |
| import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort; |
| import static org.apache.geode.test.awaitility.GeodeAwaitility.await; |
| import static org.apache.geode.test.dunit.Host.getHost; |
| import static org.apache.geode.test.util.ResourceUtils.createTempFileFromResource; |
| import static org.assertj.core.api.Assertions.assertThat; |
| import static org.hamcrest.Matchers.greaterThan; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.io.Serializable; |
| import java.net.InetAddress; |
| import java.net.InetSocketAddress; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.StringTokenizer; |
| import java.util.concurrent.BlockingQueue; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ConcurrentSkipListSet; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.ThreadFactory; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.stream.Collectors; |
| import java.util.stream.Stream; |
| |
| import javax.management.ObjectName; |
| |
| import org.apache.commons.io.FileUtils; |
| import org.apache.logging.log4j.Logger; |
| import org.junit.Before; |
| import org.junit.BeforeClass; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.cache.AttributesMutator; |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheClosedException; |
| import org.apache.geode.cache.CacheFactory; |
| import org.apache.geode.cache.CacheListener; |
| import org.apache.geode.cache.CacheTransactionManager; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.DiskStore; |
| import org.apache.geode.cache.DiskStoreFactory; |
| import org.apache.geode.cache.EntryEvent; |
| import org.apache.geode.cache.PartitionAttributesFactory; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionDestroyedException; |
| import org.apache.geode.cache.RegionFactory; |
| import org.apache.geode.cache.RegionShortcut; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.TransactionException; |
| import org.apache.geode.cache.asyncqueue.AsyncEventListener; |
| import org.apache.geode.cache.asyncqueue.AsyncEventQueue; |
| import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory; |
| import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl; |
| import org.apache.geode.cache.client.Pool; |
| import org.apache.geode.cache.client.PoolManager; |
| import org.apache.geode.cache.client.internal.ConnectionStats; |
| import org.apache.geode.cache.client.internal.LocatorDiscoveryCallbackAdapter; |
| import org.apache.geode.cache.client.internal.PoolImpl; |
| import org.apache.geode.cache.client.internal.locator.wan.LocatorMembershipListener; |
| import org.apache.geode.cache.persistence.PartitionOfflineException; |
| import org.apache.geode.cache.server.CacheServer; |
| import org.apache.geode.cache.util.CacheListenerAdapter; |
| import org.apache.geode.cache.wan.GatewayEventFilter; |
| import org.apache.geode.cache.wan.GatewayQueueEvent; |
| import org.apache.geode.cache.wan.GatewayReceiver; |
| import org.apache.geode.cache.wan.GatewayReceiverFactory; |
| import org.apache.geode.cache.wan.GatewaySender; |
| import org.apache.geode.cache.wan.GatewaySender.OrderPolicy; |
| import org.apache.geode.cache.wan.GatewaySenderFactory; |
| import org.apache.geode.cache.wan.GatewayTransportFilter; |
| import org.apache.geode.cache30.CacheTestCase; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.Locator; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.InternalLocator; |
| import org.apache.geode.distributed.internal.ServerLocation; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.admin.remote.DistributionLocatorId; |
| import org.apache.geode.internal.cache.BucketRegion; |
| import org.apache.geode.internal.cache.CacheServerImpl; |
| import org.apache.geode.internal.cache.CustomerIDPartitionResolver; |
| import org.apache.geode.internal.cache.ForceReattemptException; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.InternalCacheBuilder; |
| import org.apache.geode.internal.cache.InternalRegion; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.PoolStats; |
| import org.apache.geode.internal.cache.RegionQueue; |
| import org.apache.geode.internal.cache.execute.data.CustId; |
| import org.apache.geode.internal.cache.execute.data.Customer; |
| import org.apache.geode.internal.cache.execute.data.Order; |
| import org.apache.geode.internal.cache.execute.data.OrderId; |
| import org.apache.geode.internal.cache.execute.data.Shipment; |
| import org.apache.geode.internal.cache.execute.data.ShipmentId; |
| import org.apache.geode.internal.cache.partitioned.BecomePrimaryBucketMessage; |
| import org.apache.geode.internal.cache.partitioned.BecomePrimaryBucketMessage.BecomePrimaryBucketResponse; |
| import org.apache.geode.internal.cache.partitioned.PRLocallyDestroyedException; |
| import org.apache.geode.internal.cache.tier.sockets.CacheServerStats; |
| import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil; |
| import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderEventProcessor; |
| import org.apache.geode.internal.cache.wan.parallel.ConcurrentParallelGatewaySenderQueue; |
| import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderEventProcessor; |
| import org.apache.geode.internal.cache.wan.parallel.ParallelGatewaySenderQueue; |
| import org.apache.geode.internal.cache.wan.serial.ConcurrentSerialGatewaySenderEventProcessor; |
| import org.apache.geode.internal.cache.wan.serial.SerialGatewaySenderQueue; |
| import org.apache.geode.logging.internal.log4j.api.LogService; |
| import org.apache.geode.management.AsyncEventQueueMXBean; |
| import org.apache.geode.management.DistributedSystemMXBean; |
| import org.apache.geode.management.GatewayReceiverMXBean; |
| import org.apache.geode.management.GatewaySenderMXBean; |
| import org.apache.geode.management.MBeanUtil; |
| import org.apache.geode.management.ManagementService; |
| import org.apache.geode.management.RegionMXBean; |
| import org.apache.geode.management.internal.SystemManagementService; |
| import org.apache.geode.pdx.SimpleClass; |
| import org.apache.geode.pdx.SimpleClass1; |
| import org.apache.geode.test.dunit.Assert; |
| import org.apache.geode.test.dunit.AsyncInvocation; |
| import org.apache.geode.test.dunit.DistributedTestCase; |
| 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.SerializableRunnable; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.dunit.WaitCriterion; |
| import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase; |
| import org.apache.geode.test.junit.categories.WanTest; |
| |
| @Category({WanTest.class}) |
| public class WANTestBase extends DistributedTestCase { |
| |
| protected static Cache cache; |
| protected static Region region; |
| |
| protected static PartitionedRegion customerRegion; |
| protected static PartitionedRegion orderRegion; |
| protected static PartitionedRegion shipmentRegion; |
| |
| protected static final String customerRegionName = "CUSTOMER"; |
| protected static final String orderRegionName = "ORDER"; |
| protected static final String shipmentRegionName = "SHIPMENT"; |
| |
| protected static VM vm0; |
| protected static VM vm1; |
| protected static VM vm2; |
| protected static VM vm3; |
| protected static VM vm4; |
| protected static VM vm5; |
| protected static VM vm6; |
| protected static VM vm7; |
| |
| protected static QueueListener listener1; |
| protected static QueueListener listener2; |
| |
| protected static AsyncEventListener eventListener1; |
| protected static AsyncEventListener eventListener2; |
| |
| private static final long MAX_WAIT = 10000; |
| |
| protected static GatewayEventFilter eventFilter; |
| |
| protected static List<Integer> dispatcherThreads = new ArrayList<Integer>(Arrays.asList(1, 3, 5)); |
| // this will be set for each test method run with one of the values from above list |
| protected static int numDispatcherThreadsForTheRun = 1; |
| |
| private static final Logger logger = LogService.getLogger(); |
| |
| @BeforeClass |
| public static void beforeClassWANTestBase() throws Exception { |
| vm0 = getHost(0).getVM(0); |
| vm1 = getHost(0).getVM(1); |
| vm2 = getHost(0).getVM(2); |
| vm3 = getHost(0).getVM(3); |
| vm4 = getHost(0).getVM(4); |
| vm5 = getHost(0).getVM(5); |
| vm6 = getHost(0).getVM(6); |
| vm7 = getHost(0).getVM(7); |
| } |
| |
| @Before |
| public void setUpWANTestBase() throws Exception { |
| shuffleNumDispatcherThreads(); |
| Invoke.invokeInEveryVM(() -> setNumDispatcherThreadsForTheRun(dispatcherThreads.get(0))); |
| IgnoredException.addIgnoredException("Connection refused"); |
| IgnoredException.addIgnoredException("Software caused connection abort"); |
| IgnoredException.addIgnoredException("Connection reset"); |
| postSetUpWANTestBase(); |
| } |
| |
| protected void postSetUpWANTestBase() throws Exception { |
| // nothing |
| } |
| |
| public static void shuffleNumDispatcherThreads() { |
| Collections.shuffle(dispatcherThreads); |
| } |
| |
| public static void setNumDispatcherThreadsForTheRun(int numThreads) { |
| numDispatcherThreadsForTheRun = numThreads; |
| } |
| |
| public static void stopOldLocator() { |
| if (Locator.hasLocator()) { |
| Locator.getLocator().stop(); |
| } |
| } |
| |
| public static void createLocator(int dsId, int port, Set<String> localLocatorsList, |
| Set<String> remoteLocatorsList) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId); |
| StringBuffer localLocatorBuffer = new StringBuffer(localLocatorsList.toString()); |
| localLocatorBuffer.deleteCharAt(0); |
| localLocatorBuffer.deleteCharAt(localLocatorBuffer.lastIndexOf("]")); |
| String localLocator = localLocatorBuffer.toString(); |
| localLocator = localLocator.replace(" ", ""); |
| |
| props.setProperty(LOCATORS, localLocator); |
| props.setProperty(START_LOCATOR, |
| "localhost[" + port + "],server=true,peer=true,hostname-for-clients=localhost"); |
| StringBuffer remoteLocatorBuffer = new StringBuffer(remoteLocatorsList.toString()); |
| remoteLocatorBuffer.deleteCharAt(0); |
| remoteLocatorBuffer.deleteCharAt(remoteLocatorBuffer.lastIndexOf("]")); |
| String remoteLocator = remoteLocatorBuffer.toString(); |
| remoteLocator = remoteLocator.replace(" ", ""); |
| props.setProperty(REMOTE_LOCATORS, remoteLocator); |
| test.startLocatorDistributedSystem(props); |
| } |
| |
| private void startLocator(int dsId, int locatorPort, int startLocatorPort, int remoteLocPort, |
| boolean startServerLocator) { |
| Properties props = getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId); |
| props.setProperty(LOCATORS, "localhost[" + locatorPort + "]"); |
| props.setProperty(START_LOCATOR, "localhost[" + startLocatorPort + "],server=" |
| + startServerLocator + ",peer=true,hostname-for-clients=localhost"); |
| if (remoteLocPort != -1) { |
| props.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocPort + "]"); |
| } |
| startLocatorDistributedSystem(props); |
| } |
| |
| private void startLocatorDistributedSystem(Properties props) { |
| // Start start the locator with a LOCATOR_DM_TYPE and not a NORMAL_DM_TYPE |
| System.setProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE, "true"); |
| try { |
| getSystem(props); |
| } finally { |
| System.clearProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE); |
| } |
| } |
| |
| public static Integer createFirstLocatorWithDSId(int dsId) { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| int port = getRandomAvailableTCPPort(); |
| test.startLocator(dsId, port, port, -1, true); |
| return port; |
| } |
| |
| public static Integer createFirstPeerLocator(int dsId) { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| int port = getRandomAvailableTCPPort(); |
| test.startLocator(dsId, port, port, -1, false); |
| return port; |
| } |
| |
| public static Integer createSecondLocator(int dsId, int locatorPort) { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| int port = getRandomAvailableTCPPort(); |
| test.startLocator(dsId, locatorPort, port, -1, true); |
| return port; |
| } |
| |
| public static Integer createSecondPeerLocator(int dsId, int locatorPort) { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| int port = getRandomAvailableTCPPort(); |
| test.startLocator(dsId, locatorPort, port, -1, false); |
| return port; |
| } |
| |
| public static Integer createFirstRemoteLocator(int dsId, int remoteLocPort) { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| int port = getRandomAvailableTCPPort(); |
| test.startLocator(dsId, port, port, remoteLocPort, true); |
| return port; |
| } |
| |
| public static void bringBackLocatorOnOldPort(int dsId, int remoteLocPort, int oldPort) { |
| WANTestBase test = new WANTestBase(); |
| test.startLocator(dsId, oldPort, oldPort, remoteLocPort, true); |
| } |
| |
| |
| public static Integer createFirstRemotePeerLocator(int dsId, int remoteLocPort) { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| int port = getRandomAvailableTCPPort(); |
| test.startLocator(dsId, port, port, remoteLocPort, false); |
| return port; |
| } |
| |
| public static Integer createSecondRemoteLocator(int dsId, int localPort, int remoteLocPort) { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| int port = getRandomAvailableTCPPort(); |
| test.startLocator(dsId, localPort, port, remoteLocPort, true); |
| return port; |
| } |
| |
| public static Integer createSecondRemoteLocatorWithAPI(int dsId, int localPort, int remoteLocPort, |
| String hostnameForClients) throws IOException { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId); |
| props.setProperty(LOCATORS, "localhost[" + localPort + "]"); |
| props.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocPort + "]"); |
| Locator locator = Locator.startLocatorAndDS(0, null, InetAddress.getByName("0.0.0.0"), props, |
| true, true, hostnameForClients); |
| return locator.getPort(); |
| } |
| |
| public static Integer createSecondRemotePeerLocator(int dsId, int localPort, int remoteLocPort) { |
| stopOldLocator(); |
| WANTestBase test = new WANTestBase(); |
| int port = getRandomAvailableTCPPort(); |
| test.startLocator(dsId, localPort, port, remoteLocPort, false); |
| return port; |
| } |
| |
| public static int createReceiverInSecuredCache() { |
| GatewayReceiverFactory fact = WANTestBase.cache.createGatewayReceiverFactory(); |
| int port = getRandomAvailableTCPPort(); |
| fact.setStartPort(port); |
| fact.setEndPort(port); |
| fact.setManualStart(true); |
| GatewayReceiver receiver = fact.create(); |
| try { |
| receiver.start(); |
| } catch (IOException e) { |
| e.printStackTrace(); |
| org.apache.geode.test.dunit.Assert.fail("Failed to start GatewayReceiver on port " + port, e); |
| } |
| return port; |
| } |
| |
| public static void createReplicatedRegion(String regionName, String senderIds, Boolean offHeap) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| IgnoredException exp2 = |
| IgnoredException.addIgnoredException(GatewaySenderException.class.getName()); |
| try { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.REPLICATE); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| exp2.remove(); |
| } |
| } |
| |
| |
| public static void createReplicatedProxyRegion(String regionName, String senderIds, |
| Boolean offHeap) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| IgnoredException exp2 = |
| IgnoredException.addIgnoredException(GatewaySenderException.class.getName()); |
| try { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.REPLICATE_PROXY); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| exp2.remove(); |
| } |
| } |
| |
| public static void createNormalRegion(String regionName, String senderIds) { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.LOCAL); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } |
| |
| public static void createPersistentReplicatedRegion(String regionName, String senderIds, |
| Boolean offHeap) { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.REPLICATE_PERSISTENT); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } |
| |
| public static void createReplicatedRegionWithAsyncEventQueue(String regionName, |
| String asyncQueueIds, Boolean offHeap) { |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.REPLICATE); |
| if (asyncQueueIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(asyncQueueIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String asyncQueueId = tokenizer.nextToken(); |
| fact.addAsyncEventQueueId(asyncQueueId); |
| } |
| } |
| |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp1.remove(); |
| } |
| } |
| |
| public static void createReplicatedRegionWithSenderAndAsyncEventQueue(String regionName, |
| String senderIds, String asyncChannelId, Boolean offHeap) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.REPLICATE); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| |
| fact.setOffHeap(offHeap); |
| fact.addAsyncEventQueueId(asyncChannelId); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| } |
| } |
| |
| public static void createReplicatedRegion(String regionName, String senderIds, Scope scope, |
| DataPolicy policy, Boolean offHeap) { |
| RegionFactory fact = cache.createRegionFactory(); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| |
| fact.setDataPolicy(policy); |
| fact.setScope(scope); |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } |
| |
| public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| String diskStoreName, boolean isDiskSynchronous) { |
| |
| if (diskStoreName != null) { |
| File directory = new File( |
| asyncChannelId + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| directory.mkdir(); |
| File[] dirs1 = new File[] {directory}; |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| dsf.setDiskDirs(dirs1); |
| dsf.create(diskStoreName); |
| } |
| |
| AsyncEventListener asyncEventListener = new MyAsyncEventListener(); |
| |
| AsyncEventQueueFactory factory = cache.createAsyncEventQueueFactory(); |
| factory.setBatchSize(batchSize); |
| factory.setPersistent(isPersistent); |
| factory.setDiskStoreName(diskStoreName); |
| factory.setDiskSynchronous(isDiskSynchronous); |
| factory.setBatchConflationEnabled(isConflation); |
| factory.setMaximumQueueMemory(maxMemory); |
| factory.setParallel(isParallel); |
| // set dispatcher threads |
| factory.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| factory.create(asyncChannelId, asyncEventListener); |
| } |
| |
| public static void createPartitionedRegion(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap) { |
| createPartitionedRegion(regionName, senderIds, redundantCopies, totalNumBuckets, offHeap, |
| RegionShortcut.PARTITION); |
| } |
| |
| public static void createPartitionedRegion(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap, RegionShortcut shortcut) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(PartitionOfflineException.class.getName()); |
| try { |
| RegionFactory fact = cache.createRegionFactory(shortcut); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setRedundantCopies(redundantCopies); |
| pfact.setRecoveryDelay(0); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| // TODO:OFFHEAP: add offheap flavor |
| public static void createPartitionedRegionWithPersistence(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(PartitionOfflineException.class.getName()); |
| try { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.PARTITION_PERSISTENT); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setRedundantCopies(redundantCopies); |
| pfact.setRecoveryDelay(0); |
| fact.setPartitionAttributes(pfact.create()); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| public static void createColocatedPartitionedRegion(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets, String colocatedWith) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(PartitionOfflineException.class.getName()); |
| try { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.PARTITION); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setRedundantCopies(redundantCopies); |
| pfact.setRecoveryDelay(0); |
| pfact.setColocatedWith(colocatedWith); |
| fact.setPartitionAttributes(pfact.create()); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| public static void addSenderThroughAttributesMutator(String regionName, String senderIds) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| AttributesMutator mutator = r.getAttributesMutator(); |
| mutator.addGatewaySenderId(senderIds); |
| } |
| |
| public static void addAsyncEventQueueThroughAttributesMutator(String regionName, String queueId) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| AttributesMutator mutator = r.getAttributesMutator(); |
| mutator.addAsyncEventQueueId(queueId); |
| } |
| |
| public static void createPartitionedRegionAsAccessor(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets) { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.PARTITION_PROXY); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setRedundantCopies(redundantCopies); |
| fact.setPartitionAttributes(pfact.create()); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } |
| |
| public static void createPartitionedRegionWithSerialParallelSenderIds(String regionName, |
| String serialSenderIds, String parallelSenderIds, String colocatedWith, Boolean offHeap) { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.PARTITION); |
| if (serialSenderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(serialSenderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| if (parallelSenderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(parallelSenderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setColocatedWith(colocatedWith); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } |
| |
| public static void createPersistentPartitionedRegion(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap) { |
| |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(PartitionOfflineException.class.getName()); |
| try { |
| |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.PARTITION_PERSISTENT); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setRedundantCopies(redundantCopies); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| public static void createCustomerOrderShipmentPartitionedRegion(String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.PARTITION); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| |
| PartitionAttributesFactory paf = new PartitionAttributesFactory(); |
| paf.setRedundantCopies(redundantCopies).setTotalNumBuckets(totalNumBuckets) |
| .setPartitionResolver(new CustomerIDPartitionResolver("CustomerIDPartitionResolver")); |
| fact.setPartitionAttributes(paf.create()); |
| fact.setOffHeap(offHeap); |
| customerRegion = |
| (PartitionedRegion) fact.create(customerRegionName); |
| assertNotNull(customerRegion); |
| logger.info("Partitioned Region CUSTOMER created Successfully :" + customerRegion.toString()); |
| |
| paf = new PartitionAttributesFactory(); |
| paf.setRedundantCopies(redundantCopies).setTotalNumBuckets(totalNumBuckets) |
| .setColocatedWith(customerRegionName) |
| .setPartitionResolver(new CustomerIDPartitionResolver("CustomerIDPartitionResolver")); |
| fact = cache.createRegionFactory(RegionShortcut.PARTITION); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| fact.setPartitionAttributes(paf.create()); |
| fact.setOffHeap(offHeap); |
| orderRegion = |
| (PartitionedRegion) fact.create(orderRegionName); |
| assertNotNull(orderRegion); |
| logger.info("Partitioned Region ORDER created Successfully :" + orderRegion.toString()); |
| |
| paf = new PartitionAttributesFactory(); |
| paf.setRedundantCopies(redundantCopies).setTotalNumBuckets(totalNumBuckets) |
| .setColocatedWith(orderRegionName) |
| .setPartitionResolver(new CustomerIDPartitionResolver("CustomerIDPartitionResolver")); |
| fact = cache.createRegionFactory(RegionShortcut.PARTITION); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| fact.setPartitionAttributes(paf.create()); |
| fact.setOffHeap(offHeap); |
| shipmentRegion = |
| (PartitionedRegion) fact.create(shipmentRegionName); |
| assertNotNull(shipmentRegion); |
| logger.info("Partitioned Region SHIPMENT created Successfully :" + shipmentRegion.toString()); |
| } finally { |
| exp.remove(); |
| } |
| } |
| |
| public static void createColocatedPartitionedRegions(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap) { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.PARTITION); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setRedundantCopies(redundantCopies); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| |
| pfact.setColocatedWith(r.getName()); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r1 = fact.create(regionName + "_child1"); |
| assertNotNull(r1); |
| |
| Region r2 = fact.create(regionName + "_child2"); |
| assertNotNull(r2); |
| } |
| |
| public static void createColocatedPartitionedRegions2(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets, Boolean offHeap) { |
| RegionFactory fact = cache.createRegionFactory(RegionShortcut.PARTITION); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setRedundantCopies(redundantCopies); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r = fact.create(regionName); |
| assertNotNull(r); |
| |
| fact = cache.createRegionFactory(RegionShortcut.PARTITION); |
| pfact.setColocatedWith(r.getName()); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r1 = fact.create(regionName + "_child1"); |
| assertNotNull(r1); |
| |
| Region r2 = fact.create(regionName + "_child2"); |
| assertNotNull(r2); |
| } |
| |
| public static void createCacheInVMs(Integer locatorPort, VM... vms) { |
| for (VM vm : vms) { |
| vm.invoke(() -> createCache(locatorPort)); |
| } |
| } |
| |
| public static void addListenerToSleepAfterCreateEvent(int milliSeconds, final String regionName) { |
| cache.getRegion(regionName).getAttributesMutator() |
| .addCacheListener(new CacheListenerAdapter<Object, Object>() { |
| @Override |
| public void afterCreate(final EntryEvent<Object, Object> event) { |
| try { |
| Thread.sleep(milliSeconds); |
| } catch (InterruptedException e) { |
| Thread.currentThread().interrupt(); |
| } |
| } |
| }); |
| } |
| |
| private static CacheListener myListener; |
| |
| public static void removeCacheListener() { |
| cache.getRegion(getTestMethodName() + "_RR_1").getAttributesMutator() |
| .removeCacheListener(myListener); |
| |
| } |
| |
| |
| public static void createCache(Integer locPort) { |
| createCache(false, locPort); |
| } |
| |
| public static void createManagementCache(Integer locPort) { |
| createCache(true, locPort); |
| } |
| |
| public static void createCacheConserveSockets(Boolean conserveSockets, Integer locPort) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| props.setProperty(CONSERVE_SOCKETS, conserveSockets.toString()); |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| } |
| |
| protected static void createCache(boolean management, Integer locPort) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| if (management) { |
| props.setProperty(JMX_MANAGER, "true"); |
| props.setProperty(JMX_MANAGER_START, "false"); |
| props.setProperty(JMX_MANAGER_PORT, "0"); |
| props.setProperty(JMX_MANAGER_HTTP_PORT, "0"); |
| } |
| props.setProperty(MCAST_PORT, "0"); |
| String logLevel = System.getProperty(LOG_LEVEL, "info"); |
| props.setProperty(LOG_LEVEL, logLevel); |
| props.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| } |
| |
| protected static void createCacheWithSSL(Integer locPort) { |
| WANTestBase test = new WANTestBase(); |
| |
| boolean gatewaySslenabled = true; |
| String gatewaySslprotocols = "any"; |
| String gatewaySslciphers = "any"; |
| boolean gatewaySslRequireAuth = true; |
| |
| Properties gemFireProps = test.getDistributedSystemProperties(); |
| gemFireProps.put(LOG_LEVEL, LogWriterUtils.getDUnitLogLevel()); |
| gemFireProps.put(GATEWAY_SSL_ENABLED, String.valueOf(gatewaySslenabled)); |
| gemFireProps.put(GATEWAY_SSL_PROTOCOLS, gatewaySslprotocols); |
| gemFireProps.put(GATEWAY_SSL_CIPHERS, gatewaySslciphers); |
| gemFireProps.put(GATEWAY_SSL_REQUIRE_AUTHENTICATION, String.valueOf(gatewaySslRequireAuth)); |
| |
| gemFireProps.put(GATEWAY_SSL_KEYSTORE_TYPE, "jks"); |
| // this uses the default.keystore which is all jdk compliant in geode-dunit module |
| gemFireProps.put(GATEWAY_SSL_KEYSTORE, createTempFileFromResource(WANTestBase.class, |
| "/org/apache/geode/cache/client/internal/default.keystore").getAbsolutePath()); |
| gemFireProps.put(GATEWAY_SSL_KEYSTORE_PASSWORD, "password"); |
| gemFireProps.put(GATEWAY_SSL_TRUSTSTORE, createTempFileFromResource(WANTestBase.class, |
| "/org/apache/geode/cache/client/internal/default.keystore").getAbsolutePath()); |
| gemFireProps.put(GATEWAY_SSL_TRUSTSTORE_PASSWORD, "password"); |
| |
| gemFireProps.setProperty(MCAST_PORT, "0"); |
| gemFireProps.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| |
| logger.info("Starting cache ds with following properties \n" + gemFireProps); |
| |
| InternalDistributedSystem ds = test.getSystem(gemFireProps); |
| cache = CacheFactory.create(ds); |
| } |
| |
| public static void createCache_PDX(Integer locPort) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| InternalDistributedSystem ds = test.getSystem(props); |
| |
| cache = new InternalCacheBuilder(props) |
| .setPdxPersistent(true) |
| .setPdxDiskStore("PDX_TEST") |
| .setIsExistingOk(false) |
| .create(ds); |
| |
| File pdxDir = new File(CacheTestCase.getDiskDir(), "pdx"); |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| File[] dirs1 = new File[] {pdxDir}; |
| dsf.setDiskDirs(dirs1).setMaxOplogSize(1).create("PDX_TEST"); |
| } |
| |
| public static void createCache(Integer locPort1, Integer locPort2) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, "localhost[" + locPort1 + "],localhost[" + locPort2 + "]"); |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| } |
| |
| public static void createCacheWithoutLocator(Integer mCastPort) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "" + mCastPort); |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| } |
| |
| /** |
| * Method that creates a cache server |
| * |
| * @return Integer Port on which the server is started. |
| */ |
| public static Integer createCacheServer() { |
| CacheServer server1 = cache.addCacheServer(); |
| assertNotNull(server1); |
| server1.setPort(0); |
| try { |
| server1.start(); |
| } catch (IOException e) { |
| org.apache.geode.test.dunit.Assert.fail("Failed to start the Server", e); |
| } |
| assertTrue(server1.isRunning()); |
| |
| return new Integer(server1.getPort()); |
| } |
| |
| /** |
| * Returns a Map that contains the count for number of cache server and number of Receivers. |
| * |
| */ |
| public static Map getCacheServers() { |
| List cacheServers = cache.getCacheServers(); |
| |
| Map cacheServersMap = new HashMap(); |
| Iterator itr = cacheServers.iterator(); |
| int bridgeServerCounter = 0; |
| int receiverServerCounter = 0; |
| while (itr.hasNext()) { |
| CacheServerImpl cacheServer = (CacheServerImpl) itr.next(); |
| if (cacheServer.getAcceptor().isGatewayReceiver()) { |
| receiverServerCounter++; |
| } else { |
| bridgeServerCounter++; |
| } |
| } |
| cacheServersMap.put("BridgeServer", bridgeServerCounter); |
| cacheServersMap.put("ReceiverServer", receiverServerCounter); |
| return cacheServersMap; |
| } |
| |
| public static void startSenderInVMs(String senderId, VM... vms) { |
| for (VM vm : vms) { |
| vm.invoke(() -> startSender(senderId)); |
| } |
| } |
| |
| public static void startSenderInVMsAsync(String senderId, VM... vms) { |
| List<AsyncInvocation> tasks = new LinkedList<>(); |
| for (VM vm : vms) { |
| tasks.add(vm.invokeAsync(() -> startSender(senderId))); |
| } |
| for (AsyncInvocation invocation : tasks) { |
| try { |
| invocation.await(); |
| } catch (InterruptedException e) { |
| fail("Starting senders was interrupted"); |
| } |
| } |
| } |
| |
| |
| public static void startSenderwithCleanQueuesInVMsAsync(String senderId, VM... vms) { |
| List<AsyncInvocation> tasks = new LinkedList<>(); |
| for (VM vm : vms) { |
| tasks.add(vm.invokeAsync(() -> startSenderwithCleanQueues(senderId))); |
| } |
| for (AsyncInvocation invocation : tasks) { |
| try { |
| invocation.await(); |
| } catch (InterruptedException e) { |
| fail("Starting senders was interrupted"); |
| } |
| } |
| } |
| |
| public static void startSender(String senderId) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| try { |
| GatewaySender sender = getGatewaySender(senderId); |
| sender.start(); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| exln.remove(); |
| } |
| |
| } |
| |
| public static void startSenderwithCleanQueues(String senderId) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| try { |
| GatewaySender sender = getGatewaySender(senderId); |
| sender.startWithCleanQueue(); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| exln.remove(); |
| } |
| |
| } |
| |
| public static void enableConflation(String senderId) { |
| AbstractGatewaySender sender = (AbstractGatewaySender) getGatewaySender(senderId); |
| sender.test_setBatchConflationEnabled(true); |
| } |
| |
| public static Map getSenderToReceiverConnectionInfo(String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| Map connectionInfo = null; |
| if (!sender.isParallel() && ((AbstractGatewaySender) sender).isPrimary()) { |
| connectionInfo = new HashMap(); |
| GatewaySenderEventDispatcher dispatcher = |
| ((AbstractGatewaySender) sender).getEventProcessor().getDispatcher(); |
| if (dispatcher instanceof GatewaySenderEventRemoteDispatcher) { |
| ServerLocation serverLocation = |
| ((GatewaySenderEventRemoteDispatcher) dispatcher).getConnection(false).getServer(); |
| connectionInfo.put("serverHost", serverLocation.getHostName()); |
| connectionInfo.put("serverPort", serverLocation.getPort()); |
| |
| } |
| } |
| return connectionInfo; |
| } |
| |
| public static void moveAllPrimaryBuckets(String senderId, final DistributedMember destination, |
| final String regionName) { |
| |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId); |
| final RegionQueue regionQueue; |
| regionQueue = sender.getQueues().toArray(new RegionQueue[1])[0]; |
| if (sender.isParallel()) { |
| ConcurrentParallelGatewaySenderQueue parallelGatewaySenderQueue = |
| (ConcurrentParallelGatewaySenderQueue) regionQueue; |
| PartitionedRegion prQ = |
| parallelGatewaySenderQueue.getRegions().toArray(new PartitionedRegion[1])[0]; |
| |
| Set<Integer> primaryBucketIds = prQ.getDataStore().getAllLocalPrimaryBucketIds(); |
| for (int bid : primaryBucketIds) { |
| movePrimary(destination, regionName, bid); |
| } |
| |
| // double check after moved all primary buckets |
| primaryBucketIds = prQ.getDataStore().getAllLocalPrimaryBucketIds(); |
| assertTrue(primaryBucketIds.isEmpty()); |
| } |
| } |
| |
| public static void movePrimary(final DistributedMember destination, final String regionName, |
| final int bucketId) { |
| PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName); |
| |
| BecomePrimaryBucketResponse response = BecomePrimaryBucketMessage |
| .send((InternalDistributedMember) destination, region, bucketId, true); |
| assertNotNull(response); |
| assertTrue(response.waitForResponse()); |
| } |
| |
| public static int getSecondaryQueueSizeInStats(String senderId) { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId); |
| GatewaySenderStats statistics = sender.getStatistics(); |
| return statistics.getSecondaryEventQueueSize(); |
| } |
| |
| public static void checkQueueSizeInStats(String senderId, final int expectedQueueSize) { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId); |
| GatewaySenderStats statistics = sender.getStatistics(); |
| await() |
| .untilAsserted(() -> assertEquals("Expected queue size: " + expectedQueueSize |
| + " but actual size: " + statistics.getEventQueueSize(), expectedQueueSize, |
| statistics.getEventQueueSize())); |
| } |
| |
| public static void checkConnectionStats(String senderId) { |
| AbstractGatewaySender sender = |
| (AbstractGatewaySender) CacheFactory.getAnyInstance().getGatewaySender(senderId); |
| assertNotNull(sender); |
| |
| Collection statsCollection = sender.getProxy().getEndpointManager().getAllStats().values(); |
| assertTrue(statsCollection.iterator().next() instanceof ConnectionStats); |
| } |
| |
| public static List<Integer> getSenderStats(String senderId, final int expectedQueueSize) { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId); |
| GatewaySenderStats statistics = sender.getStatistics(); |
| if (expectedQueueSize != -1) { |
| final RegionQueue regionQueue; |
| regionQueue = sender.getQueues().toArray(new RegionQueue[1])[0]; |
| if (sender.isParallel()) { |
| ConcurrentParallelGatewaySenderQueue parallelGatewaySenderQueue = |
| (ConcurrentParallelGatewaySenderQueue) regionQueue; |
| PartitionedRegion pr = |
| parallelGatewaySenderQueue.getRegions().toArray(new PartitionedRegion[1])[0]; |
| } |
| await() |
| .untilAsserted(() -> assertEquals("Expected queue entries: " + expectedQueueSize |
| + " but actual entries: " + regionQueue.size(), expectedQueueSize, |
| regionQueue.size())); |
| } |
| ArrayList<Integer> stats = new ArrayList<Integer>(); |
| stats.add(statistics.getEventQueueSize()); |
| stats.add(statistics.getEventsReceived()); |
| stats.add(statistics.getEventsQueued()); |
| stats.add(statistics.getEventsDistributed()); |
| stats.add(statistics.getBatchesDistributed()); |
| stats.add(statistics.getBatchesRedistributed()); |
| stats.add(statistics.getEventsFiltered()); |
| stats.add(statistics.getEventsNotQueuedConflated()); |
| stats.add(statistics.getEventsConflatedFromBatches()); |
| stats.add(statistics.getConflationIndexesMapSize()); |
| stats.add(statistics.getSecondaryEventQueueSize()); |
| stats.add(statistics.getEventsProcessedByPQRM()); |
| stats.add(statistics.getEventsExceedingAlertThreshold()); |
| return stats; |
| } |
| |
| public static int getGatewaySenderPoolDisconnects(String senderId) { |
| AbstractGatewaySender sender = |
| (AbstractGatewaySender) CacheFactory.getAnyInstance().getGatewaySender(senderId); |
| assertNotNull(sender); |
| |
| PoolStats poolStats = sender.getProxy().getStats(); |
| |
| return poolStats.getDisConnects(); |
| } |
| |
| public static List<Integer> getSenderStatsForDroppedEvents(String senderId) { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId); |
| GatewaySenderStats statistics = sender.getStatistics(); |
| ArrayList<Integer> stats = new ArrayList<Integer>(); |
| int eventNotQueued = statistics.getEventsDroppedDueToPrimarySenderNotRunning(); |
| if (eventNotQueued > 0) { |
| logger |
| .info("Found " + eventNotQueued + " events dropped due to primary sender is not running"); |
| } |
| stats.add(eventNotQueued); |
| stats.add(statistics.getEventsNotQueued()); |
| stats.add(statistics.getEventsNotQueuedConflated()); |
| return stats; |
| } |
| |
| public static void checkQueueStats(String senderId, final int queueSize, final int eventsReceived, |
| final int eventsQueued, final int eventsDistributed) { |
| GatewaySenderStats statistics = getGatewaySenderStats(senderId); |
| assertEquals(queueSize, statistics.getEventQueueSize()); |
| assertEquals(eventsReceived, statistics.getEventsReceived()); |
| assertEquals(eventsQueued, statistics.getEventsQueued()); |
| assert (statistics.getEventsDistributed() >= eventsDistributed); |
| } |
| |
| public static void validateGatewaySenderQueueHasContent(String senderId, VM... targetVms) { |
| await() |
| .until(() -> Arrays.stream(targetVms).map(vm -> vm.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId); |
| logger.info(displaySerialQueueContent(sender)); |
| return sender.getEventQueueSize(); |
| })).mapToInt(i -> i).sum(), greaterThan(0)); |
| } |
| |
| public static void checkGatewayReceiverStats(int processBatches, int eventsReceived, |
| int creates) { |
| checkGatewayReceiverStats(processBatches, eventsReceived, creates, false); |
| } |
| |
| public static void checkGatewayReceiverStats(int processBatches, int eventsReceived, |
| int creates, boolean isExact) { |
| Set<GatewayReceiver> gatewayReceivers = cache.getGatewayReceivers(); |
| GatewayReceiver receiver = gatewayReceivers.iterator().next(); |
| CacheServerStats stats = ((CacheServerImpl) receiver.getServer()).getAcceptor().getStats(); |
| |
| assertTrue(stats instanceof GatewayReceiverStats); |
| GatewayReceiverStats gatewayReceiverStats = (GatewayReceiverStats) stats; |
| if (isExact) { |
| assertTrue(gatewayReceiverStats.getProcessBatchRequests() == processBatches); |
| } else { |
| assertTrue(gatewayReceiverStats.getProcessBatchRequests() >= processBatches); |
| } |
| assertEquals(eventsReceived, gatewayReceiverStats.getEventsReceived()); |
| assertEquals(creates, gatewayReceiverStats.getCreateRequest()); |
| } |
| |
| public static void checkMinimumGatewayReceiverStats(int processBatches, int eventsReceived) { |
| Set<GatewayReceiver> gatewayReceivers = cache.getGatewayReceivers(); |
| GatewayReceiver receiver = gatewayReceivers.iterator().next(); |
| CacheServerStats stats = ((CacheServerImpl) receiver.getServer()).getAcceptor().getStats(); |
| |
| assertTrue(stats instanceof GatewayReceiverStats); |
| GatewayReceiverStats gatewayReceiverStats = (GatewayReceiverStats) stats; |
| assertTrue(gatewayReceiverStats.getProcessBatchRequests() >= processBatches); |
| assertTrue(gatewayReceiverStats.getEventsReceived() >= eventsReceived); |
| } |
| |
| public static void checkExceptionStats(int exceptionsOccurred) { |
| Set<GatewayReceiver> gatewayReceivers = cache.getGatewayReceivers(); |
| GatewayReceiver receiver = gatewayReceivers.iterator().next(); |
| CacheServerStats stats = ((CacheServerImpl) receiver.getServer()).getAcceptor().getStats(); |
| |
| assertTrue(stats instanceof GatewayReceiverStats); |
| GatewayReceiverStats gatewayReceiverStats = (GatewayReceiverStats) stats; |
| if (exceptionsOccurred == 0) { |
| assertEquals(exceptionsOccurred, gatewayReceiverStats.getExceptionsOccurred()); |
| } else { |
| assertTrue(gatewayReceiverStats.getExceptionsOccurred() >= exceptionsOccurred); |
| } |
| } |
| |
| public static void checkGatewayReceiverStatsHA(int processBatches, int eventsReceived, |
| int creates) { |
| Set<GatewayReceiver> gatewayReceivers = cache.getGatewayReceivers(); |
| GatewayReceiver receiver = gatewayReceivers.iterator().next(); |
| CacheServerStats stats = ((CacheServerImpl) receiver.getServer()).getAcceptor().getStats(); |
| |
| assertTrue(stats instanceof GatewayReceiverStats); |
| GatewayReceiverStats gatewayReceiverStats = (GatewayReceiverStats) stats; |
| assertTrue(gatewayReceiverStats.getProcessBatchRequests() >= processBatches); |
| assertTrue(gatewayReceiverStats.getEventsReceived() >= eventsReceived); |
| assertTrue(gatewayReceiverStats.getCreateRequest() >= creates); |
| } |
| |
| public static void checkEventFilteredStats(String senderId, final int eventsFiltered) { |
| GatewaySenderStats statistics = getGatewaySenderStats(senderId); |
| assertEquals(eventsFiltered, statistics.getEventsFiltered()); |
| } |
| |
| public static void checkConflatedStats(String senderId, final int eventsConflated) { |
| GatewaySenderStats statistics = getGatewaySenderStats(senderId); |
| assertEquals(eventsConflated, statistics.getEventsNotQueuedConflated()); |
| } |
| |
| public static void checkStats_Failover(String senderId, final int eventsReceived) { |
| GatewaySenderStats statistics = getGatewaySenderStats(senderId); |
| assertEquals(eventsReceived, statistics.getEventsReceived()); |
| assertEquals(eventsReceived, |
| (statistics.getEventsQueued() + statistics.getUnprocessedTokensAddedByPrimary() |
| + statistics.getUnprocessedEventsRemovedByPrimary())); |
| } |
| |
| public static void checkBatchStats(String senderId, final int batches) { |
| checkBatchStats(senderId, batches, false); |
| } |
| |
| public static void checkBatchStats(String senderId, final int batches, boolean isExact) { |
| GatewaySenderStats statistics = getGatewaySenderStats(senderId); |
| if (isExact) { |
| assert (statistics.getBatchesDistributed() == batches); |
| } else { |
| assert (statistics.getBatchesDistributed() >= batches); |
| } |
| assertEquals(0, statistics.getBatchesRedistributed()); |
| } |
| |
| public static void checkBatchStats(String senderId, final int batches, |
| boolean isExact, final boolean batchesRedistributed) { |
| GatewaySenderStats statistics = getGatewaySenderStats(senderId); |
| if (isExact) { |
| assert (statistics.getBatchesDistributed() == batches); |
| } else { |
| assert (statistics.getBatchesDistributed() >= batches); |
| } |
| assertEquals(batchesRedistributed, (statistics.getBatchesRedistributed() > 0)); |
| } |
| |
| public static void checkBatchStats(String senderId, final boolean batchesDistributed, |
| final boolean batchesRedistributed) { |
| GatewaySenderStats statistics = getGatewaySenderStats(senderId); |
| assertEquals(batchesDistributed, (statistics.getBatchesDistributed() > 0)); |
| assertEquals(batchesRedistributed, (statistics.getBatchesRedistributed() > 0)); |
| } |
| |
| public static void checkUnProcessedStats(String senderId, int events) { |
| GatewaySenderStats statistics = getGatewaySenderStats(senderId); |
| assertEquals(events, (statistics.getUnprocessedEventsAddedBySecondary() |
| + statistics.getUnprocessedTokensRemovedBySecondary())); |
| assertEquals(events, (statistics.getUnprocessedEventsRemovedByPrimary() |
| + statistics.getUnprocessedTokensAddedByPrimary())); |
| } |
| |
| public static GatewaySenderStats getGatewaySenderStats(String senderId) { |
| GatewaySender sender = cache.getGatewaySender(senderId); |
| return ((AbstractGatewaySender) sender).getStatistics(); |
| } |
| |
| public static void waitForSenderRunningState(String senderId) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| try { |
| Set<GatewaySender> senders = cache.getGatewaySenders(); |
| final GatewaySender sender = getGatewaySenderById(senders, senderId); |
| await() |
| .untilAsserted( |
| () -> assertEquals("Expected sender isRunning state to " + "be true but is false", |
| true, (sender != null && sender.isRunning()))); |
| } finally { |
| exln.remove(); |
| } |
| } |
| |
| public static void waitForSenderToBecomePrimary(String senderId) { |
| Set<GatewaySender> senders = ((GemFireCacheImpl) cache).getAllGatewaySenders(); |
| final GatewaySender sender = getGatewaySenderById(senders, senderId); |
| await() |
| .untilAsserted( |
| () -> assertEquals("Expected sender primary state to " + "be true but is false", |
| true, (sender != null && ((AbstractGatewaySender) sender).isPrimary()))); |
| } |
| |
| private static GatewaySender getGatewaySenderById(Set<GatewaySender> senders, String senderId) { |
| for (GatewaySender s : senders) { |
| if (s.getId().equals(senderId)) { |
| return s; |
| } |
| } |
| // if none of the senders matches with the supplied senderid, return null |
| return null; |
| } |
| |
| public static HashMap checkQueue() { |
| HashMap listenerAttrs = new HashMap(); |
| listenerAttrs.put("Create", listener1.createList); |
| listenerAttrs.put("Update", listener1.updateList); |
| listenerAttrs.put("Destroy", listener1.destroyList); |
| return listenerAttrs; |
| } |
| |
| public static void checkQueueOnSecondary(final Map primaryUpdatesMap) { |
| final HashMap secondaryUpdatesMap = new HashMap(); |
| secondaryUpdatesMap.put("Create", listener1.createList); |
| secondaryUpdatesMap.put("Update", listener1.updateList); |
| secondaryUpdatesMap.put("Destroy", listener1.destroyList); |
| |
| await().untilAsserted(() -> { |
| secondaryUpdatesMap.put("Create", listener1.createList); |
| secondaryUpdatesMap.put("Update", listener1.updateList); |
| secondaryUpdatesMap.put("Destroy", listener1.destroyList); |
| assertEquals( |
| "Expected secondary map to be " + primaryUpdatesMap + " but it is " + secondaryUpdatesMap, |
| true, secondaryUpdatesMap.equals(primaryUpdatesMap)); |
| }); |
| } |
| |
| public static HashMap checkQueue2() { |
| HashMap listenerAttrs = new HashMap(); |
| listenerAttrs.put("Create", listener2.createList); |
| listenerAttrs.put("Update", listener2.updateList); |
| listenerAttrs.put("Destroy", listener2.destroyList); |
| return listenerAttrs; |
| } |
| |
| public static HashMap checkPR(String regionName) { |
| PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName); |
| QueueListener listener = (QueueListener) region.getCacheListener(); |
| |
| HashMap listenerAttrs = new HashMap(); |
| listenerAttrs.put("Create", listener.createList); |
| listenerAttrs.put("Update", listener.updateList); |
| listenerAttrs.put("Destroy", listener.destroyList); |
| return listenerAttrs; |
| } |
| |
| public static HashMap checkBR(String regionName, int numBuckets) { |
| PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName); |
| HashMap listenerAttrs = new HashMap(); |
| for (int i = 0; i < numBuckets; i++) { |
| BucketRegion br = region.getBucketRegion(i); |
| QueueListener listener = (QueueListener) br.getCacheListener(); |
| listenerAttrs.put("Create" + i, listener.createList); |
| listenerAttrs.put("Update" + i, listener.updateList); |
| listenerAttrs.put("Destroy" + i, listener.destroyList); |
| } |
| return listenerAttrs; |
| } |
| |
| public static HashMap checkQueue_BR(String senderId, int numBuckets) { |
| GatewaySender sender = getGatewaySender(senderId); |
| RegionQueue parallelQueue = |
| ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| |
| PartitionedRegion region = (PartitionedRegion) parallelQueue.getRegion(); |
| HashMap listenerAttrs = new HashMap(); |
| for (int i = 0; i < numBuckets; i++) { |
| BucketRegion br = region.getBucketRegion(i); |
| if (br != null) { |
| QueueListener listener = (QueueListener) br.getCacheListener(); |
| if (listener != null) { |
| listenerAttrs.put("Create" + i, listener.createList); |
| listenerAttrs.put("Update" + i, listener.updateList); |
| listenerAttrs.put("Destroy" + i, listener.destroyList); |
| } |
| } |
| } |
| return listenerAttrs; |
| } |
| |
| public static void addListenerOnBucketRegion(String regionName, int numBuckets) { |
| WANTestBase test = new WANTestBase(); |
| test.addCacheListenerOnBucketRegion(regionName, numBuckets); |
| } |
| |
| private void addCacheListenerOnBucketRegion(String regionName, int numBuckets) { |
| PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName); |
| for (int i = 0; i < numBuckets; i++) { |
| BucketRegion br = region.getBucketRegion(i); |
| AttributesMutator mutator = br.getAttributesMutator(); |
| listener1 = new QueueListener(); |
| mutator.addCacheListener(listener1); |
| } |
| } |
| |
| public static void addListenerOnQueueBucketRegion(String senderId, int numBuckets) { |
| WANTestBase test = new WANTestBase(); |
| test.addCacheListenerOnQueueBucketRegion(senderId, numBuckets); |
| } |
| |
| private void addCacheListenerOnQueueBucketRegion(String senderId, int numBuckets) { |
| GatewaySender sender = getGatewaySender(senderId); |
| RegionQueue parallelQueue = |
| ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| |
| PartitionedRegion region = (PartitionedRegion) parallelQueue.getRegion(); |
| for (int i = 0; i < numBuckets; i++) { |
| BucketRegion br = region.getBucketRegion(i); |
| if (br != null) { |
| AttributesMutator mutator = br.getAttributesMutator(); |
| CacheListener listener = new QueueListener(); |
| mutator.addCacheListener(listener); |
| } |
| } |
| |
| } |
| |
| public static void addQueueListener(String senderId, boolean isParallel) { |
| WANTestBase test = new WANTestBase(); |
| test.addCacheQueueListener(senderId, isParallel); |
| } |
| |
| public static void addSecondQueueListener(String senderId, boolean isParallel) { |
| WANTestBase test = new WANTestBase(); |
| test.addSecondCacheQueueListener(senderId, isParallel); |
| } |
| |
| public static void addListenerOnRegion(String regionName) { |
| WANTestBase test = new WANTestBase(); |
| test.addCacheListenerOnRegion(regionName); |
| } |
| |
| private void addCacheListenerOnRegion(String regionName) { |
| Region region = cache.getRegion(regionName); |
| AttributesMutator mutator = region.getAttributesMutator(); |
| listener1 = new QueueListener(); |
| mutator.addCacheListener(listener1); |
| } |
| |
| private void addCacheQueueListener(String senderId, boolean isParallel) { |
| GatewaySender sender = getGatewaySender(senderId); |
| listener1 = new QueueListener(); |
| if (!isParallel) { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| for (RegionQueue q : queues) { |
| q.addCacheListener(listener1); |
| } |
| } else { |
| RegionQueue parallelQueue = |
| ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| parallelQueue.addCacheListener(listener1); |
| } |
| } |
| |
| private void addSecondCacheQueueListener(String senderId, boolean isParallel) { |
| GatewaySender sender = getGatewaySender(senderId); |
| listener2 = new QueueListener(); |
| if (!isParallel) { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| for (RegionQueue q : queues) { |
| q.addCacheListener(listener2); |
| } |
| } else { |
| RegionQueue parallelQueue = |
| ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| parallelQueue.addCacheListener(listener2); |
| } |
| } |
| |
| public static void pauseSender(String senderId) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| GatewaySender sender = getGatewaySender(senderId); |
| sender.pause(); |
| ((AbstractGatewaySender) sender).getEventProcessor().waitForDispatcherToPause(); |
| |
| } finally { |
| exp.remove(); |
| exln.remove(); |
| } |
| } |
| |
| public static void resumeSender(String senderId) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| GatewaySender sender = getGatewaySender(senderId); |
| sender.resume(); |
| } finally { |
| exp.remove(); |
| exln.remove(); |
| } |
| } |
| |
| public static void stopSenderInVMsAsync(String senderId, VM... vms) { |
| List<AsyncInvocation<Void>> tasks = new LinkedList<>(); |
| for (VM vm : vms) { |
| tasks.add(vm.invokeAsync(() -> stopSender(senderId))); |
| } |
| for (AsyncInvocation invocation : tasks) { |
| try { |
| invocation.await(); |
| } catch (InterruptedException e) { |
| fail("Stopping senders was interrupted"); |
| } |
| } |
| } |
| |
| public static void stopSender(String senderId) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| GatewaySender sender = getGatewaySender(senderId); |
| AbstractGatewaySenderEventProcessor eventProcessor = null; |
| if (sender instanceof AbstractGatewaySender) { |
| eventProcessor = ((AbstractGatewaySender) sender).getEventProcessor(); |
| } |
| sender.stop(); |
| |
| Set<RegionQueue> queues = null; |
| if (eventProcessor instanceof ConcurrentSerialGatewaySenderEventProcessor) { |
| queues = ((ConcurrentSerialGatewaySenderEventProcessor) eventProcessor).getQueues(); |
| for (RegionQueue queue : queues) { |
| if (queue instanceof SerialGatewaySenderQueue) { |
| assertFalse(((SerialGatewaySenderQueue) queue).isRemovalThreadAlive()); |
| } |
| } |
| } |
| } finally { |
| exp.remove(); |
| exln.remove(); |
| } |
| } |
| |
| public static void stopReceivers() { |
| Set<GatewayReceiver> receivers = cache.getGatewayReceivers(); |
| for (GatewayReceiver receiver : receivers) { |
| receiver.stop(); |
| } |
| } |
| |
| public static void startReceivers() { |
| Set<GatewayReceiver> receivers = cache.getGatewayReceivers(); |
| for (GatewayReceiver receiver : receivers) { |
| try { |
| receiver.start(); |
| } catch (IOException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| } |
| |
| public static GatewaySenderFactory configureGateway(DiskStoreFactory dsf, File[] dirs1, |
| String dsName, boolean isParallel, Integer maxMemory, Integer batchSize, boolean isConflation, |
| boolean isPersistent, GatewayEventFilter filter, boolean isManualStart, int numDispatchers, |
| OrderPolicy policy, int socketBufferSize) { |
| |
| InternalGatewaySenderFactory gateway = |
| (InternalGatewaySenderFactory) cache.createGatewaySenderFactory(); |
| gateway.setParallel(isParallel); |
| gateway.setMaximumQueueMemory(maxMemory); |
| gateway.setBatchSize(batchSize); |
| gateway.setBatchConflationEnabled(isConflation); |
| gateway.setManualStart(isManualStart); |
| gateway.setDispatcherThreads(numDispatchers); |
| gateway.setOrderPolicy(policy); |
| gateway.setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| gateway.setSocketBufferSize(socketBufferSize); |
| if (filter != null) { |
| eventFilter = filter; |
| gateway.addGatewayEventFilter(filter); |
| } |
| if (isPersistent) { |
| gateway.setPersistenceEnabled(true); |
| gateway.setDiskStoreName(dsf.setDiskDirs(dirs1).create(dsName).getName()); |
| } else { |
| DiskStore store = dsf.setDiskDirs(dirs1).create(dsName); |
| gateway.setDiskStoreName(store.getName()); |
| } |
| return gateway; |
| } |
| |
| public static void createSender(String dsName, int remoteDsId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| GatewayEventFilter filter, boolean isManualStart) { |
| createSender(dsName, remoteDsId, isParallel, maxMemory, batchSize, isConflation, isPersistent, |
| filter, isManualStart, false); |
| } |
| |
| public static void createSender(String dsName, int remoteDsId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| GatewayEventFilter filter, boolean isManualStart, boolean groupTransactionEvents) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| try { |
| File persistentDirectory = |
| new File(dsName + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| persistentDirectory.mkdir(); |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| File[] dirs1 = new File[] {persistentDirectory}; |
| GatewaySenderFactory gateway = configureGateway(dsf, dirs1, dsName, isParallel, maxMemory, |
| batchSize, isConflation, isPersistent, filter, isManualStart, |
| numDispatcherThreadsForTheRun, GatewaySender.DEFAULT_ORDER_POLICY, |
| GatewaySender.DEFAULT_SOCKET_BUFFER_SIZE); |
| gateway.setGroupTransactionEvents(groupTransactionEvents); |
| gateway.create(dsName, remoteDsId); |
| } finally { |
| exln.remove(); |
| } |
| } |
| |
| public static void createSenderWithMultipleDispatchers(String dsName, int remoteDsId, |
| boolean isParallel, Integer maxMemory, Integer batchSize, boolean isConflation, |
| boolean isPersistent, GatewayEventFilter filter, boolean isManualStart, int numDispatchers, |
| OrderPolicy orderPolicy) { |
| createSenderWithMultipleDispatchers(dsName, remoteDsId, |
| isParallel, maxMemory, batchSize, isConflation, |
| isPersistent, filter, isManualStart, numDispatchers, |
| orderPolicy, GatewaySender.DEFAULT_SOCKET_BUFFER_SIZE); |
| } |
| |
| public static void createSenderWithMultipleDispatchers(String dsName, int remoteDsId, |
| boolean isParallel, Integer maxMemory, Integer batchSize, boolean isConflation, |
| boolean isPersistent, GatewayEventFilter filter, boolean isManualStart, int numDispatchers, |
| OrderPolicy orderPolicy, int socketBufferSize) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| try { |
| File persistentDirectory = |
| new File(dsName + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| persistentDirectory.mkdir(); |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| File[] dirs1 = new File[] {persistentDirectory}; |
| GatewaySenderFactory gateway = |
| configureGateway(dsf, dirs1, dsName, isParallel, maxMemory, batchSize, isConflation, |
| isPersistent, filter, isManualStart, numDispatchers, orderPolicy, socketBufferSize); |
| gateway.create(dsName, remoteDsId); |
| |
| } finally { |
| exln.remove(); |
| } |
| } |
| |
| public static void createSenderWithoutDiskStore(String dsName, int remoteDsId, Integer maxMemory, |
| Integer batchSize, boolean isConflation, boolean isManualStart) { |
| |
| GatewaySenderFactory gateway = cache.createGatewaySenderFactory(); |
| gateway.setParallel(true); |
| gateway.setMaximumQueueMemory(maxMemory); |
| gateway.setBatchSize(batchSize); |
| gateway.setManualStart(isManualStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| gateway.setBatchConflationEnabled(isConflation); |
| gateway.create(dsName, remoteDsId); |
| } |
| |
| public static void createSenderAlertThresholdWithoutDiskStore(String dsName, int remoteDsId, |
| Integer maxMemory, |
| Integer batchSize, boolean isConflation, boolean isManualStart, int alertthreshold) { |
| |
| GatewaySenderFactory gateway = cache.createGatewaySenderFactory(); |
| gateway.setAlertThreshold(alertthreshold); |
| gateway.setParallel(true); |
| gateway.setMaximumQueueMemory(maxMemory); |
| gateway.setBatchSize(batchSize); |
| gateway.setManualStart(isManualStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| gateway.setBatchConflationEnabled(isConflation); |
| gateway.create(dsName, remoteDsId); |
| } |
| |
| public static void createConcurrentSender(String dsName, int remoteDsId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| GatewayEventFilter filter, boolean isManualStart, int concurrencyLevel, OrderPolicy policy) { |
| |
| File persistentDirectory = |
| new File(dsName + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| persistentDirectory.mkdir(); |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| File[] dirs1 = new File[] {persistentDirectory}; |
| GatewaySenderFactory gateway = configureGateway(dsf, dirs1, dsName, isParallel, maxMemory, |
| batchSize, isConflation, isPersistent, filter, isManualStart, concurrencyLevel, policy, |
| GatewaySender.DEFAULT_SOCKET_BUFFER_SIZE); |
| gateway.create(dsName, remoteDsId); |
| } |
| |
| public static void createSenderForValidations(String dsName, int remoteDsId, boolean isParallel, |
| Integer alertThreshold, boolean isConflation, boolean isPersistent, |
| List<GatewayEventFilter> eventFilters, List<GatewayTransportFilter> transportFilters, |
| boolean isManualStart, boolean isDiskSync) { |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(RegionDestroyedException.class.getName()); |
| try { |
| File persistentDirectory = |
| new File(dsName + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| persistentDirectory.mkdir(); |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| File[] dirs1 = new File[] {persistentDirectory}; |
| |
| if (isParallel) { |
| GatewaySenderFactory gateway = cache.createGatewaySenderFactory(); |
| gateway.setParallel(true); |
| gateway.setAlertThreshold(alertThreshold); |
| ((InternalGatewaySenderFactory) gateway) |
| .setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| if (eventFilters != null) { |
| for (GatewayEventFilter filter : eventFilters) { |
| gateway.addGatewayEventFilter(filter); |
| } |
| } |
| if (transportFilters != null) { |
| for (GatewayTransportFilter filter : transportFilters) { |
| gateway.addGatewayTransportFilter(filter); |
| } |
| } |
| if (isPersistent) { |
| gateway.setPersistenceEnabled(true); |
| gateway.setDiskStoreName(dsf.setDiskDirs(dirs1).create(dsName + "_Parallel").getName()); |
| } else { |
| DiskStore store = dsf.setDiskDirs(dirs1).create(dsName + "_Parallel"); |
| gateway.setDiskStoreName(store.getName()); |
| } |
| gateway.setDiskSynchronous(isDiskSync); |
| gateway.setBatchConflationEnabled(isConflation); |
| gateway.setManualStart(isManualStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| gateway.create(dsName, remoteDsId); |
| |
| } else { |
| GatewaySenderFactory gateway = cache.createGatewaySenderFactory(); |
| gateway.setAlertThreshold(alertThreshold); |
| gateway.setManualStart(isManualStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| ((InternalGatewaySenderFactory) gateway) |
| .setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| if (eventFilters != null) { |
| for (GatewayEventFilter filter : eventFilters) { |
| gateway.addGatewayEventFilter(filter); |
| } |
| } |
| if (transportFilters != null) { |
| for (GatewayTransportFilter filter : transportFilters) { |
| gateway.addGatewayTransportFilter(filter); |
| } |
| } |
| gateway.setBatchConflationEnabled(isConflation); |
| if (isPersistent) { |
| gateway.setPersistenceEnabled(true); |
| gateway.setDiskStoreName(dsf.setDiskDirs(dirs1).create(dsName + "_Serial").getName()); |
| } else { |
| DiskStore store = dsf.setDiskDirs(dirs1).create(dsName + "_Serial"); |
| gateway.setDiskStoreName(store.getName()); |
| } |
| gateway.setDiskSynchronous(isDiskSync); |
| gateway.create(dsName, remoteDsId); |
| } |
| } finally { |
| exp1.remove(); |
| } |
| } |
| |
| public static String createSenderWithDiskStore(String dsName, int remoteDsId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| GatewayEventFilter filter, String dsStore, boolean isManualStart) { |
| File persistentDirectory = null; |
| if (dsStore == null) { |
| persistentDirectory = |
| new File(dsName + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| } else { |
| persistentDirectory = new File(dsStore); |
| } |
| logger.info("The ds is : " + persistentDirectory.getName()); |
| |
| persistentDirectory.mkdir(); |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| File[] dirs1 = new File[] {persistentDirectory}; |
| |
| if (isParallel) { |
| GatewaySenderFactory gateway = cache.createGatewaySenderFactory(); |
| gateway.setParallel(true); |
| gateway.setMaximumQueueMemory(maxMemory); |
| gateway.setBatchSize(batchSize); |
| gateway.setManualStart(isManualStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| ((InternalGatewaySenderFactory) gateway).setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| if (filter != null) { |
| gateway.addGatewayEventFilter(filter); |
| } |
| if (isPersistent) { |
| gateway.setPersistenceEnabled(true); |
| String dsname = dsf.setDiskDirs(dirs1).create(dsName).getName(); |
| gateway.setDiskStoreName(dsname); |
| logger.info("The DiskStoreName is : " + dsname); |
| } else { |
| DiskStore store = dsf.setDiskDirs(dirs1).create(dsName); |
| gateway.setDiskStoreName(store.getName()); |
| logger.info("The ds is : " + store.getName()); |
| } |
| gateway.setBatchConflationEnabled(isConflation); |
| gateway.create(dsName, remoteDsId); |
| |
| } else { |
| GatewaySenderFactory gateway = cache.createGatewaySenderFactory(); |
| gateway.setMaximumQueueMemory(maxMemory); |
| gateway.setBatchSize(batchSize); |
| gateway.setManualStart(isManualStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| ((InternalGatewaySenderFactory) gateway).setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| if (filter != null) { |
| gateway.addGatewayEventFilter(filter); |
| } |
| gateway.setBatchConflationEnabled(isConflation); |
| if (isPersistent) { |
| gateway.setPersistenceEnabled(true); |
| gateway.setDiskStoreName(dsf.setDiskDirs(dirs1).create(dsName).getName()); |
| } else { |
| DiskStore store = dsf.setDiskDirs(dirs1).create(dsName); |
| |
| gateway.setDiskStoreName(store.getName()); |
| } |
| gateway.create(dsName, remoteDsId); |
| } |
| return persistentDirectory.getName(); |
| } |
| |
| |
| public static void createSenderWithListener(String dsName, int remoteDsName, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| GatewayEventFilter filter, boolean attachTwoListeners, boolean isManualStart) { |
| File persistentDirectory = |
| new File(dsName + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| persistentDirectory.mkdir(); |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| File[] dirs1 = new File[] {persistentDirectory}; |
| |
| if (isParallel) { |
| GatewaySenderFactory gateway = cache.createGatewaySenderFactory(); |
| gateway.setParallel(true); |
| gateway.setMaximumQueueMemory(maxMemory); |
| gateway.setBatchSize(batchSize); |
| gateway.setManualStart(isManualStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| ((InternalGatewaySenderFactory) gateway).setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| if (filter != null) { |
| gateway.addGatewayEventFilter(filter); |
| } |
| if (isPersistent) { |
| gateway.setPersistenceEnabled(true); |
| gateway.setDiskStoreName(dsf.setDiskDirs(dirs1).create(dsName).getName()); |
| } else { |
| DiskStore store = dsf.setDiskDirs(dirs1).create(dsName); |
| gateway.setDiskStoreName(store.getName()); |
| } |
| gateway.setBatchConflationEnabled(isConflation); |
| gateway.create(dsName, remoteDsName); |
| |
| } else { |
| GatewaySenderFactory gateway = cache.createGatewaySenderFactory(); |
| gateway.setMaximumQueueMemory(maxMemory); |
| gateway.setBatchSize(batchSize); |
| gateway.setManualStart(isManualStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| ((InternalGatewaySenderFactory) gateway).setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| if (filter != null) { |
| gateway.addGatewayEventFilter(filter); |
| } |
| gateway.setBatchConflationEnabled(isConflation); |
| if (isPersistent) { |
| gateway.setPersistenceEnabled(true); |
| gateway.setDiskStoreName(dsf.setDiskDirs(dirs1).create(dsName).getName()); |
| } else { |
| DiskStore store = dsf.setDiskDirs(dirs1).create(dsName); |
| gateway.setDiskStoreName(store.getName()); |
| } |
| |
| eventListener1 = new MyGatewaySenderEventListener(); |
| ((InternalGatewaySenderFactory) gateway).addAsyncEventListener(eventListener1); |
| if (attachTwoListeners) { |
| eventListener2 = new MyGatewaySenderEventListener2(); |
| ((InternalGatewaySenderFactory) gateway).addAsyncEventListener(eventListener2); |
| } |
| ((InternalGatewaySenderFactory) gateway).create(dsName); |
| } |
| } |
| |
| public static void createReceiverInVMs(int maximumTimeBetweenPings, VM... vms) { |
| for (VM vm : vms) { |
| vm.invoke(() -> createReceiverWithMaximumTimeBetweenPings(maximumTimeBetweenPings)); |
| } |
| } |
| |
| |
| public static void createReceiverInVMs(VM... vms) { |
| createReceiverInVMs(-1, vms); |
| } |
| |
| public static int createReceiver() { |
| return createReceiverWithMaximumTimeBetweenPings(-1); |
| } |
| |
| public static int createReceiverWithMaximumTimeBetweenPings(int maximumTimeBetweenPings) { |
| GatewayReceiverFactory fact = cache.createGatewayReceiverFactory(); |
| int port = getRandomAvailableTCPPort(); |
| fact.setStartPort(port); |
| fact.setEndPort(port); |
| fact.setManualStart(true); |
| if (maximumTimeBetweenPings > 0) { |
| fact.setMaximumTimeBetweenPings(maximumTimeBetweenPings); |
| } |
| GatewayReceiver receiver = fact.create(); |
| try { |
| receiver.start(); |
| } catch (IOException e) { |
| e.printStackTrace(); |
| Assert.fail( |
| "Test " + getTestMethodName() + " failed to start GatewayReceiver on port " + port, e); |
| } |
| return port; |
| } |
| |
| public static int createReceiverWithSSL(int locPort) { |
| WANTestBase test = new WANTestBase(); |
| |
| Properties gemFireProps = test.getDistributedSystemProperties(); |
| |
| gemFireProps.put(LOG_LEVEL, LogWriterUtils.getDUnitLogLevel()); |
| gemFireProps.put(GATEWAY_SSL_ENABLED, "true"); |
| gemFireProps.put(GATEWAY_SSL_PROTOCOLS, "any"); |
| gemFireProps.put(GATEWAY_SSL_CIPHERS, "any"); |
| gemFireProps.put(GATEWAY_SSL_REQUIRE_AUTHENTICATION, "true"); |
| |
| gemFireProps.put(GATEWAY_SSL_KEYSTORE_TYPE, "jks"); |
| gemFireProps.put(GATEWAY_SSL_KEYSTORE, createTempFileFromResource(WANTestBase.class, |
| "/org/apache/geode/cache/client/internal/default.keystore").getAbsolutePath()); |
| gemFireProps.put(GATEWAY_SSL_KEYSTORE_PASSWORD, "password"); |
| gemFireProps.put(GATEWAY_SSL_TRUSTSTORE, createTempFileFromResource(WANTestBase.class, |
| "/org/apache/geode/cache/client/internal/default.keystore").getAbsolutePath()); |
| gemFireProps.put(GATEWAY_SSL_TRUSTSTORE_PASSWORD, "password"); |
| |
| gemFireProps.setProperty(MCAST_PORT, "0"); |
| gemFireProps.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| |
| logger.info("Starting cache ds with following properties \n" + gemFireProps); |
| |
| InternalDistributedSystem ds = test.getSystem(gemFireProps); |
| cache = CacheFactory.create(ds); |
| GatewayReceiverFactory fact = cache.createGatewayReceiverFactory(); |
| int port = getRandomAvailableTCPPort(); |
| fact.setStartPort(port); |
| fact.setEndPort(port); |
| fact.setManualStart(true); |
| GatewayReceiver receiver = fact.create(); |
| try { |
| receiver.start(); |
| } catch (IOException e) { |
| e.printStackTrace(); |
| fail("Test " + test.getName() + " failed to start GatewayReceiver on port " + port); |
| } |
| return port; |
| } |
| |
| public static void createReceiverAndServer(int locPort) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| GatewayReceiverFactory fact = cache.createGatewayReceiverFactory(); |
| int receiverPort = getRandomAvailableTCPPort(); |
| fact.setStartPort(receiverPort); |
| fact.setEndPort(receiverPort); |
| fact.setManualStart(true); |
| GatewayReceiver receiver = fact.create(); |
| try { |
| receiver.start(); |
| } catch (IOException e) { |
| e.printStackTrace(); |
| fail("Test " + test.getName() + " failed to start GatewayReceiver on port " + receiverPort); |
| } |
| CacheServer server = cache.addCacheServer(); |
| int serverPort = getRandomAvailableTCPPort(); |
| server.setPort(serverPort); |
| server.setHostnameForClients("localhost"); |
| try { |
| server.start(); |
| } catch (IOException e) { |
| org.apache.geode.test.dunit.Assert.fail("Failed to start server ", e); |
| } |
| } |
| |
| public static int createServer(int locPort) { |
| return createServer(locPort, -1); |
| } |
| |
| public static int createServer(int locPort, int maximumTimeBetweenPings) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| |
| CacheServer server = cache.addCacheServer(); |
| int port = getRandomAvailableTCPPort(); |
| server.setPort(port); |
| server.setHostnameForClients("localhost"); |
| if (maximumTimeBetweenPings > 0) { |
| server.setMaximumTimeBetweenPings(maximumTimeBetweenPings); |
| } |
| try { |
| server.start(); |
| } catch (IOException e) { |
| org.apache.geode.test.dunit.Assert.fail("Failed to start server ", e); |
| } |
| return port; |
| } |
| |
| public static void createClientWithLocator(int port0, String host, String regionName) { |
| createClientWithLocator(port0, host); |
| |
| RegionFactory factory = cache.createRegionFactory(RegionShortcut.LOCAL); |
| factory.setPoolName("pool"); |
| |
| region = factory.create(regionName); |
| region.registerInterest("ALL_KEYS"); |
| assertNotNull(region); |
| logger.info("Distributed Region " + regionName + " created Successfully :" + region.toString()); |
| } |
| |
| public static void createClientWithLocator(final int port0, final String host) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, ""); |
| |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| |
| assertNotNull(cache); |
| CacheServerTestUtil.disableShufflingOfEndpoints(); |
| Pool p; |
| try { |
| p = PoolManager.createFactory().addLocator(host, port0).setPingInterval(250) |
| .setSubscriptionEnabled(true).setSubscriptionRedundancy(-1).setReadTimeout(20000) |
| .setSocketBufferSize(1000).setMinConnections(6).setMaxConnections(10).setRetryAttempts(3) |
| .create("pool"); |
| } finally { |
| CacheServerTestUtil.enableShufflingOfEndpoints(); |
| } |
| } |
| |
| public static int createReceiver_PDX(int locPort) { |
| WANTestBase test = new WANTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| InternalDistributedSystem ds = test.getSystem(props); |
| File pdxDir = new File(CacheTestCase.getDiskDir(), "pdx"); |
| |
| cache = new InternalCacheBuilder(props) |
| .setPdxPersistent(true) |
| .setPdxDiskStore("pdxStore") |
| .setIsExistingOk(false) |
| .create(ds); |
| |
| cache.createDiskStoreFactory().setDiskDirs(new File[] {pdxDir}).setMaxOplogSize(1) |
| .create("pdxStore"); |
| GatewayReceiverFactory fact = cache.createGatewayReceiverFactory(); |
| int port = getRandomAvailableTCPPort(); |
| fact.setStartPort(port); |
| fact.setEndPort(port); |
| fact.setManualStart(true); |
| GatewayReceiver receiver = fact.create(); |
| try { |
| receiver.start(); |
| } catch (IOException e) { |
| e.printStackTrace(); |
| fail("Test " + test.getName() + " failed to start GatewayReceiver on port " + port); |
| } |
| return port; |
| } |
| |
| public static void doDistTXPuts(String regionName, int numPuts) { |
| CacheTransactionManager txMgr = cache.getCacheTransactionManager(); |
| txMgr.setDistributed(true); |
| |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| IgnoredException exp2 = |
| IgnoredException.addIgnoredException(GatewaySenderException.class.getName()); |
| try { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 1; i <= numPuts; i++) { |
| txMgr.begin(); |
| r.put(i, i); |
| txMgr.commit(); |
| } |
| } finally { |
| exp1.remove(); |
| exp2.remove(); |
| } |
| } |
| |
| |
| public static void doPuts(String regionName, int numPuts, Object value) { |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| IgnoredException exp2 = |
| IgnoredException.addIgnoredException(GatewaySenderException.class.getName()); |
| try { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numPuts; i++) { |
| r.put(i, value); |
| } |
| } finally { |
| exp1.remove(); |
| exp2.remove(); |
| } |
| } |
| |
| public static void doPuts(String regionName, int numPuts) { |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| IgnoredException exp2 = |
| IgnoredException.addIgnoredException(GatewaySenderException.class.getName()); |
| try { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numPuts; i++) { |
| r.put(i, "Value_" + i); |
| } |
| } finally { |
| exp1.remove(); |
| exp2.remove(); |
| } |
| } |
| |
| public static void doPutsSameKey(String regionName, int numPuts, String key) { |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| IgnoredException exp2 = |
| IgnoredException.addIgnoredException(GatewaySenderException.class.getName()); |
| try { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numPuts; i++) { |
| r.put(key, "Value_" + i); |
| } |
| } finally { |
| exp1.remove(); |
| exp2.remove(); |
| } |
| } |
| |
| |
| public static void doPutsAfter300(String regionName, int numPuts) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 300; i < numPuts; i++) { |
| r.put(i, "Value_" + i); |
| } |
| } |
| |
| public static void doPutsFrom(String regionName, int from, int numPuts) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = from; i < numPuts; i++) { |
| r.put(i, "Value_" + i); |
| } |
| } |
| |
| public static void doDestroys(String regionName, int keyNum) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < keyNum; i++) { |
| r.destroy(i); |
| } |
| } |
| |
| public static void doPutAll(String regionName, int numPuts, int size) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numPuts; i++) { |
| Map putAllMap = new HashMap(); |
| for (long j = 0; j < size; j++) { |
| putAllMap.put((size * i) + j, "Value_" + i); |
| } |
| r.putAll(putAllMap, "putAllCallback"); |
| putAllMap.clear(); |
| } |
| } |
| |
| |
| public static void doPutsWithKeyAsString(String regionName, int numPuts) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numPuts; i++) { |
| r.put("Object_" + i, "Value_" + i); |
| } |
| } |
| |
| public static void putGivenKeyValue(String regionName, Map keyValues) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (Object key : keyValues.keySet()) { |
| r.put(key, keyValues.get(key)); |
| } |
| } |
| |
| public static void doOrderAndShipmentPutsInsideTransactions(Map keyValues, |
| int eventsPerTransaction) { |
| Region orderRegion = cache.getRegion(orderRegionName); |
| Region shipmentRegion = cache.getRegion(shipmentRegionName); |
| assertNotNull(orderRegion); |
| assertNotNull(shipmentRegion); |
| int eventInTransaction = 0; |
| CacheTransactionManager cacheTransactionManager = cache.getCacheTransactionManager(); |
| for (Object key : keyValues.keySet()) { |
| if (eventInTransaction == 0) { |
| cacheTransactionManager.begin(); |
| } |
| Region r; |
| if (key instanceof OrderId) { |
| r = orderRegion; |
| } else { |
| r = shipmentRegion; |
| } |
| r.put(key, keyValues.get(key)); |
| if (++eventInTransaction == eventsPerTransaction) { |
| cacheTransactionManager.commit(); |
| eventInTransaction = 0; |
| } |
| } |
| if (eventInTransaction != 0) { |
| cacheTransactionManager.commit(); |
| } |
| } |
| |
| public static void doPutsInsideTransactions(String regionName, Map keyValues, |
| int eventsPerTransaction) { |
| Region r = cache.getRegion(Region.SEPARATOR + regionName); |
| assertNotNull(r); |
| int eventInTransaction = 0; |
| CacheTransactionManager cacheTransactionManager = cache.getCacheTransactionManager(); |
| for (Object key : keyValues.keySet()) { |
| if (eventInTransaction == 0) { |
| cacheTransactionManager.begin(); |
| } |
| r.put(key, keyValues.get(key)); |
| if (++eventInTransaction == eventsPerTransaction) { |
| cacheTransactionManager.commit(); |
| eventInTransaction = 0; |
| } |
| } |
| if (eventInTransaction != 0) { |
| cacheTransactionManager.commit(); |
| } |
| } |
| |
| public static void destroyRegion(String regionName) { |
| destroyRegion(regionName, -1); |
| } |
| |
| public static void destroyRegion(String regionName, final int min) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| await().until(() -> r.size() > min); |
| r.destroyRegion(); |
| } |
| |
| public static void localDestroyRegion(String regionName) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(PRLocallyDestroyedException.class.getName()); |
| try { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| r.localDestroyRegion(); |
| } finally { |
| exp.remove(); |
| } |
| } |
| |
| |
| public static Map putCustomerPartitionedRegion(int numPuts) { |
| String valueSuffix = ""; |
| return putCustomerPartitionedRegion(numPuts, valueSuffix); |
| } |
| |
| public static Map updateCustomerPartitionedRegion(int numPuts) { |
| String valueSuffix = "_update"; |
| return putCustomerPartitionedRegion(numPuts, valueSuffix); |
| } |
| |
| protected static Map putCustomerPartitionedRegion(int numPuts, String valueSuffix) { |
| assertNotNull(cache); |
| assertNotNull(customerRegion); |
| Map custKeyValues = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| Customer customer = new Customer("name" + i, "Address" + i + valueSuffix); |
| try { |
| customerRegion.put(custid, customer); |
| assertTrue(customerRegion.containsKey(custid)); |
| assertEquals(customer, customerRegion.get(custid)); |
| custKeyValues.put(custid, customer); |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "putCustomerPartitionedRegion : failed while doing put operation in CustomerPartitionedRegion ", |
| e); |
| } |
| logger.info("Customer :- { " + custid + " : " + customer + " }"); |
| } |
| return custKeyValues; |
| } |
| |
| public static Map putOrderPartitionedRegion(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(orderRegion); |
| Map orderKeyValues = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| int oid = i + 1; |
| OrderId orderId = new OrderId(oid, custid); |
| Order order = new Order("ORDER" + oid); |
| try { |
| orderRegion.put(orderId, order); |
| orderKeyValues.put(orderId, order); |
| assertTrue(orderRegion.containsKey(orderId)); |
| assertEquals(order, orderRegion.get(orderId)); |
| |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "putOrderPartitionedRegion : failed while doing put operation in OrderPartitionedRegion ", |
| e); |
| } |
| logger.info("Order :- { " + orderId + " : " + order + " }"); |
| } |
| return orderKeyValues; |
| } |
| |
| public static Map putOrderPartitionedRegionUsingCustId(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(orderRegion); |
| Map orderKeyValues = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| Order order = new Order("ORDER" + i); |
| try { |
| orderRegion.put(custid, order); |
| orderKeyValues.put(custid, order); |
| assertTrue(orderRegion.containsKey(custid)); |
| assertEquals(order, orderRegion.get(custid)); |
| |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "putOrderPartitionedRegionUsingCustId : failed while doing put operation in OrderPartitionedRegion ", |
| e); |
| } |
| logger.info("Order :- { " + custid + " : " + order + " }"); |
| } |
| return orderKeyValues; |
| } |
| |
| public static Map updateOrderPartitionedRegion(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(orderRegion); |
| Map orderKeyValues = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| int oid = i + 1; |
| OrderId orderId = new OrderId(oid, custid); |
| Order order = new Order("ORDER" + oid + "_update"); |
| try { |
| orderRegion.put(orderId, order); |
| orderKeyValues.put(orderId, order); |
| assertTrue(orderRegion.containsKey(orderId)); |
| assertEquals(order, orderRegion.get(orderId)); |
| |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "updateOrderPartitionedRegion : failed while doing put operation in OrderPartitionedRegion ", |
| e); |
| } |
| logger.info("Order :- { " + orderId + " : " + order + " }"); |
| } |
| return orderKeyValues; |
| } |
| |
| public static Map updateOrderPartitionedRegionUsingCustId(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(orderRegion); |
| Map orderKeyValues = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| Order order = new Order("ORDER" + i + "_update"); |
| try { |
| orderRegion.put(custid, order); |
| assertTrue(orderRegion.containsKey(custid)); |
| assertEquals(order, orderRegion.get(custid)); |
| orderKeyValues.put(custid, order); |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "updateOrderPartitionedRegionUsingCustId : failed while doing put operation in OrderPartitionedRegion ", |
| e); |
| } |
| logger.info("Order :- { " + custid + " : " + order + " }"); |
| } |
| return orderKeyValues; |
| } |
| |
| public static Map putShipmentPartitionedRegion(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(shipmentRegion); |
| Map shipmentKeyValue = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| int oid = i + 1; |
| OrderId orderId = new OrderId(oid, custid); |
| int sid = oid + 1; |
| ShipmentId shipmentId = new ShipmentId(sid, orderId); |
| Shipment shipment = new Shipment("Shipment" + sid); |
| try { |
| shipmentRegion.put(shipmentId, shipment); |
| assertTrue(shipmentRegion.containsKey(shipmentId)); |
| assertEquals(shipment, shipmentRegion.get(shipmentId)); |
| shipmentKeyValue.put(shipmentId, shipment); |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "putShipmentPartitionedRegion : failed while doing put operation in ShipmentPartitionedRegion ", |
| e); |
| } |
| logger.info("Shipment :- { " + shipmentId + " : " + shipment + " }"); |
| } |
| return shipmentKeyValue; |
| } |
| |
| public static void putcolocatedPartitionedRegion(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(customerRegion); |
| assertNotNull(orderRegion); |
| assertNotNull(shipmentRegion); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| Customer customer = new Customer("Customer" + custid, "Address" + custid); |
| customerRegion.put(custid, customer); |
| int oid = i + 1; |
| OrderId orderId = new OrderId(oid, custid); |
| Order order = new Order("Order" + orderId); |
| orderRegion.put(orderId, order); |
| int sid = oid + 1; |
| ShipmentId shipmentId = new ShipmentId(sid, orderId); |
| Shipment shipment = new Shipment("Shipment" + sid); |
| shipmentRegion.put(shipmentId, shipment); |
| } |
| } |
| |
| public static Map putShipmentPartitionedRegionUsingCustId(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(shipmentRegion); |
| Map shipmentKeyValue = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| Shipment shipment = new Shipment("Shipment" + i); |
| try { |
| shipmentRegion.put(custid, shipment); |
| assertTrue(shipmentRegion.containsKey(custid)); |
| assertEquals(shipment, shipmentRegion.get(custid)); |
| shipmentKeyValue.put(custid, shipment); |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "putShipmentPartitionedRegionUsingCustId : failed while doing put operation in ShipmentPartitionedRegion ", |
| e); |
| } |
| logger.info("Shipment :- { " + custid + " : " + shipment + " }"); |
| } |
| return shipmentKeyValue; |
| } |
| |
| public static Map updateShipmentPartitionedRegion(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(shipmentRegion); |
| Map shipmentKeyValue = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| int oid = i + 1; |
| OrderId orderId = new OrderId(oid, custid); |
| int sid = oid + 1; |
| ShipmentId shipmentId = new ShipmentId(sid, orderId); |
| Shipment shipment = new Shipment("Shipment" + sid + "_update"); |
| try { |
| shipmentRegion.put(shipmentId, shipment); |
| assertTrue(shipmentRegion.containsKey(shipmentId)); |
| assertEquals(shipment, shipmentRegion.get(shipmentId)); |
| shipmentKeyValue.put(shipmentId, shipment); |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "updateShipmentPartitionedRegion : failed while doing put operation in ShipmentPartitionedRegion ", |
| e); |
| } |
| logger.info("Shipment :- { " + shipmentId + " : " + shipment + " }"); |
| } |
| return shipmentKeyValue; |
| } |
| |
| public static Map updateShipmentPartitionedRegionUsingCustId(int numPuts) { |
| assertNotNull(cache); |
| assertNotNull(shipmentRegion); |
| Map shipmentKeyValue = new HashMap(); |
| for (int i = 1; i <= numPuts; i++) { |
| CustId custid = new CustId(i); |
| Shipment shipment = new Shipment("Shipment" + i + "_update"); |
| try { |
| shipmentRegion.put(custid, shipment); |
| assertTrue(shipmentRegion.containsKey(custid)); |
| assertEquals(shipment, shipmentRegion.get(custid)); |
| shipmentKeyValue.put(custid, shipment); |
| } catch (Exception e) { |
| org.apache.geode.test.dunit.Assert.fail( |
| "updateShipmentPartitionedRegionUsingCustId : failed while doing put operation in ShipmentPartitionedRegion ", |
| e); |
| } |
| logger.info("Shipment :- { " + custid + " : " + shipment + " }"); |
| } |
| return shipmentKeyValue; |
| } |
| |
| public static void doPutsPDXSerializable(String regionName, int numPuts) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (int i = 0; i < numPuts; i++) { |
| r.put("Key_" + i, new SimpleClass(i, (byte) i)); |
| } |
| } |
| |
| public static void doPutsPDXSerializable2(String regionName, int numPuts) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (int i = 0; i < numPuts; i++) { |
| r.put("Key_" + i, new SimpleClass1(false, (short) i, "" + i, i, "" + i, "" + i, i, i)); |
| } |
| } |
| |
| |
| public static void doTxPuts(String regionName) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| CacheTransactionManager mgr = cache.getCacheTransactionManager(); |
| |
| mgr.begin(); |
| r.put(0, 0); |
| r.put(100, 100); |
| r.put(200, 200); |
| mgr.commit(); |
| } |
| |
| public static void doTxPutsWithRetryIfError(String regionName, final long putsPerTransaction, |
| final long transactions, long offset) { |
| Region r = cache.getRegion(Region.SEPARATOR + regionName); |
| assertNotNull(r); |
| |
| long keyOffset = offset * ((putsPerTransaction + (10 * transactions)) * 100); |
| long i = 0; |
| long j = 0; |
| CacheTransactionManager mgr = cache.getCacheTransactionManager();; |
| for (i = 0; i < transactions; i++) { |
| boolean done = false; |
| do { |
| try { |
| mgr.begin(); |
| for (j = 0; j < putsPerTransaction; j++) { |
| long key = keyOffset + ((j + (10 * i)) * 100); |
| String value = "Value_" + key; |
| r.put(key, value); |
| } |
| mgr.commit(); |
| done = true; |
| } catch (TransactionException e) { |
| logger.info("Something went wrong with transaction [{},{}]. Retrying. Error: {}", i, j, |
| e.getMessage()); |
| e.printStackTrace(); |
| } catch (IllegalStateException e1) { |
| logger.info("Something went wrong with transaction [{},{}]. Retrying. Error: {}", i, j, |
| e1.getMessage()); |
| e1.printStackTrace(); |
| try { |
| mgr.rollback(); |
| logger.info("Rolled back transaction [{},{}]. Retrying. Error: {}", i, j, |
| e1.getMessage()); |
| } catch (Exception e2) { |
| logger.info( |
| "Something went wrong when rolling back transaction [{},{}]. Retrying transaction. Error: {}", |
| i, j, e2.getMessage()); |
| e2.printStackTrace(); |
| } |
| } |
| } while (!done); |
| } |
| } |
| |
| public static void doNextPuts(String regionName, int start, int numPuts) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(CacheClosedException.class.getName()); |
| try { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = start; i < numPuts; i++) { |
| r.put(i, i); |
| } |
| } finally { |
| exp.remove(); |
| } |
| } |
| |
| public static void checkQueueSize(String senderId, int numQueueEntries) { |
| await() |
| .untilAsserted(() -> testQueueSize(senderId, numQueueEntries)); |
| } |
| |
| public static void testQueueSize(String senderId, int numQueueEntries) { |
| GatewaySender sender = cache.getGatewaySender(senderId); |
| if (sender.isParallel()) { |
| int totalSize = 0; |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| for (RegionQueue q : queues) { |
| ConcurrentParallelGatewaySenderQueue prQ = (ConcurrentParallelGatewaySenderQueue) q; |
| totalSize += prQ.size(); |
| } |
| assertEquals(numQueueEntries, totalSize); |
| } else { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| assertEquals(numQueueEntries, size); |
| } |
| } |
| |
| /** |
| * To be used only for ParallelGatewaySender. |
| * |
| * @param senderId Id of the ParallelGatewaySender |
| * @param numQueueEntries Expected number of ParallelGatewaySenderQueue entries |
| */ |
| public static void checkPRQLocalSize(String senderId, final int numQueueEntries) { |
| GatewaySender sender = null; |
| for (GatewaySender s : cache.getGatewaySenders()) { |
| if (s.getId().equals(senderId)) { |
| sender = s; |
| break; |
| } |
| } |
| |
| if (sender.isParallel()) { |
| final Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| await().untilAsserted(() -> { |
| int size = 0; |
| for (RegionQueue q : queues) { |
| ConcurrentParallelGatewaySenderQueue prQ = (ConcurrentParallelGatewaySenderQueue) q; |
| size += prQ.localSize(); |
| } |
| assertEquals( |
| " Expected local queue entries: " + numQueueEntries + " but actual entries: " + size, |
| numQueueEntries, size); |
| }); |
| } |
| } |
| |
| /** |
| * To be used only for ParallelGatewaySender. |
| * |
| * @param senderId Id of the ParallelGatewaySender |
| */ |
| public static int getPRQLocalSize(String senderId) { |
| GatewaySender sender = null; |
| for (GatewaySender s : cache.getGatewaySenders()) { |
| if (s.getId().equals(senderId)) { |
| sender = s; |
| break; |
| } |
| } |
| if (sender.isParallel()) { |
| int totalSize = 0; |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| for (RegionQueue q : queues) { |
| ConcurrentParallelGatewaySenderQueue prQ = (ConcurrentParallelGatewaySenderQueue) q; |
| totalSize += prQ.localSize(); |
| } |
| return totalSize; |
| } |
| return -1; |
| } |
| |
| public static void doMultiThreadedPuts(String regionName, int numPuts) { |
| final AtomicInteger ai = new AtomicInteger(-1); |
| final ExecutorService execService = Executors.newFixedThreadPool(5, new ThreadFactory() { |
| AtomicInteger threadNum = new AtomicInteger(); |
| |
| @Override |
| public Thread newThread(final Runnable r) { |
| Thread result = new Thread(r, "Client Put Thread-" + threadNum.incrementAndGet()); |
| result.setDaemon(true); |
| return result; |
| } |
| }); |
| |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| |
| List<Callable<Object>> tasks = new ArrayList<Callable<Object>>(); |
| for (long i = 0; i < 5; i++) |
| tasks.add(new PutTask(r, ai, numPuts)); |
| |
| try { |
| List<Future<Object>> l = execService.invokeAll(tasks); |
| for (Future<Object> f : l) |
| f.get(); |
| } catch (InterruptedException e1) { // TODO: eats exception |
| e1.printStackTrace(); |
| } catch (ExecutionException e) { // TODO: eats exceptions |
| e.printStackTrace(); |
| } |
| execService.shutdown(); |
| } |
| |
| public static void validateRegionSize(String regionName, final int regionSize) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(CacheClosedException.class.getName()); |
| try { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| if (regionSize != r.keySet().size()) { |
| await() |
| .untilAsserted(() -> assertEquals( |
| "Expected region entries: " + regionSize + " but actual entries: " |
| + r.keySet().size() + " present region keyset " + r.keySet(), |
| regionSize, r.keySet().size())); |
| } |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| public static void validateAsyncEventListener(String asyncQueueId, final int expectedSize) { |
| AsyncEventListener theListener = null; |
| |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncQueue : asyncEventQueues) { |
| if (asyncQueueId.equals(asyncQueue.getId())) { |
| theListener = asyncQueue.getAsyncEventListener(); |
| } |
| } |
| |
| final Map eventsMap = ((MyAsyncEventListener) theListener).getEventsMap(); |
| assertNotNull(eventsMap); |
| await() |
| .untilAsserted(() -> assertEquals( |
| "Expected map entries: " + expectedSize + " but actual entries: " + eventsMap.size(), |
| expectedSize, eventsMap.size())); |
| } |
| |
| public static void waitForAsyncQueueToGetEmpty(String asyncQueueId) { |
| AsyncEventQueue theAsyncEventQueue = null; |
| |
| Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncChannel : asyncEventChannels) { |
| if (asyncQueueId.equals(asyncChannel.getId())) { |
| theAsyncEventQueue = asyncChannel; |
| } |
| } |
| |
| final GatewaySender sender = ((AsyncEventQueueImpl) theAsyncEventQueue).getSender(); |
| |
| if (sender.isParallel()) { |
| final Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| await().untilAsserted(() -> { |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| assertEquals("Expected queue size to be : " + 0 + " but actual entries: " + size, 0, size); |
| }); |
| } else { |
| await().untilAsserted(() -> { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| assertEquals("Expected queue size to be : " + 0 + " but actual entries: " + size, 0, size); |
| }); |
| } |
| } |
| |
| public static int getAsyncEventListenerMapSize(String asyncEventQueueId) { |
| AsyncEventListener theListener = null; |
| |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncQueue : asyncEventQueues) { |
| if (asyncEventQueueId.equals(asyncQueue.getId())) { |
| theListener = asyncQueue.getAsyncEventListener(); |
| } |
| } |
| |
| final Map eventsMap = ((MyAsyncEventListener) theListener).getEventsMap(); |
| assertNotNull(eventsMap); |
| logger.info("The events map size is " + eventsMap.size()); |
| return eventsMap.size(); |
| } |
| |
| public static void validateRegionSize_PDX(String regionName, final int regionSize) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| await().untilAsserted(() -> { |
| assertEquals("Expected region entries: " + regionSize + " but actual entries: " |
| + r.keySet().size() + " present region keyset " + r.keySet(), true, |
| (regionSize <= r.keySet().size())); |
| assertEquals("Expected region size: " + regionSize + " but actual size: " + r.size(), true, |
| (regionSize == r.size())); |
| }); |
| for (int i = 0; i < regionSize; i++) { |
| final int temp = i; |
| await() |
| .untilAsserted(() -> assertEquals( |
| "keySet = " + r.keySet() + " values() = " + r.values() + "Region Size = " + r.size(), |
| new SimpleClass(temp, (byte) temp), r.get("Key_" + temp))); |
| } |
| } |
| |
| public static void validateRegionSizeOnly_PDX(String regionName, final int regionSize) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| await() |
| .untilAsserted( |
| () -> assertEquals( |
| "Expected region entries: " + regionSize + " but actual entries: " |
| + r.keySet().size() + " present region keyset " + r.keySet(), |
| true, (regionSize <= r.keySet().size()))); |
| } |
| |
| public static void validateQueueSizeStat(String id, final int queueSize) { |
| final AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(id); |
| await() |
| .untilAsserted(() -> assertEquals(queueSize, sender.getEventQueueSize())); |
| assertEquals(queueSize, sender.getEventQueueSize()); |
| } |
| |
| public static void validateSecondaryQueueSizeStat(String id, final int queueSize) { |
| final AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(id); |
| await() |
| .untilAsserted(() -> assertEquals( |
| "Expected unprocessedEventMap is drained but actual is " |
| + sender.getStatistics().getUnprocessedEventMapSize(), |
| queueSize, sender.getStatistics().getUnprocessedEventMapSize())); |
| assertEquals(queueSize, sender.getStatistics().getUnprocessedEventMapSize()); |
| } |
| |
| /** |
| * This method is specifically written for pause and stop operations. This method validates that |
| * the region size remains same for at least minimum number of verification attempts and also it |
| * remains below a specified limit value. This validation will suffice for testing of pause/stop |
| * operations. |
| * |
| */ |
| public static void validateRegionSizeRemainsSame(String regionName, final int regionSizeLimit) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| WaitCriterion wc = new WaitCriterion() { |
| final int MIN_VERIFICATION_RUNS = 20; |
| int sameRegionSizeCounter = 0; |
| long previousSize = -1; |
| |
| @Override |
| public boolean done() { |
| if (r.keySet().size() == previousSize) { |
| sameRegionSizeCounter++; |
| int s = r.keySet().size(); |
| // if the sameRegionSizeCounter exceeds the minimum verification runs and regionSize is |
| // below specified limit, then return true |
| if (sameRegionSizeCounter >= MIN_VERIFICATION_RUNS && s <= regionSizeLimit) { |
| return true; |
| } else { |
| return false; |
| } |
| |
| } else { // current regionSize is not same as recorded previous regionSize |
| previousSize = r.keySet().size(); // update the previousSize variable with current region |
| // size |
| sameRegionSizeCounter = 0;// reset the sameRegionSizeCounter |
| return false; |
| } |
| } |
| |
| @Override |
| public String description() { |
| return "Expected region size to remain same below a specified limit but actual region size does not remain same or exceeded the specified limit " |
| + sameRegionSizeCounter + " :regionSize " + previousSize; |
| } |
| }; |
| await().untilAsserted(wc); |
| } |
| |
| public static String getRegionFullPath(String regionName) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| return r.getFullPath(); |
| } |
| |
| public static Integer getRegionSize(String regionName) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| return r.keySet().size(); |
| } |
| |
| public static void validateRegionContents(String regionName, final Map keyValues) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| await().untilAsserted(() -> { |
| boolean matchFlag = true; |
| for (Object key : keyValues.keySet()) { |
| if (!r.get(key).equals(keyValues.get(key))) { |
| logger.info("The values are for key " + " " + key + " " + r.get(key) + " in the map " |
| + keyValues.get(key)); |
| matchFlag = false; |
| } |
| } |
| assertEquals("Expected region entries doesn't match", true, matchFlag); |
| }); |
| } |
| |
| |
| |
| public static void doHeavyPuts(String regionName, int numPuts) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| // GatewaySender.DEFAULT_BATCH_SIZE * OBJECT_SIZE should be more than MAXIMUM_QUEUE_MEMORY |
| // to guarantee overflow |
| for (long i = 0; i < numPuts; i++) { |
| r.put(i, new byte[1024 * 1024]); |
| } |
| } |
| |
| public static void addCacheListenerAndDestroyRegion(String regionName) { |
| final Region region = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(region); |
| CacheListenerAdapter cl = new CacheListenerAdapter() { |
| @Override |
| public void afterCreate(EntryEvent event) { |
| if ((Long) event.getKey() == 99) { |
| region.destroyRegion(); |
| } |
| } |
| }; |
| region.getAttributesMutator().addCacheListener(cl); |
| } |
| |
| public static Boolean killSender(String senderId) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(CacheClosedException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| AbstractGatewaySender sender = (AbstractGatewaySender) getGatewaySender(senderId); |
| if (sender.isPrimary()) { |
| logger.info("Gateway sender is killed by a test"); |
| cache.getDistributedSystem().disconnect(); |
| return Boolean.TRUE; |
| } |
| return Boolean.FALSE; |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| exln.remove(); |
| } |
| } |
| |
| public static void killSender() { |
| logger.info("Gateway sender is going to be killed by a test"); |
| cache.close(); |
| cache.getDistributedSystem().disconnect(); |
| logger.info("Gateway sender is killed by a test"); |
| } |
| |
| public static void checkAllSiteMetaData( |
| Map<Integer, Set<InetSocketAddress>> dsIdToLocatorAddresses, final int siteSizeToCheck) { |
| List<Locator> locatorsConfigured = Locator.getLocators(); |
| Locator locator = locatorsConfigured.get(0); |
| await().untilAsserted(() -> { |
| Map<Integer, Set<DistributionLocatorId>> allSiteMetaData = |
| ((InternalLocator) locator).getLocatorMembershipListener().getAllLocatorsInfo(); |
| assertThat(allSiteMetaData.size()).isEqualTo(siteSizeToCheck); |
| for (Map.Entry<Integer, Set<InetSocketAddress>> entry : dsIdToLocatorAddresses.entrySet()) { |
| Set<DistributionLocatorId> foundLocatorIds = allSiteMetaData.get(entry.getKey()); |
| Set<InetSocketAddress> expectedLocators = entry.getValue(); |
| final Set<InetSocketAddress> foundLocators = foundLocatorIds.stream() |
| .map(distributionLocatorId -> new InetSocketAddress( |
| distributionLocatorId.getHostnameForClients(), distributionLocatorId.getPort())) |
| .collect(Collectors.toSet()); |
| assertEquals(expectedLocators, foundLocators); |
| } |
| }); |
| } |
| |
| public static Long checkAllSiteMetaDataFor3Sites(final Map<Integer, Set<String>> dsVsPort) { |
| await() |
| .untilAsserted( |
| () -> assertEquals("System is not initialized", true, (getSystemStatic() != null))); |
| List<Locator> locatorsConfigured = Locator.getLocators(); |
| Locator locator = locatorsConfigured.get(0); |
| LocatorMembershipListener listener = ((InternalLocator) locator).getLocatorMembershipListener(); |
| if (listener == null) { |
| fail( |
| "No locator membership listener available. WAN is likely not enabled. Is this test in the WAN project?"); |
| } |
| final Map<Integer, Set<DistributionLocatorId>> allSiteMetaData = listener.getAllLocatorsInfo(); |
| System.out.println("allSiteMetaData : " + allSiteMetaData); |
| |
| await().untilAsserted(() -> { |
| assertEquals(true, (dsVsPort.size() == allSiteMetaData.size())); |
| boolean completeFlag = true; |
| for (Map.Entry<Integer, Set<String>> entry : dsVsPort.entrySet()) { |
| Set<DistributionLocatorId> locators = allSiteMetaData.get(entry.getKey()); |
| for (String locatorInMetaData : entry.getValue()) { |
| DistributionLocatorId locatorId = new DistributionLocatorId(locatorInMetaData); |
| if (!locators.contains(locatorId)) { |
| completeFlag = false; |
| break; |
| } |
| } |
| if (false == completeFlag) { |
| break; |
| } |
| } |
| assertEquals( |
| "Expected site Metadata: " + dsVsPort + " but actual meta data: " + allSiteMetaData, true, |
| completeFlag); |
| }); |
| return System.currentTimeMillis(); |
| } |
| |
| public static void putRemoteSiteLocators(int remoteDsId, Set<String> remoteLocators) { |
| List<Locator> locatorsConfigured = Locator.getLocators(); |
| Locator locator = locatorsConfigured.get(0); |
| if (remoteLocators != null) { |
| // Add fake remote locators to the locators map |
| ((InternalLocator) locator).getLocatorMembershipListener().getAllServerLocatorsInfo() |
| .put(remoteDsId, remoteLocators); |
| } |
| } |
| |
| public static void checkLocatorsinSender(String senderId, InetSocketAddress locatorToWaitFor) |
| throws InterruptedException { |
| |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| MyLocatorCallback callback = |
| (MyLocatorCallback) ((AbstractGatewaySender) sender).getLocatorDiscoveryCallback(); |
| |
| boolean discovered = callback.waitForDiscovery(locatorToWaitFor, MAX_WAIT); |
| assertTrue("Waited " + MAX_WAIT + " for " + locatorToWaitFor |
| + " to be discovered on client. List is now: " + callback.getDiscovered(), discovered); |
| } |
| |
| public static void validateQueueContents(final String senderId, final int regionSize) { |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| IgnoredException exp2 = |
| IgnoredException.addIgnoredException(GatewaySenderException.class.getName()); |
| try { |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| if (!sender.isParallel()) { |
| final Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| await().untilAsserted(() -> { |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| assertEquals("Expected queue entries: " + regionSize + " but actual entries: " + size, |
| regionSize, size); |
| }); |
| } else if (sender.isParallel()) { |
| final RegionQueue regionQueue; |
| regionQueue = ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| await().untilAsserted(() -> assertEquals( |
| "Expected queue entries: " + regionSize + " but actual entries: " + regionQueue.size(), |
| regionSize, regionQueue.size())); |
| } |
| } finally { |
| exp1.remove(); |
| exp2.remove(); |
| } |
| |
| } |
| |
| // Ensure that the sender's queue(s) have been closed. |
| public static void validateQueueClosedForConcurrentSerialGatewaySender(final String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| final Set<RegionQueue> regionQueue; |
| if (sender instanceof AbstractGatewaySender) { |
| regionQueue = ((AbstractGatewaySender) sender).getQueues(); |
| } else { |
| regionQueue = null; |
| } |
| assertEquals(null, regionQueue); |
| } |
| |
| public static void validateQueueContentsForConcurrentSerialGatewaySender(final String senderId, |
| final int regionSize) { |
| GatewaySender sender = getGatewaySender(senderId); |
| final Set<RegionQueue> regionQueue; |
| if (!sender.isParallel()) { |
| regionQueue = ((AbstractGatewaySender) sender).getQueues(); |
| } else { |
| regionQueue = null; |
| } |
| await().untilAsserted(() -> { |
| int size = 0; |
| for (RegionQueue q : regionQueue) { |
| size += q.size(); |
| } |
| assertEquals(true, regionSize == size); |
| }); |
| } |
| |
| public static Integer getSecondaryQueueContentSize(final String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| AbstractGatewaySender abstractSender = (AbstractGatewaySender) sender; |
| int size = abstractSender.getSecondaryEventQueueSize(); |
| return size; |
| } |
| |
| public static String displayQueueContent(final RegionQueue queue) { |
| if (queue instanceof ParallelGatewaySenderQueue) { |
| ParallelGatewaySenderQueue pgsq = (ParallelGatewaySenderQueue) queue; |
| return pgsq.displayContent(); |
| } else if (queue instanceof ConcurrentParallelGatewaySenderQueue) { |
| ConcurrentParallelGatewaySenderQueue pgsq = (ConcurrentParallelGatewaySenderQueue) queue; |
| return pgsq.displayContent(); |
| } |
| return null; |
| } |
| |
| public static String displaySerialQueueContent(final AbstractGatewaySender sender) { |
| StringBuilder message = new StringBuilder(); |
| message.append("Is Primary: ").append(sender.isPrimary()).append(", ").append("Queue Size: ") |
| .append(sender.getEventQueueSize()); |
| |
| if (sender.getQueues() != null) { |
| message.append(", ").append("Queue Count: ").append(sender.getQueues().size()); |
| Stream<Object> stream = sender.getQueues().stream() |
| .map(regionQueue -> ((SerialGatewaySenderQueue) regionQueue).displayContent()); |
| |
| List<Object> list = stream.collect(Collectors.toList()); |
| message.append(", ").append("Keys: ").append(list.toString()); |
| } |
| |
| AbstractGatewaySenderEventProcessor abstractProcessor = sender.getEventProcessor(); |
| if (abstractProcessor == null) { |
| message.append(", ").append("Null Event Processor: "); |
| } |
| if (!sender.isPrimary()) { |
| message.append("\n").append("Unprocessed Events: ") |
| .append(abstractProcessor.printUnprocessedEvents()).append("\n"); |
| message.append("\n").append("Unprocessed Tokens: ") |
| .append(abstractProcessor.printUnprocessedTokens()).append("\n"); |
| } |
| |
| return message.toString(); |
| } |
| |
| public static Integer getQueueContentSize(final String senderId) { |
| return getQueueContentSize(senderId, false); |
| } |
| |
| public static Integer getQueueContentSize(final String senderId, boolean includeSecondary) { |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| if (!sender.isParallel()) { |
| // if sender is serial, the queues will be all primary or all secondary at one member |
| final Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| return size; |
| } else if (sender.isParallel()) { |
| RegionQueue regionQueue = null; |
| regionQueue = ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| if (regionQueue instanceof ConcurrentParallelGatewaySenderQueue) { |
| return ((ConcurrentParallelGatewaySenderQueue) regionQueue).localSize(includeSecondary); |
| } else if (regionQueue instanceof ParallelGatewaySenderQueue) { |
| return ((ParallelGatewaySenderQueue) regionQueue).localSize(includeSecondary); |
| } else { |
| fail("Not implemented yet"); |
| } |
| } |
| fail("Not yet implemented?"); |
| return 0; |
| } |
| |
| public static void validateParallelSenderQueueBucketSize(final String senderId, |
| final int bucketSize) { |
| GatewaySender sender = getGatewaySender(senderId); |
| RegionQueue regionQueue = |
| ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| Set<BucketRegion> buckets = ((PartitionedRegion) regionQueue.getRegion()).getDataStore() |
| .getAllLocalPrimaryBucketRegions(); |
| for (BucketRegion bucket : buckets) { |
| assertEquals( |
| "Expected bucket entries for bucket " + bucket.getId() + " is different than actual.", |
| bucketSize, bucket.keySet().size()); |
| } |
| } |
| |
| public static void validateParallelSenderQueueAllBucketsDrained(final String senderId) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(RegionDestroyedException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| GatewaySender sender = getGatewaySender(senderId); |
| final AbstractGatewaySender abstractSender = (AbstractGatewaySender) sender; |
| RegionQueue queue = abstractSender.getEventProcessor().queue; |
| await().untilAsserted(() -> { |
| assertEquals("Expected events in all primary queues are drained but actual is " |
| + abstractSender.getEventQueueSize() + ". Queue content is: " |
| + displayQueueContent(queue), 0, abstractSender.getEventQueueSize()); |
| }); |
| assertEquals("Expected events in all primary queues after drain is 0", 0, |
| abstractSender.getEventQueueSize()); |
| await().untilAsserted(() -> { |
| assertEquals("Expected events in all secondary queues are drained but actual is " |
| + abstractSender.getSecondaryEventQueueSize() + ". Queue content is: " |
| + displayQueueContent(queue), 0, abstractSender.getSecondaryEventQueueSize()); |
| }); |
| assertEquals("Expected events in all secondary queues after drain is 0", 0, |
| abstractSender.getSecondaryEventQueueSize()); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| |
| } |
| |
| public static Integer validateAfterAck(final String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| final MyGatewayEventFilter_AfterAck filter = |
| (MyGatewayEventFilter_AfterAck) sender.getGatewayEventFilters().get(0); |
| return filter.getAckList().size(); |
| } |
| |
| public static int verifyAndGetEventsDispatchedByConcurrentDispatchers(final String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| ConcurrentParallelGatewaySenderEventProcessor cProc = |
| (ConcurrentParallelGatewaySenderEventProcessor) ((AbstractGatewaySender) sender) |
| .getEventProcessor(); |
| if (cProc == null) |
| return 0; |
| |
| int totalDispatched = 0; |
| for (ParallelGatewaySenderEventProcessor lProc : cProc.getProcessors()) { |
| totalDispatched += lProc.getNumEventsDispatched(); |
| } |
| assertTrue(totalDispatched > 0); |
| return totalDispatched; |
| } |
| |
| public static Long getNumberOfEntriesOverflownToDisk(final String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| long numEntries = 0; |
| if (sender.isParallel()) { |
| RegionQueue regionQueue; |
| regionQueue = ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| numEntries = ((ConcurrentParallelGatewaySenderQueue) regionQueue) |
| .getNumEntriesOverflowOnDiskTestOnly(); |
| } |
| return numEntries; |
| } |
| |
| public static Long getNumberOfEntriesInVM(final String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| RegionQueue regionQueue; |
| long numEntries = 0; |
| if (sender.isParallel()) { |
| regionQueue = ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| numEntries = ((ConcurrentParallelGatewaySenderQueue) regionQueue).getNumEntriesInVMTestOnly(); |
| } |
| return numEntries; |
| } |
| |
| public static void verifyTmpDroppedEventSize(String senderId, int size) { |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| AbstractGatewaySender ags = (AbstractGatewaySender) sender; |
| await().untilAsserted(() -> assertEquals("Expected tmpDroppedEvents size: " + size |
| + " but actual size: " + ags.getTmpDroppedEventSize(), size, ags.getTmpDroppedEventSize())); |
| } |
| |
| /** |
| * Checks that the bucketToTempQueueMap for a partitioned region |
| * that holds events for buckets that are not available locally, is empty. |
| */ |
| public static void validateEmptyBucketToTempQueueMap(String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| int size = 0; |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| for (Object queue : queues) { |
| PartitionedRegion region = |
| (PartitionedRegion) ((ConcurrentParallelGatewaySenderQueue) queue).getRegion(); |
| int buckets = region.getTotalNumberOfBuckets(); |
| for (int bucket = 0; bucket < buckets; bucket++) { |
| BlockingQueue<GatewaySenderEventImpl> newQueue = |
| ((ConcurrentParallelGatewaySenderQueue) queue).getBucketTmpQueue(bucket); |
| if (newQueue != null) { |
| size += newQueue.size(); |
| } |
| } |
| } |
| |
| final int finalSize = size; |
| assertEquals("Expected elements in TempQueueMap: " + 0 |
| + " but actual size: " + finalSize, 0, finalSize); |
| |
| } |
| |
| private static GatewaySender getGatewaySender(String senderId) { |
| Set<GatewaySender> senders = cache.getGatewaySenders(); |
| GatewaySender sender = null; |
| for (GatewaySender s : senders) { |
| if (s.getId().equals(senderId)) { |
| sender = s; |
| break; |
| } |
| } |
| return sender; |
| } |
| |
| public static void verifyQueueSize(String senderId, int size) { |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| if (!sender.isParallel()) { |
| final Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| int queueSize = 0; |
| for (RegionQueue q : queues) { |
| queueSize += q.size(); |
| } |
| |
| assertEquals("verifyQueueSize failed for sender " + senderId, size, queueSize); |
| } else if (sender.isParallel()) { |
| RegionQueue regionQueue = |
| ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| assertEquals("verifyQueueSize failed for sender " + senderId, size, regionQueue.size()); |
| } |
| } |
| |
| public static void verifyRegionQueueNotEmpty(String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| |
| if (!sender.isParallel()) { |
| final Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| int queueSize = 0; |
| for (RegionQueue q : queues) { |
| queueSize += q.size(); |
| } |
| assertTrue(queues.size() > 0); |
| assertTrue(queueSize > 0); |
| } else if (sender.isParallel()) { |
| RegionQueue regionQueue = |
| ((AbstractGatewaySender) sender).getQueues().toArray(new RegionQueue[1])[0]; |
| assertTrue(regionQueue.size() > 0); |
| } |
| } |
| |
| public static void waitForConcurrentSerialSenderQueueToDrain(String senderId) { |
| Set<GatewaySender> senders = cache.getGatewaySenders(); |
| GatewaySender sender = |
| senders.stream().filter(s -> s.getId().equals(senderId)).findFirst().get(); |
| |
| await().untilAsserted(() -> { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| for (RegionQueue q : queues) { |
| assertEquals(0, q.size()); |
| } |
| }); |
| } |
| |
| /** |
| * Test methods for sender operations |
| * |
| */ |
| public static void verifySenderPausedState(String senderId) { |
| GatewaySender sender = cache.getGatewaySender(senderId); |
| assertTrue(sender.isPaused()); |
| } |
| |
| public static void verifySenderResumedState(String senderId) { |
| GatewaySender sender = cache.getGatewaySender(senderId); |
| assertFalse(sender.isPaused()); |
| assertTrue(sender.isRunning()); |
| } |
| |
| public static void verifySenderStoppedState(String senderId) { |
| GatewaySender sender = cache.getGatewaySender(senderId); |
| assertFalse(sender.isRunning()); |
| } |
| |
| public static void verifySenderRunningState(String senderId) { |
| GatewaySender sender = cache.getGatewaySender(senderId); |
| assertTrue(sender.isRunning()); |
| } |
| |
| public static void verifySenderConnectedState(String senderId, boolean shouldBeConnected) { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId); |
| if (shouldBeConnected) { |
| assertThat(sender.getEventProcessor().getDispatcher().isConnectedToRemote()).isTrue(); |
| } else { |
| assertThat(sender.getEventProcessor().getDispatcher().isConnectedToRemote()).isFalse(); |
| } |
| } |
| |
| public static void verifyPool(String senderId, boolean poolShouldExist, |
| int expectedPoolLocatorsSize) { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender(senderId); |
| PoolImpl pool = sender.getProxy(); |
| if (poolShouldExist) { |
| assertNotNull(pool); |
| assertEquals(expectedPoolLocatorsSize, pool.getLocators().size()); |
| } else { |
| assertNull(pool); |
| } |
| } |
| |
| public static void removeSenderFromTheRegion(String senderId, String regionName) { |
| Region region = cache.getRegion(regionName); |
| assertNotNull(region); |
| region.getAttributesMutator().removeGatewaySenderId(senderId); |
| } |
| |
| public static void destroySender(String senderId) { |
| GatewaySender sender = getGatewaySender(senderId); |
| sender.destroy(); |
| } |
| |
| public static void verifySenderDestroyed(String senderId, boolean isParallel) { |
| AbstractGatewaySender sender = (AbstractGatewaySender) getGatewaySender(senderId); |
| assertNull(sender); |
| |
| String queueRegionNameSuffix = null; |
| if (isParallel) { |
| queueRegionNameSuffix = ParallelGatewaySenderQueue.QSTRING; |
| } else { |
| queueRegionNameSuffix = "_SERIAL_GATEWAY_SENDER_QUEUE"; |
| } |
| |
| Set<InternalRegion> allRegions = ((GemFireCacheImpl) cache).getAllRegions(); |
| for (InternalRegion region : allRegions) { |
| if (region.getName().indexOf(senderId + queueRegionNameSuffix) != -1) { |
| fail("Region underlying the sender is not destroyed."); |
| } |
| } |
| } |
| |
| public static void destroyAsyncEventQueue(String id) { |
| AsyncEventQueueImpl aeq = (AsyncEventQueueImpl) cache.getAsyncEventQueue(id); |
| assertNotNull(aeq); |
| aeq.destroy(); |
| } |
| |
| protected static void verifyListenerEvents(final long expectedNumEvents) { |
| await() |
| .until(() -> listener1.getNumEvents() == expectedNumEvents); |
| } |
| |
| protected Integer[] createLNAndNYLocators() { |
| Integer lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1)); |
| Integer nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort)); |
| return new Integer[] {lnPort, nyPort}; |
| } |
| |
| protected void validateRegionSizes(String regionName, int expectedRegionSize, VM... vms) { |
| for (VM vm : vms) { |
| vm.invoke(() -> validateRegionSize(regionName, expectedRegionSize)); |
| } |
| } |
| |
| public static class MyLocatorCallback extends LocatorDiscoveryCallbackAdapter { |
| |
| private final Set discoveredLocators = new HashSet(); |
| |
| private final Set removedLocators = new HashSet(); |
| |
| @Override |
| public synchronized void locatorsDiscovered(List locators) { |
| discoveredLocators.addAll(locators); |
| notifyAll(); |
| } |
| |
| @Override |
| public synchronized void locatorsRemoved(List locators) { |
| removedLocators.addAll(locators); |
| notifyAll(); |
| } |
| |
| public boolean waitForDiscovery(InetSocketAddress locator, long time) |
| throws InterruptedException { |
| return waitFor(discoveredLocators, locator, time); |
| } |
| |
| |
| private synchronized boolean waitFor(Set set, InetSocketAddress locator, long time) |
| throws InterruptedException { |
| long remaining = time; |
| long endTime = System.currentTimeMillis() + time; |
| while (!set.contains(locator) && remaining >= 0) { |
| wait(remaining); |
| remaining = endTime - System.currentTimeMillis(); |
| } |
| return set.contains(locator); |
| } |
| |
| public synchronized Set getDiscovered() { |
| return new HashSet(discoveredLocators); |
| } |
| |
| } |
| |
| protected static class PutTask implements Callable { |
| private final Region region; |
| |
| private final AtomicInteger key_value; |
| |
| private final int numPuts; |
| |
| public PutTask(Region region, AtomicInteger key_value, int numPuts) { |
| this.region = region; |
| this.key_value = key_value; |
| this.numPuts = numPuts; |
| } |
| |
| @Override |
| public Object call() throws Exception { |
| while (true) { |
| int key = key_value.incrementAndGet(); |
| if (key < numPuts) { |
| region.put(key, key); |
| } else { |
| break; |
| } |
| } |
| return null; |
| } |
| } |
| |
| public static class MyGatewayEventFilter implements GatewayEventFilter, Serializable { |
| |
| String Id = "MyGatewayEventFilter"; |
| |
| boolean beforeEnqueueInvoked; |
| boolean beforeTransmitInvoked; |
| boolean afterAckInvoked; |
| |
| public MyGatewayEventFilter() {} |
| |
| @Override |
| public boolean beforeEnqueue(GatewayQueueEvent event) { |
| this.beforeEnqueueInvoked = true; |
| return !((Long) event.getKey() >= 500 && (Long) event.getKey() < 600); |
| } |
| |
| @Override |
| public boolean beforeTransmit(GatewayQueueEvent event) { |
| this.beforeTransmitInvoked = true; |
| return !((Long) event.getKey() >= 600 && (Long) event.getKey() < 700); |
| } |
| |
| @Override |
| public void close() { |
| // TODO Auto-generated method stub |
| |
| } |
| |
| public String toString() { |
| return Id; |
| } |
| |
| @Override |
| public void afterAcknowledgement(GatewayQueueEvent event) { |
| this.afterAckInvoked = true; |
| // TODO Auto-generated method stub |
| } |
| |
| public boolean equals(Object obj) { |
| if (this == obj) { |
| return true; |
| } |
| if (!(obj instanceof MyGatewayEventFilter)) |
| return false; |
| MyGatewayEventFilter filter = (MyGatewayEventFilter) obj; |
| return this.Id.equals(filter.Id); |
| } |
| } |
| |
| public static class MyGatewayEventFilter_AfterAck implements GatewayEventFilter, Serializable { |
| |
| String Id = "MyGatewayEventFilter_AfterAck"; |
| |
| ConcurrentSkipListSet<Long> ackList = new ConcurrentSkipListSet<Long>(); |
| |
| public MyGatewayEventFilter_AfterAck() {} |
| |
| @Override |
| public boolean beforeEnqueue(GatewayQueueEvent event) { |
| return true; |
| } |
| |
| @Override |
| public boolean beforeTransmit(GatewayQueueEvent event) { |
| return true; |
| } |
| |
| @Override |
| public void close() { |
| // TODO Auto-generated method stub |
| |
| } |
| |
| public String toString() { |
| return Id; |
| } |
| |
| @Override |
| public void afterAcknowledgement(GatewayQueueEvent event) { |
| ackList.add((Long) event.getKey()); |
| } |
| |
| public Set getAckList() { |
| return ackList; |
| } |
| |
| public boolean equals(Object obj) { |
| if (this == obj) { |
| return true; |
| } |
| if (!(obj instanceof MyGatewayEventFilter)) |
| return false; |
| MyGatewayEventFilter filter = (MyGatewayEventFilter) obj; |
| return this.Id.equals(filter.Id); |
| } |
| } |
| |
| public static class PDXGatewayEventFilter implements GatewayEventFilter, Serializable { |
| |
| String Id = "PDXGatewayEventFilter"; |
| |
| public int beforeEnqueueInvoked; |
| public int beforeTransmitInvoked; |
| public int afterAckInvoked; |
| |
| public PDXGatewayEventFilter() {} |
| |
| @Override |
| public boolean beforeEnqueue(GatewayQueueEvent event) { |
| System.out.println("Invoked enqueue for " + event); |
| this.beforeEnqueueInvoked++; |
| return true; |
| } |
| |
| @Override |
| public boolean beforeTransmit(GatewayQueueEvent event) { |
| System.out.println("Invoked transmit for " + event); |
| this.beforeTransmitInvoked++; |
| return true; |
| } |
| |
| @Override |
| public void close() { |
| // TODO Auto-generated method stub |
| |
| } |
| |
| public String toString() { |
| return Id; |
| } |
| |
| @Override |
| public void afterAcknowledgement(GatewayQueueEvent event) { |
| System.out.println("Invoked afterAck for " + event); |
| this.afterAckInvoked++; |
| // TODO Auto-generated method stub |
| } |
| |
| public boolean equals(Object obj) { |
| if (this == obj) { |
| return true; |
| } |
| if (!(obj instanceof MyGatewayEventFilter)) |
| return false; |
| MyGatewayEventFilter filter = (MyGatewayEventFilter) obj; |
| return this.Id.equals(filter.Id); |
| } |
| } |
| |
| @Override |
| public final void preTearDown() throws Exception { |
| cleanupVM(); |
| List<AsyncInvocation> invocations = new ArrayList<AsyncInvocation>(); |
| final Host host = getHost(0); |
| for (int i = 0; i < host.getVMCount(); i++) { |
| invocations.add(host.getVM(i).invokeAsync(() -> WANTestBase.cleanupVM())); |
| } |
| for (AsyncInvocation invocation : invocations) { |
| invocation.await(); |
| } |
| } |
| |
| public static void cleanupVM() throws IOException { |
| if (Locator.hasLocator()) { |
| Locator.getLocator().stop(); |
| } |
| closeCache(); |
| JUnit4DistributedTestCase.cleanDiskDirs(); |
| } |
| |
| public static void closeCache() { |
| if (cache != null && !cache.isClosed()) { |
| cache.close(); |
| cache.getDistributedSystem().disconnect(); |
| cache = null; |
| } else { |
| WANTestBase test = new WANTestBase(); |
| if (test.isConnectedToDS()) { |
| test.getSystem().disconnect(); |
| } |
| } |
| } |
| |
| public static void deletePDXDir() throws IOException { |
| File pdxDir = new File(CacheTestCase.getDiskDir(), "pdx"); |
| FileUtils.deleteDirectory(pdxDir); |
| } |
| |
| public static void shutdownLocator() { |
| WANTestBase test = new WANTestBase(); |
| test.getSystem().disconnect(); |
| } |
| |
| public static void printEventListenerMap() { |
| ((MyGatewaySenderEventListener) eventListener1).printMap(); |
| } |
| |
| @Override |
| public Properties getDistributedSystemProperties() { |
| // For now all WANTestBase tests allocate off-heap memory even though |
| // many of them never use it. |
| // The problem is that WANTestBase has static methods that create instances |
| // of WANTestBase (instead of instances of the subclass). So we can't override |
| // this method so that only the off-heap subclasses allocate off heap memory. |
| Properties props = new Properties(); |
| props.setProperty(OFF_HEAP_MEMORY_SIZE, "200m"); |
| |
| return props; |
| } |
| |
| /** |
| * Returns true if the test should create off-heap regions. OffHeap tests should over-ride this |
| * method and return false. |
| */ |
| public boolean isOffHeap() { |
| return false; |
| } |
| |
| /** |
| * Checks whether a Async Queue MBean is created or not |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void checkAsyncQueueMBean(final VM vm, final boolean shouldExist) { |
| SerializableRunnable checkAsyncQueueMBean = |
| new SerializableRunnable("Check Async Queue MBean") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| AsyncEventQueueMXBean bean = service.getLocalAsyncEventQueueMXBean("pn"); |
| if (shouldExist) { |
| assertNotNull(bean); |
| } else { |
| assertNull(bean); |
| } |
| } |
| }; |
| vm.invoke(checkAsyncQueueMBean); |
| } |
| |
| /** |
| * Checks Proxy GatewayReceiver |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void checkProxyReceiver(final VM vm, final DistributedMember senderMember) { |
| SerializableRunnable checkProxySender = new SerializableRunnable("Check Proxy Receiver") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| GatewayReceiverMXBean bean = null; |
| try { |
| bean = MBeanUtil.getGatewayReceiverMbeanProxy(senderMember); |
| } catch (Exception e) { |
| fail("Could not obtain Sender Proxy in desired time " + e); |
| } |
| assertNotNull(bean); |
| final ObjectName receiverMBeanName = service.getGatewayReceiverMBeanName(senderMember); |
| try { |
| MBeanUtil.printBeanDetails(receiverMBeanName); |
| } catch (Exception e) { |
| fail("Error while Printing Bean Details " + e); |
| } |
| |
| } |
| }; |
| vm.invoke(checkProxySender); |
| } |
| |
| /** |
| * Checks Proxy GatewaySender |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void checkProxySender(final VM vm, final DistributedMember senderMember) { |
| SerializableRunnable checkProxySender = new SerializableRunnable("Check Proxy Sender") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| GatewaySenderMXBean bean = null; |
| try { |
| bean = MBeanUtil.getGatewaySenderMbeanProxy(senderMember, "pn"); |
| } catch (Exception e) { |
| fail("Could not obtain Sender Proxy in desired time " + e); |
| } |
| assertNotNull(bean); |
| final ObjectName senderMBeanName = service.getGatewaySenderMBeanName(senderMember, "pn"); |
| try { |
| MBeanUtil.printBeanDetails(senderMBeanName); |
| } catch (Exception e) { |
| fail("Error while Printing Bean Details " + e); |
| } |
| |
| if (service.isManager()) { |
| DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean(); |
| await().untilAsserted(() -> { |
| Map<String, Boolean> dsMap = dsBean.viewRemoteClusterStatus(); |
| dsMap.entrySet().stream() |
| .forEach(entry -> assertTrue("Should be true " + entry.getKey(), entry.getValue())); |
| }); |
| } |
| |
| } |
| }; |
| vm.invoke(checkProxySender); |
| } |
| |
| /** |
| * Checks whether a GatewayReceiverMBean is created or not |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void checkReceiverMBean(final VM vm) { |
| SerializableRunnable checkMBean = new SerializableRunnable("Check Receiver MBean") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| GatewayReceiverMXBean bean = service.getLocalGatewayReceiverMXBean(); |
| assertNotNull(bean); |
| } |
| }; |
| vm.invoke(checkMBean); |
| } |
| |
| @SuppressWarnings("serial") |
| public static void checkReceiverNavigationAPIS(final VM vm, |
| final DistributedMember receiverMember) { |
| SerializableRunnable checkNavigationAPIS = |
| new SerializableRunnable("Check Receiver Navigation APIs") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| DistributedSystemMXBean bean = service.getDistributedSystemMXBean(); |
| ObjectName expectedName = service.getGatewayReceiverMBeanName(receiverMember); |
| try { |
| await("wait for member to be added to DistributedSystemBridge membership list") |
| .untilAsserted( |
| () -> assertThat(bean.fetchGatewayReceiverObjectName(receiverMember.getId())) |
| .isNotNull()); |
| ObjectName actualName = bean.fetchGatewayReceiverObjectName(receiverMember.getId()); |
| assertEquals(expectedName, actualName); |
| } catch (Exception e) { |
| fail("Receiver Navigation Failed " + e); |
| } |
| |
| assertEquals(1, bean.listGatewayReceiverObjectNames().length); |
| |
| } |
| }; |
| vm.invoke(checkNavigationAPIS); |
| } |
| |
| /** |
| * Checks whether a GatewayReceiverMBean is created or not |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void checkSenderMBean(final VM vm, final String regionPath, boolean connected) { |
| SerializableRunnable checkMBean = new SerializableRunnable("Check Sender MBean") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| |
| GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn"); |
| assertNotNull(bean); |
| await() |
| .untilAsserted(() -> assertEquals(connected, bean.isConnected())); |
| |
| ObjectName regionBeanName = service.getRegionMBeanName( |
| cache.getDistributedSystem().getDistributedMember(), SEPARATOR + regionPath); |
| RegionMXBean rBean = service.getMBeanInstance(regionBeanName, RegionMXBean.class); |
| assertTrue(rBean.isGatewayEnabled()); |
| |
| |
| } |
| }; |
| vm.invoke(checkMBean); |
| } |
| |
| @SuppressWarnings("serial") |
| public static void checkSenderNavigationAPIS(final VM vm, final DistributedMember senderMember) { |
| SerializableRunnable checkNavigationAPIS = |
| new SerializableRunnable("Check Sender Navigation APIs") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| DistributedSystemMXBean bean = service.getDistributedSystemMXBean(); |
| ObjectName expectedName = service.getGatewaySenderMBeanName(senderMember, "pn"); |
| try { |
| ObjectName actualName = bean.fetchGatewaySenderObjectName(senderMember.getId(), "pn"); |
| assertEquals(expectedName, actualName); |
| } catch (Exception e) { |
| fail("Sender Navigation Failed " + e); |
| } |
| |
| assertEquals(2, bean.listGatewaySenderObjectNames().length); |
| try { |
| assertEquals(1, bean.listGatewaySenderObjectNames(senderMember.getId()).length); |
| } catch (Exception e) { |
| fail("Sender Navigation Failed " + e); |
| } |
| |
| } |
| }; |
| vm.invoke(checkNavigationAPIS); |
| } |
| |
| /** |
| * start a gateway sender |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void startGatewaySender(final VM vm) { |
| SerializableRunnable stopGatewaySender = new SerializableRunnable("Start Gateway Sender") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn"); |
| assertNotNull(bean); |
| bean.start(); |
| assertTrue(bean.isRunning()); |
| } |
| }; |
| vm.invoke(stopGatewaySender); |
| } |
| |
| /** |
| * stops a gateway sender |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void stopGatewaySender(final VM vm) { |
| SerializableRunnable stopGatewaySender = new SerializableRunnable("Stop Gateway Sender") { |
| @Override |
| public void run() { |
| ManagementService service = ManagementService.getManagementService(cache); |
| GatewaySenderMXBean bean = service.getLocalGatewaySenderMXBean("pn"); |
| assertNotNull(bean); |
| bean.stop(); |
| assertFalse(bean.isRunning()); |
| } |
| }; |
| vm.invoke(stopGatewaySender); |
| } |
| |
| /** |
| * Checks Proxy Async Queue |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void checkProxyAsyncQueue(final VM vm, final DistributedMember senderMember, |
| final boolean shouldExist) { |
| SerializableRunnable checkProxyAsyncQueue = |
| new SerializableRunnable("Check Proxy Async Queue") { |
| @Override |
| public void run() { |
| SystemManagementService service = |
| (SystemManagementService) ManagementService.getManagementService(cache); |
| final ObjectName queueMBeanName = |
| service.getAsyncEventQueueMBeanName(senderMember, "pn"); |
| AsyncEventQueueMXBean bean = null; |
| if (shouldExist) { |
| // Verify the MBean proxy exists |
| try { |
| bean = MBeanUtil.getAsyncEventQueueMBeanProxy(senderMember, "pn"); |
| } catch (Exception e) { |
| fail("Could not obtain Sender Proxy in desired time " + e); |
| } |
| assertNotNull(bean); |
| |
| try { |
| MBeanUtil.printBeanDetails(queueMBeanName); |
| } catch (Exception e) { |
| fail("Error while Printing Bean Details " + e); |
| } |
| } else { |
| // Verify the MBean proxy doesn't exist |
| bean = service.getMBeanProxy(queueMBeanName, AsyncEventQueueMXBean.class); |
| assertNull(bean); |
| } |
| } |
| }; |
| vm.invoke(checkProxyAsyncQueue); |
| } |
| |
| public static DistributedMember getMember() { |
| return ((GemFireCacheImpl) cache).getMyId(); |
| } |
| |
| public static ManagementService getManagementService() { |
| return ManagementService.getManagementService(cache); |
| } |
| |
| /** |
| * Checks Proxy GatewaySender |
| * |
| * @param vm reference to VM |
| */ |
| @SuppressWarnings("serial") |
| public static void checkRemoteClusterStatus(final VM vm, final DistributedMember senderMember) { |
| SerializableRunnable checkProxySender = new SerializableRunnable("DS Map Size") { |
| @Override |
| public void run() { |
| await().untilAsserted(() -> { |
| final ManagementService service = ManagementService.getManagementService(cache); |
| final DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean(); |
| assertEquals( |
| "Failed while waiting for getDistributedSystemMXBean to complete and get results", |
| true, dsBean != null); |
| }); |
| ManagementService service = ManagementService.getManagementService(cache); |
| final DistributedSystemMXBean dsBean = service.getDistributedSystemMXBean(); |
| assertNotNull(dsBean); |
| Map<String, Boolean> dsMap = dsBean.viewRemoteClusterStatus(); |
| logger.info("Ds Map is: " + dsMap.size()); |
| assertNotNull(dsMap); |
| assertEquals(true, dsMap.size() > 0); |
| } |
| }; |
| vm.invoke(checkProxySender); |
| } |
| |
| |
| } |
