You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by rv...@apache.org on 2016/01/20 03:22:11 UTC
[05/51] [partial] incubator-geode git commit: WAN and CQ code drop
under the Pivotal SGA
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderEventListenerDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderEventListenerDUnitTest.java b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderEventListenerDUnitTest.java
new file mode 100644
index 0000000..85c45f7
--- /dev/null
+++ b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderEventListenerDUnitTest.java
@@ -0,0 +1,382 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.internal.cache.wan.serial;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.gemstone.gemfire.cache.asyncqueue.AsyncEventListener;
+import com.gemstone.gemfire.cache.wan.GatewaySender;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+import com.gemstone.gemfire.internal.cache.wan.AbstractGatewaySender;
+import com.gemstone.gemfire.internal.cache.wan.MyGatewaySenderEventListener;
+import com.gemstone.gemfire.internal.cache.wan.MyGatewaySenderEventListener2;
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase;
+
+import dunit.DistributedTestCase;
+
+/**
+ * @author skumar
+ *
+ */
+public class SerialGatewaySenderEventListenerDUnitTest extends WANTestBase {
+
+ private static final long serialVersionUID = 1L;
+
+ public SerialGatewaySenderEventListenerDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ super.setUp();
+ }
+
+ /**
+ * Test validates whether the listener attached receives all the events.
+ * this test hangs after the Darrel's checkin 36685. Need to work with Darrel.Commenting it out so that test suit will not hang
+ */
+ public void __testGatewaySenderEventListenerInvocationWithoutLocator() {
+ int mPort = AvailablePortHelper.getRandomAvailablePortForDUnitSite();
+ vm4.invoke(WANTestBase.class, "createCacheWithoutLocator", new Object[] {mPort });
+ vm5.invoke(WANTestBase.class, "createCacheWithoutLocator", new Object[] {mPort });
+ vm6.invoke(WANTestBase.class, "createCacheWithoutLocator", new Object[] {mPort });
+ vm7.invoke(WANTestBase.class, "createCacheWithoutLocator", new Object[] {mPort });
+
+ vm4.invoke(WANTestBase.class, "createSenderWithListener", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, false, true});
+ vm5.invoke(WANTestBase.class, "createSenderWithListener", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, false, true});
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ final Map keyValues = new HashMap();
+ for(int i=0; i< 1000; i++) {
+ keyValues.put(i, i);
+ }
+
+ vm4.invoke(WANTestBase.class, "putGivenKeyValue", new Object[] { testName + "_RR",
+ keyValues });
+
+ vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", keyValues.size() });
+
+ vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", keyValues.size() });
+
+ vm4.invoke(WANTestBase.class, "printEventListenerMap", new Object[]{});
+ vm5.invoke(WANTestBase.class, "printEventListenerMap", new Object[]{});
+
+ vm4.invoke(SerialGatewaySenderEventListenerDUnitTest.class, "validateReceivedEventsMapSizeListener1", new Object[] {"ln", keyValues });
+ }
+
+ /**
+ * Test validates whether the listener attached receives all the events.
+ */
+ public void testGatewaySenderEventListenerInvocation() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] {lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] {lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] {lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] {lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSenderWithListener", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, false, true});
+ vm5.invoke(WANTestBase.class, "createSenderWithListener", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, false, true});
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ final Map keyValues = new HashMap();
+ for(int i=0; i< 1000; i++) {
+ keyValues.put(i, i);
+ }
+
+ vm4.invoke(WANTestBase.class, "putGivenKeyValue", new Object[] { testName + "_RR",
+ keyValues });
+
+ vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", keyValues.size() });
+
+ vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", keyValues.size() });
+
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 0 });
+ vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 0 });
+
+ vm4.invoke(SerialGatewaySenderEventListenerDUnitTest.class, "validateReceivedEventsMapSizeListener1", new Object[] {"ln", keyValues });
+ }
+
+ /**
+ * Test validates whether the listener attached receives all the events.
+ * When there are 2 listeners attcahed to the GatewaySender.
+ */
+ public void testGatewaySender2EventListenerInvocation() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSenderWithListener", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, true, true});
+ vm5.invoke(WANTestBase.class, "createSenderWithListener", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, true, true});
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ final Map keyValues = new HashMap();
+ for(int i=0; i< 1000; i++) {
+ keyValues.put(i, i);
+ }
+
+
+ vm4.invoke(WANTestBase.class, "putGivenKeyValue", new Object[] { testName + "_RR",
+ keyValues });
+
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 0 });
+ vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 0 });
+
+ vm4.invoke(SerialGatewaySenderEventListenerDUnitTest.class, "validateReceivedEventsMapSizeListener2", new Object[] {"ln", keyValues });
+ }
+
+ /**
+ * Test vaildates whether the PoolImpl is created. Ideally when a listener is attached
+ * pool should not be created.
+ */
+ public void testGatewaySenderEventListenerPoolImpl() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class, "createFirstLocatorWithDSId",
+ new Object[] {1});
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] {2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver",
+ new Object[] {nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver",
+ new Object[] {nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] {lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSenderWithListener", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, false, false });
+
+ vm4.invoke(SerialGatewaySenderEventListenerDUnitTest.class, "validateNoPoolCreation", new Object[] {"ln" });
+ }
+
+ // Test start/stop/resume on listener invocation
+ //this test hangs after the Darrel's checkin 36685. Need to work with Darrel.Commenting it out so that test suit will not hang
+ public void __testGatewaySenderEventListener_GatewayOperations() {
+
+ int mPort = AvailablePortHelper.getRandomAvailablePortForDUnitSite();
+ vm4.invoke(WANTestBase.class, "createCacheWithoutLocator", new Object[] {mPort });
+ vm5.invoke(WANTestBase.class, "createCacheWithoutLocator", new Object[] {mPort });
+ vm6.invoke(WANTestBase.class, "createCacheWithoutLocator", new Object[] {mPort });
+ vm7.invoke(WANTestBase.class, "createCacheWithoutLocator", new Object[] {mPort });
+
+ vm4.invoke(WANTestBase.class, "createSenderWithListener", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, false, true});
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ final Map initialKeyValues = new HashMap();
+ for(int i=0; i< 1000; i++) {
+ initialKeyValues.put(i, i);
+ }
+
+ vm4.invoke(WANTestBase.class, "putGivenKeyValue", new Object[] { testName + "_RR",
+ initialKeyValues });
+
+ vm4.invoke(SerialGatewaySenderEventListenerDUnitTest.class, "validateReceivedEventsMapSizeListener1", new Object[] {"ln", initialKeyValues });
+
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ final Map keyValues = new HashMap();
+ for(int i=1000; i< 2000; i++) {
+ keyValues.put(i, i);
+ }
+
+ vm4.invoke(WANTestBase.class, "putGivenKeyValue", new Object[] { testName + "_RR",
+ keyValues });
+
+ vm4.invoke(SerialGatewaySenderEventListenerDUnitTest.class, "validateReceivedEventsMapSizeListener1", new Object[] {"ln", initialKeyValues });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ final Map finalKeyValues = new HashMap();
+ for(int i=2000; i< 3000; i++) {
+ finalKeyValues.put(i, i);
+ }
+
+ vm4.invoke(WANTestBase.class, "putGivenKeyValue", new Object[] { testName + "_RR",
+ finalKeyValues });
+
+ finalKeyValues.putAll(initialKeyValues);
+ vm4.invoke(SerialGatewaySenderEventListenerDUnitTest.class, "validateReceivedEventsMapSizeListener1", new Object[] {"ln", finalKeyValues });
+
+ }
+
+ public static void validateNoPoolCreation(final String siteId) {
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ for(GatewaySender sender: senders) {
+ if (sender.getId().equals(siteId)) {
+ AbstractGatewaySender sImpl = (AbstractGatewaySender)sender;
+ assertNull(sImpl.getProxy());
+ }
+ }
+ }
+
+ public static void validateReceivedEventsMapSizeListener1(final String senderId, final HashMap map) {
+
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ GatewaySender sender = null;
+ for(GatewaySender s : senders){
+ if(s.getId().equals(senderId)){
+ sender = s;
+ break;
+ }
+ }
+
+ final List<AsyncEventListener> listeners = ((AbstractGatewaySender)sender).getAsyncEventListeners();
+ if(listeners.size() == 1) {
+ final AsyncEventListener l = listeners.get(0);
+
+ WaitCriterion wc = new WaitCriterion() {
+ Map listenerMap;
+ public boolean done() {
+ listenerMap = ((MyGatewaySenderEventListener)l)
+ .getEventsMap();
+ boolean sizeCorrect = map.size() == listenerMap.size();
+ boolean keySetCorrect = listenerMap.keySet().containsAll(map.keySet());
+ boolean valuesCorrect = listenerMap.values().containsAll(map.values());
+ return sizeCorrect && keySetCorrect && valuesCorrect;
+ }
+
+ public String description() {
+ return "Waiting for all sites to get updated, the sizes are " + listenerMap.size() + " and " + map.size();
+ }
+ };
+ DistributedTestCase.waitForCriterion(wc, 60000, 500, true);
+ }
+ }
+
+ public static void validateReceivedEventsMapSizeListener2(final String senderId, final HashMap map) {
+
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ GatewaySender sender = null;
+ for(GatewaySender s : senders){
+ if(s.getId().equals(senderId)){
+ sender = s;
+ break;
+ }
+ }
+
+ final List<AsyncEventListener> listeners = ((AbstractGatewaySender)sender).getAsyncEventListeners();
+ if(listeners.size() == 2) {
+ final AsyncEventListener l1 = listeners.get(0);
+ final AsyncEventListener l2 = listeners.get(1);
+
+ WaitCriterion wc = new WaitCriterion() {
+ Map listenerMap1, listenerMap2;
+ public boolean done() {
+ listenerMap1 = ((MyGatewaySenderEventListener)l1)
+ .getEventsMap();
+
+ listenerMap2 = ((MyGatewaySenderEventListener2)l2)
+ .getEventsMap();
+
+ boolean sizeCorrect = map.size() == listenerMap1.size();
+ boolean keySetCorrect = listenerMap1.keySet().containsAll(map.keySet());
+ boolean valuesCorrect = listenerMap1.values().containsAll(map.values());
+
+ boolean sizeCorrect2 = map.size() == listenerMap2.size();
+ boolean keySetCorrect2 = listenerMap2.keySet().containsAll(map.keySet());
+ boolean valuesCorrect2 = listenerMap2.values().containsAll(map.values());
+
+
+ return sizeCorrect && keySetCorrect && valuesCorrect && sizeCorrect2 && keySetCorrect2 && valuesCorrect2;
+ }
+
+ public String description() {
+ return "Waiting for all sites to get updated, the sizes are "
+ + listenerMap1.size() + " and listenerMap2 "
+ + listenerMap2.size() + " expected map size " + map.size();
+ }
+ };
+ DistributedTestCase.waitForCriterion(wc, 60000, 500, true);
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
new file mode 100644
index 0000000..70b6ecb
--- /dev/null
+++ b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderOperationsDUnitTest.java
@@ -0,0 +1,688 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+/**
+ *
+ */
+package com.gemstone.gemfire.internal.cache.wan.serial;
+
+import java.util.Set;
+
+import org.junit.Ignore;
+
+import com.gemstone.gemfire.cache.RegionDestroyedException;
+import com.gemstone.gemfire.cache.wan.GatewaySender;
+import com.gemstone.gemfire.internal.cache.wan.AbstractGatewaySender;
+import com.gemstone.gemfire.internal.cache.wan.GatewaySenderException;
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase;
+
+import dunit.AsyncInvocation;
+import dunit.RMIException;
+
+/**
+ * @author skumar
+ *
+ */
+public class SerialGatewaySenderOperationsDUnitTest extends WANTestBase {
+
+ private static final long serialVersionUID = 1L;
+
+ public SerialGatewaySenderOperationsDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ super.setUp();
+ addExpectedException("Connection reset");
+ addExpectedException("Broken pipe");
+ addExpectedException("Connection refused");
+ addExpectedException("could not get remote locator information");
+ addExpectedException("Unexpected IOException");
+
+ //Stopping the gateway closed the region,
+ //which causes this exception to get logged
+ addExpectedException(RegionDestroyedException.class.getSimpleName());
+ }
+
+ public void testSerialGatewaySenderOperationsWithoutStarting() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 10000 });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifyGatewaySenderOperations", new Object[] { "ln" });
+ vm5.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifyGatewaySenderOperations", new Object[] { "ln" });
+
+ }
+
+
+ public void testStartPauseResumeSerialGatewaySender() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 10000 });
+
+ vm4.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln" });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderPausedState", new Object[] { "ln" });
+ vm5.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderPausedState", new Object[] { "ln" });
+
+ AsyncInvocation inv1 = vm4.invokeAsync(WANTestBase.class, "doPuts",
+ new Object[] { testName + "_RR", 1000 });
+
+ vm4.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln" });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderResumedState", new Object[] { "ln" });
+ vm5.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderResumedState", new Object[] { "ln" });
+
+ try {
+ inv1.join();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ fail("Interrupted the async invocation.");
+ }
+
+ getLogWriter().info("Completed puts in the region");
+
+ vm4.invoke(WANTestBase.class,
+ "validateQueueContents", new Object[] { "ln", 0 });
+ vm5.invoke(WANTestBase.class,
+ "validateQueueContents", new Object[] { "ln", 0 });
+
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+
+ }
+
+ public void testStopSerialGatewaySender() throws Throwable {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 1000 });
+ vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 1000 });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 200 });
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+ vm5.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "validateQueueSizeStat", new Object[] { "ln", 0 });
+ vm5.invoke(WANTestBase.class, "validateQueueSizeStat", new Object[] { "ln", 0 });
+ /**
+ * Should have no effect on GatewaySenderState
+ */
+ vm4.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln" });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+ vm5.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+
+ AsyncInvocation vm4async = vm4.invokeAsync(WANTestBase.class, "startSender", new Object[] { "ln" });
+ AsyncInvocation vm5async = vm5.invokeAsync(WANTestBase.class, "startSender", new Object[] { "ln" });
+ int START_WAIT_TIME = 30000;
+ vm4async.getResult(START_WAIT_TIME);
+ vm5async.getResult(START_WAIT_TIME);
+
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 10000 });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderResumedState", new Object[] { "ln" });
+ vm5.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderResumedState", new Object[] { "ln" });
+
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ }
+
+ public void testStopOneSerialGatewaySenderBothPrimary() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 10000 });
+
+ getLogWriter().info("Completed puts in the region");
+
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ }
+
+ public void testStopOneSerialGatewaySender_PrimarySecondary() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 10000 });
+
+ getLogWriter().info("Completed puts in the region");
+
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ }
+
+ public void testStopOneSender_StartAnotherSender() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm5.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 10000 });
+ getLogWriter().info("Completed puts in the region");
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ }
+
+ public void test_Bug44153_StopOneSender_StartAnotherSender_CheckQueueSize() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+ vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln",
+ 1000 });
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ vm4.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm5.invoke(WANTestBase.class, "doPutsFrom", new Object[] { testName + "_RR", 1000, 11000 });
+
+ vm5.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln",
+ 10000 });
+ vm5.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+ vm5.invoke(SerialGatewaySenderOperationsDUnitTest.class,
+ "verifySenderStoppedState", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln",
+ 1000 });
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ getLogWriter().info("Completed puts in the region");
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 10000 });
+ vm5.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 11000 });
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+ }
+
+ /**
+ * Destroy SerialGatewaySender on all the nodes.
+ */
+ public void testDestroySerialGatewaySenderOnAllNodes() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+
+ //before destroying, stop the sender
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "removeSenderFromTheRegion", new Object[] { "ln", testName + "_RR" });
+ vm5.invoke(WANTestBase.class, "removeSenderFromTheRegion", new Object[] { "ln", testName + "_RR" });
+
+ vm4.invoke(WANTestBase.class, "destroySender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "destroySender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "verifySenderDestroyed", new Object[] { "ln", false });
+ vm5.invoke(WANTestBase.class, "verifySenderDestroyed", new Object[] { "ln", false });
+ }
+
+ /**
+ * Destroy SerialGatewaySender on a single node.
+ */
+ public void testDestroySerialGatewaySenderOnSingleNode() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+
+ //before destroying, stop the sender
+ vm4.invoke(WANTestBase.class, "stopSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "removeSenderFromTheRegion", new Object[] { "ln", testName + "_RR" });
+
+ vm4.invoke(WANTestBase.class, "destroySender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "verifySenderDestroyed", new Object[] { "ln", false });
+ vm5.invoke(WANTestBase.class, "verifySenderRunningState", new Object[] { "ln" });
+ }
+
+ /**
+ * Since the sender is attached to a region and in use, it can not be destroyed.
+ * Hence, exception is thrown by the sender API.
+ */
+ public void testDestroySerialGatewaySenderExceptionScenario() {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+ vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+ false, 100, 10, false, true, null, true });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+
+ try {
+ vm4.invoke(WANTestBase.class, "destroySender", new Object[] { "ln" });
+ } catch (RMIException e) {
+ assertTrue("Cause of the exception should be GatewaySenderException", e.getCause() instanceof GatewaySenderException);
+ }
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 1000 });
+ }
+
+ public static void verifySenderPausedState(String senderId) {
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ GatewaySender sender = null;
+ for (GatewaySender s : senders) {
+ if (s.getId().equals(senderId)) {
+ sender = (AbstractGatewaySender)s;
+ break;
+ }
+ }
+ assertTrue(sender.isPaused());
+ }
+
+ public static void verifySenderResumedState(String senderId) {
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ AbstractGatewaySender sender = null;
+ for (GatewaySender s : senders) {
+ if (s.getId().equals(senderId)) {
+ sender = (AbstractGatewaySender)s;
+ break;
+ }
+ }
+ assertFalse(sender.isPaused());
+ assertTrue(sender.isRunning());
+ }
+
+ public static void verifySenderStoppedState(String senderId) {
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ AbstractGatewaySender sender = null;
+ for (GatewaySender s : senders) {
+ if (s.getId().equals(senderId)) {
+ sender = (AbstractGatewaySender)s;
+ break;
+ }
+ }
+ assertFalse(sender.isRunning());
+ assertFalse(sender.isPaused());
+ }
+
+ public static void verifyGatewaySenderOperations(String senderId) {
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ GatewaySender sender = null;
+ for (GatewaySender s : senders) {
+ if (s.getId().equals(senderId)) {
+ sender = s;
+ break;
+ }
+ }
+ assertFalse(sender.isPaused());
+ assertFalse(((AbstractGatewaySender)sender).isRunning());
+ sender.pause();
+ sender.resume();
+ sender.stop();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueueDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueueDUnitTest.java b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueueDUnitTest.java
new file mode 100644
index 0000000..240d7f2
--- /dev/null
+++ b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/serial/SerialGatewaySenderQueueDUnitTest.java
@@ -0,0 +1,329 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.internal.cache.wan.serial;
+
+import java.io.File;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import com.gemstone.gemfire.cache.AttributesFactory;
+import com.gemstone.gemfire.cache.CacheFactory;
+import com.gemstone.gemfire.cache.DataPolicy;
+import com.gemstone.gemfire.cache.DiskStore;
+import com.gemstone.gemfire.cache.DiskStoreFactory;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.wan.GatewaySender.OrderPolicy;
+import com.gemstone.gemfire.cache.wan.GatewayEventFilter;
+import com.gemstone.gemfire.cache.wan.GatewaySender;
+import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
+import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
+import com.gemstone.gemfire.cache30.MyGatewayEventFilter1;
+import com.gemstone.gemfire.cache30.MyGatewayTransportFilter1;
+import com.gemstone.gemfire.cache30.MyGatewayTransportFilter2;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
+import com.gemstone.gemfire.internal.cache.RegionQueue;
+import com.gemstone.gemfire.internal.cache.wan.AbstractGatewaySender;
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase;
+
+import dunit.VM;
+
+
+public class SerialGatewaySenderQueueDUnitTest extends WANTestBase{
+
+ private static final long serialVersionUID = 1L;
+
+ public SerialGatewaySenderQueueDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ super.setUp();
+ }
+
+
+ public void testPrimarySecondaryQueueDrainInOrder_RR() throws Exception {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class, "createFirstLocatorWithDSId",
+ new Object[] { 1 });
+
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createCache", new Object[] {nyPort });
+ vm3.invoke(WANTestBase.class, "createCache", new Object[] {nyPort });
+
+ vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", null, isOffHeap() });
+
+ vm2.invoke(WANTestBase.class, "createReceiver2", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver2", new Object[] { nyPort });
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSenderWithMultipleDispatchers", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, true, 1, OrderPolicy.KEY });
+ vm5.invoke(WANTestBase.class, "createSenderWithMultipleDispatchers", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, true, 1, OrderPolicy.KEY });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+ testName + "_RR", "ln", isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "addQueueListener", new Object[] { "ln", false});
+ vm5.invoke(WANTestBase.class, "addQueueListener", new Object[] { "ln", false});
+
+ vm2.invoke(WANTestBase.class, "addListenerOnRegion", new Object[] {testName + "_RR"});
+ vm3.invoke(WANTestBase.class, "addListenerOnRegion", new Object[] {testName + "_RR"});
+
+ vm4.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln"});
+
+ vm6.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+ 1000 });
+ pause(5000);
+ HashMap primarySenderUpdates = (HashMap)vm4.invoke(WANTestBase.class, "checkQueue");
+ HashMap secondarySenderUpdates = (HashMap)vm5.invoke(WANTestBase.class, "checkQueue");
+ assertEquals(primarySenderUpdates, secondarySenderUpdates);
+
+ vm4.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln"});
+ pause(2000);
+ vm4.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln"});
+ pause(2000);
+ // We should wait till primarySenderUpdates and secondarySenderUpdates become same
+ // If in 300000ms they don't then throw error.
+ primarySenderUpdates = (HashMap)vm4.invoke(WANTestBase.class, "checkQueue");
+ secondarySenderUpdates = (HashMap)vm5.invoke(WANTestBase.class, "checkQueue");
+
+ vm5.invoke(WANTestBase.class,
+ "checkQueueOnSecondary", new Object[] { primarySenderUpdates });
+// assertEquals(primarySenderUpdates, secondarySenderUpdates);
+
+ vm4.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln"});
+ pause(5000);
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_RR", 1000 });
+ primarySenderUpdates = (HashMap)vm4.invoke(WANTestBase.class, "checkQueue");
+ HashMap receiverUpdates = (HashMap)vm2.invoke(WANTestBase.class, "checkQueue");
+
+ List destroyList = (List)primarySenderUpdates.get("Destroy");
+ List createList = (List)receiverUpdates.get("Create");
+ for(int i = 0; i< 1000; i++){
+ assertEquals(destroyList.get(i), createList.get(i));
+ }
+ assertEquals(primarySenderUpdates.get("Destroy"), receiverUpdates.get("Create"));
+
+ pause(5000);
+ // We expect that after this much time secondary would have got batchremoval message
+ // removing all the keys.
+ secondarySenderUpdates = (HashMap)vm5.invoke(WANTestBase.class, "checkQueue");
+ assertEquals(secondarySenderUpdates.get("Destroy"), receiverUpdates.get("Create"));
+ }
+
+ public void testPrimarySecondaryQueueDrainInOrder_PR() throws Exception {
+ Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+ Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+ vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+ vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+ vm2.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+ testName + "_PR", null, 1, 100, isOffHeap() });
+ vm3.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+ testName + "_PR", null, 1, 100, isOffHeap() });
+
+ vm2.invoke(WANTestBase.class, "addListenerOnRegion", new Object[] {testName + "_PR"});
+ vm3.invoke(WANTestBase.class, "addListenerOnRegion", new Object[] {testName + "_PR"});
+
+ vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+ vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+ vm4.invoke(WANTestBase.class, "createSenderWithMultipleDispatchers", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, true, 1, OrderPolicy.KEY });
+ vm5.invoke(WANTestBase.class, "createSenderWithMultipleDispatchers", new Object[] { "ln", 2,
+ false, 100, 10, false, false, null, true,1, OrderPolicy.KEY });
+
+ vm4.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+ testName + "_PR", "ln", 1, 100, isOffHeap() });
+ vm5.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+ testName + "_PR", "ln", 1, 100, isOffHeap() });
+ vm6.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+ testName + "_PR", "ln", 1, 100, isOffHeap() });
+ vm7.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+ testName + "_PR", "ln", 1, 100, isOffHeap() });
+
+ vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+ vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+ vm4.invoke(WANTestBase.class, "addQueueListener", new Object[] { "ln", false});
+ vm5.invoke(WANTestBase.class, "addQueueListener", new Object[] { "ln", false});
+
+ vm4.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln"});
+
+ vm6.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_PR",
+ 1000 });
+ pause(5000);
+ HashMap primarySenderUpdates = (HashMap)vm4.invoke(WANTestBase.class, "checkQueue");
+ HashMap secondarySenderUpdates = (HashMap)vm5.invoke(WANTestBase.class, "checkQueue");
+ vm5.invoke(WANTestBase.class,
+ "checkQueueOnSecondary", new Object[] { primarySenderUpdates });
+
+ vm4.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln"});
+ pause(4000);
+ vm4.invoke(WANTestBase.class, "pauseSender", new Object[] { "ln"});
+ pause(15000);
+ primarySenderUpdates = (HashMap)vm4.invoke(WANTestBase.class, "checkQueue");
+ secondarySenderUpdates = (HashMap)vm5.invoke(WANTestBase.class, "checkQueue");
+ assertEquals(primarySenderUpdates, secondarySenderUpdates);
+
+ vm4.invoke(WANTestBase.class, "resumeSender", new Object[] { "ln"});
+ pause(5000);
+ vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+ testName + "_PR", 1000 });
+ }
+
+ /**
+ * Test to validate that serial gateway sender queue diskSynchronous attribute
+ * when persistence of sender is enabled.
+ */
+ public void test_ValidateSerialGatewaySenderQueueAttributes_1() {
+ Integer localLocPort = (Integer) vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+
+ Integer remoteLocPort = (Integer) vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, localLocPort });
+
+ WANTestBase test = new WANTestBase(testName);
+ Properties props = new Properties();
+ props.setProperty(DistributionConfig.MCAST_PORT_NAME, "0");
+ props.setProperty(DistributionConfig.LOCATORS_NAME, "localhost["
+ + localLocPort + "]");
+ InternalDistributedSystem ds = test.getSystem(props);
+ cache = CacheFactory.create(ds);
+
+ File directory = new File("TKSender" + "_disk_"
+ + System.currentTimeMillis() + "_" + VM.getCurrentVMNum());
+ directory.mkdir();
+ File[] dirs1 = new File[] { directory };
+ DiskStoreFactory dsf = cache.createDiskStoreFactory();
+ dsf.setDiskDirs(dirs1);
+ DiskStore diskStore = dsf.create("FORNY");
+
+ GatewaySenderFactory fact = cache.createGatewaySenderFactory();
+ fact.setBatchConflationEnabled(true);
+ fact.setBatchSize(200);
+ fact.setBatchTimeInterval(300);
+ fact.setPersistenceEnabled(true);// enable the persistence
+ fact.setDiskSynchronous(true);
+ fact.setDiskStoreName("FORNY");
+ fact.setMaximumQueueMemory(200);
+ fact.setAlertThreshold(1200);
+ GatewayEventFilter myeventfilter1 = new MyGatewayEventFilter1();
+ fact.addGatewayEventFilter(myeventfilter1);
+ GatewayTransportFilter myStreamfilter1 = new MyGatewayTransportFilter1();
+ fact.addGatewayTransportFilter(myStreamfilter1);
+ GatewayTransportFilter myStreamfilter2 = new MyGatewayTransportFilter2();
+ fact.addGatewayTransportFilter(myStreamfilter2);
+ final ExpectedException exTKSender = addExpectedException("Could not connect");
+ try {
+ GatewaySender sender1 = fact.create("TKSender", 2);
+
+ AttributesFactory factory = new AttributesFactory();
+ factory.addGatewaySenderId(sender1.getId());
+ factory.setDataPolicy(DataPolicy.PARTITION);
+ Region region = cache.createRegionFactory(factory.create()).create(
+ "test_ValidateGatewaySenderAttributes");
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ assertEquals(senders.size(), 1);
+ GatewaySender gatewaySender = senders.iterator().next();
+ Set<RegionQueue> regionQueues = ((AbstractGatewaySender) gatewaySender)
+ .getQueues();
+ assertEquals(regionQueues.size(), GatewaySender.DEFAULT_DISPATCHER_THREADS);
+ RegionQueue regionQueue = regionQueues.iterator().next();
+ assertEquals(true, regionQueue.getRegion().getAttributes()
+ .isDiskSynchronous());
+ } finally {
+ exTKSender.remove();
+ }
+
+ }
+
+ /**
+ * Test to validate that serial gateway sender queue diskSynchronous attribute
+ * when persistence of sender is not enabled.
+ */
+ public void test_ValidateSerialGatewaySenderQueueAttributes_2() {
+ Integer localLocPort = (Integer)vm0.invoke(WANTestBase.class,
+ "createFirstLocatorWithDSId", new Object[] { 1 });
+
+ Integer remoteLocPort = (Integer)vm1.invoke(WANTestBase.class,
+ "createFirstRemoteLocator", new Object[] { 2, localLocPort });
+
+ WANTestBase test = new WANTestBase(testName);
+ Properties props = new Properties();
+ props.setProperty(DistributionConfig.MCAST_PORT_NAME, "0");
+ props.setProperty(DistributionConfig.LOCATORS_NAME, "localhost[" + localLocPort + "]");
+ InternalDistributedSystem ds = test.getSystem(props);
+ cache = CacheFactory.create(ds);
+
+ GatewaySenderFactory fact = cache.createGatewaySenderFactory();
+ fact.setBatchConflationEnabled(true);
+ fact.setBatchSize(200);
+ fact.setBatchTimeInterval(300);
+ fact.setPersistenceEnabled(false);//set persistence to false
+ fact.setDiskSynchronous(true);
+ fact.setMaximumQueueMemory(200);
+ fact.setAlertThreshold(1200);
+ GatewayEventFilter myeventfilter1 = new MyGatewayEventFilter1();
+ fact.addGatewayEventFilter(myeventfilter1);
+ GatewayTransportFilter myStreamfilter1 = new MyGatewayTransportFilter1();
+ fact.addGatewayTransportFilter(myStreamfilter1);
+ GatewayTransportFilter myStreamfilter2 = new MyGatewayTransportFilter2();
+ fact.addGatewayTransportFilter(myStreamfilter2);
+ final ExpectedException exp = addExpectedException("Could not connect");
+ try {
+ GatewaySender sender1 = fact.create("TKSender", 2);
+
+ AttributesFactory factory = new AttributesFactory();
+ factory.addGatewaySenderId(sender1.getId());
+ factory.setDataPolicy(DataPolicy.PARTITION);
+ Region region = cache.createRegionFactory(factory.create()).create(
+ "test_ValidateGatewaySenderAttributes");
+ Set<GatewaySender> senders = cache.getGatewaySenders();
+ assertEquals(senders.size(), 1);
+ GatewaySender gatewaySender = senders.iterator().next();
+ Set<RegionQueue> regionQueues = ((AbstractGatewaySender) gatewaySender)
+ .getQueues();
+ assertEquals(regionQueues.size(), GatewaySender.DEFAULT_DISPATCHER_THREADS);
+ RegionQueue regionQueue = regionQueues.iterator().next();
+
+ assertEquals(false, regionQueue.getRegion().getAttributes()
+ .isDiskSynchronous());
+ } finally {
+ exp.remove();
+ }
+ }
+
+}