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:38 UTC
[32/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-cq/src/test/java/com/gemstone/gemfire/internal/cache/ha/HADispatcherDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/ha/HADispatcherDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/ha/HADispatcherDUnitTest.java
new file mode 100755
index 0000000..649297e
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/ha/HADispatcherDUnitTest.java
@@ -0,0 +1,687 @@
+/*=========================================================================
+ * 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.ha;
+
+import hydra.Log;
+
+import java.util.Iterator;
+import java.util.Properties;
+
+import com.gemstone.gemfire.cache.AttributesFactory;
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.CacheException;
+import com.gemstone.gemfire.cache.CacheFactory;
+import com.gemstone.gemfire.cache.CacheListener;
+import com.gemstone.gemfire.cache.DataPolicy;
+import com.gemstone.gemfire.cache.Declarable;
+import com.gemstone.gemfire.cache.EntryEvent;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.RegionAttributes;
+import com.gemstone.gemfire.cache.RegionEvent;
+import com.gemstone.gemfire.cache.Scope;
+import com.gemstone.gemfire.cache.client.internal.PoolImpl;
+import com.gemstone.gemfire.cache.query.CqAttributes;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqListener;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.cache.query.cq.dunit.CqQueryTestListener;
+import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
+import com.gemstone.gemfire.cache30.ClientServerTestCase;
+import com.gemstone.gemfire.cache30.CacheSerializableRunnable;
+import com.gemstone.gemfire.distributed.DistributedSystem;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.internal.AvailablePort;
+import com.gemstone.gemfire.internal.cache.CacheServerImpl;
+import com.gemstone.gemfire.internal.cache.Conflatable;
+import com.gemstone.gemfire.internal.cache.HARegion;
+import com.gemstone.gemfire.internal.cache.LocalRegion;
+import com.gemstone.gemfire.internal.cache.tier.sockets.CacheClientProxy;
+import com.gemstone.gemfire.internal.cache.tier.sockets.CacheServerTestUtil;
+import com.gemstone.gemfire.internal.cache.tier.sockets.ConflationDUnitTest;
+import com.gemstone.gemfire.internal.cache.tier.sockets.HAEventWrapper;
+
+import dunit.DistributedTestCase;
+import dunit.Host;
+import dunit.VM;
+
+/**
+ * This Dunit test is to verify that when the dispatcher of CS dispatches the
+ * Event , the peer's HARegionQueue should get the events removed from the HA
+ * RegionQueues assuming the QRM thread has acted upon by that time
+ * This is done in the following steps
+ * 1. start server1 and server2
+ * 2. start client1 and client2
+ * 3. perform put operation from client1
+ * 4. check the entry in the regionque of client2 on server2.It should be present.
+ * 5. Wait till client2 receives event
+ * 6. Make sure that QRM is envoked
+ * 7. Again the entry in the regionque of client2 on server2.It should not be present.
+ * 8. close client1 and client2
+ * 9. close server1 and server2
+ *
+ * @author Girish Thombare
+ */
+
+public class HADispatcherDUnitTest extends DistributedTestCase
+{
+
+ VM server1 = null;
+
+ VM server2 = null;
+
+ VM client1 = null;
+
+ VM client2 = null;
+
+ public static int PORT1;
+
+ public static int PORT2;
+
+ private static final String REGION_NAME = "HADispatcherDUnitTest_region";
+
+ protected static Cache cache = null;
+
+ public static final Object dummyObj = "dummyObject";
+
+ static volatile boolean isObjectPresent = false;
+
+ final static String KEY1 = "KEY1";
+
+ final static String VALUE1 = "VALUE1";
+
+ final static String KEY2 = "KEY2";
+
+ final static String VALUE2 = "VALUE2";
+
+ static volatile boolean waitFlag = true;
+
+ public HADispatcherDUnitTest(String name) {
+ super(name);
+ }
+
+ @Override
+ public void setUp() throws Exception
+ {
+ super.setUp();
+ final Host host = Host.getHost(0);
+
+ // Server1 VM
+ server1 = host.getVM(0);
+
+ // Server2 VM
+ server2 = host.getVM(1);
+
+ // Client 1 VM
+ client1 = host.getVM(2);
+
+ // client 2 VM
+ client2 = host.getVM(3);
+
+ PORT1 = ((Integer)server1.invoke(HADispatcherDUnitTest.class,
+ "createServerCache", new Object[] { new Boolean(false) })).intValue();
+ server1.invoke(ConflationDUnitTest.class, "setIsSlowStart");
+ server1.invoke(HADispatcherDUnitTest.class, "makeDispatcherSlow");
+ server1.invoke(HADispatcherDUnitTest.class, "setQRMslow");
+ PORT2 = ((Integer)server2.invoke(HADispatcherDUnitTest.class,
+ "createServerCache", new Object[] { new Boolean(true) })).intValue();
+
+ client1.invoke( CacheServerTestUtil.class, "disableShufflingOfEndpoints");
+ client2.invoke( CacheServerTestUtil.class, "disableShufflingOfEndpoints");
+ client1.invoke(HADispatcherDUnitTest.class, "createClientCache",
+ new Object[] {
+ getServerHostName(Host.getHost(0)),
+ new Integer(PORT1), new Integer(PORT2),
+ new Boolean(false) });
+ client2.invoke(HADispatcherDUnitTest.class, "createClientCache",
+ new Object[] {
+ getServerHostName(Host.getHost(0)),
+ new Integer(PORT1), new Integer(PORT2),
+ new Boolean(true) });
+ //createClientCache(new Integer(PORT1), new Integer(PORT2), new Boolean(true) );
+
+ }
+
+ @Override
+ public void tearDown2() throws Exception
+ {
+ super.tearDown2();
+ client1.invoke(HADispatcherDUnitTest.class, "closeCache");
+ client2.invoke(HADispatcherDUnitTest.class, "closeCache");
+ // close server
+ server1.invoke(HADispatcherDUnitTest.class, "resetQRMslow");
+ server1.invoke(HADispatcherDUnitTest.class, "closeCache");
+ server2.invoke(HADispatcherDUnitTest.class, "closeCache");
+
+ }
+
+ public static void closeCache()
+ {
+ if (cache != null && !cache.isClosed()) {
+ cache.close();
+ cache.getDistributedSystem().disconnect();
+ }
+ }
+
+ public static void setQRMslow()
+ {
+ int oldMessageSyncInterval = cache.getMessageSyncInterval();
+ cache.setMessageSyncInterval(6);
+ try {
+ Thread.sleep((oldMessageSyncInterval + 1)*1000);
+ }
+ catch (InterruptedException e) {
+ fail("Unexcepted InterruptedException Occurred");
+ }
+ }
+
+ public static void resetQRMslow()
+ {
+ cache.setMessageSyncInterval(HARegionQueue.DEFAULT_MESSAGE_SYNC_INTERVAL);
+ }
+
+
+ public static void makeDispatcherSlow()
+ {
+ System.setProperty("slowStartTimeForTesting", "5000");
+ }
+
+ private void clientPut(VM vm, final Object key, final Object value) {
+ // performing put from the client1
+ vm.invoke(new CacheSerializableRunnable("putFromClient") {
+ @Override
+ public void run2() throws CacheException
+ {
+ Region region = cache.getRegion(Region.SEPARATOR + REGION_NAME);
+ assertNotNull(region);
+ region.put(key, value);
+ }
+ });
+ }
+
+ private void checkFromClient(VM vm) {
+ // Waiting in the client till it receives the event for the key.
+ vm.invoke(new CacheSerializableRunnable("checkFromClient") {
+ @Override
+ public void run2() throws CacheException
+ {
+ Region region = cache.getRegion(Region.SEPARATOR + REGION_NAME);
+ assertNotNull(region);
+ cache.getLogger().fine("starting the wait");
+ synchronized (dummyObj) {
+ while (waitFlag) {
+ try {
+ dummyObj.wait(30000);
+ }
+ catch (InterruptedException e) {
+ fail("interrupted");
+ }
+ }
+ }
+ cache.getLogger().fine("wait over...waitFlag=" + waitFlag);
+ if (waitFlag)
+ fail("test failed");
+ }
+ });
+ }
+
+ private void checkFromServer(VM vm, final Object key) {
+ // Thread.sleep(10000); // why sleep if the invoke will retry?
+ // performing check in the regionqueue of the server2
+ vm.invoke(new CacheSerializableRunnable("checkFromServer") {
+ @Override
+ public void run2() throws CacheException
+ {
+ Iterator iter = cache.getCacheServers().iterator();
+ CacheServerImpl server = (CacheServerImpl)iter.next();
+ Iterator iter_prox = server.getAcceptor().getCacheClientNotifier()
+ .getClientProxies().iterator();
+ isObjectPresent = false;
+ while (iter_prox.hasNext()) {
+ final CacheClientProxy proxy = (CacheClientProxy)iter_prox.next();
+// ClientProxyMembershipID proxyID = proxy.getProxyID();
+/* Conflict from CQ branch ------------------------------------------------------
+ Region regionqueue = cache.getRegion(Region.SEPARATOR
+ + HARegionQueue.createRegionName(proxyID.toString()));
+ assertNotNull(regionqueue);
+ Iterator itr = regionqueue.values().iterator();
+ while (itr.hasNext()) {
+ Object obj = itr.next();
+ if (obj
+ .getClass()
+ .getName()
+ .equals(
+ "com.gemstone.gemfire.internal.cache.tier.sockets.ClientUpdateMessage")) {
+ Conflatable confObj = (Conflatable)obj;
+ Log.getLogWriter().info("value of the object ");
+ if (key.equals(confObj.getKeyToConflate()))
+ isObjectPresent = true;
+-------------------------------------------------------------------------------*/
+ //HARegion region = (HARegion)cache.getRegion(Region.SEPARATOR
+ // + HAHelper.getRegionQueueName(proxyID.toString()));
+ //assertNotNull(region);
+ HARegion region = (HARegion) proxy.getHARegion();
+ assertNotNull(region);
+ final HARegionQueue regionQueue = region.getOwner();
+
+ WaitCriterion wc = new WaitCriterion() {
+
+ public boolean done() {
+ int sz = regionQueue.size();
+ cache.getLogger().fine("regionQueue.size()::"+ sz);
+ return sz == 0 || !proxy.isConnected();
+ }
+
+ public String description() {
+ return "regionQueue not empty with size " + regionQueue.size()
+ + " for proxy " + proxy;
+ }
+ };
+ DistributedTestCase.waitForCriterion(wc, 60 * 1000, 1000, true);
+ cache.getLogger().fine("processed a proxy");
+ }
+ }
+ });
+ }
+
+ public void testDispatcher() throws Exception
+ {
+ clientPut(client1, KEY1, VALUE1);
+ // Waiting in the client2 till it receives the event for the key.
+ checkFromClient(client2);
+
+ // performing check in the regionqueue of the server2
+ checkFromServer(server2, KEY1);
+
+ // For CQ Only.
+ // performing put from the client1
+ clientPut(client1, KEY2, VALUE2);
+ checkFromClient(client2);
+ checkFromServer(server2, KEY2);
+
+ Log.getLogWriter().info("testDispatcher() completed successfully");
+ }
+
+ /*
+ * This is to test the serialization mechanism of ClientUpdateMessage.
+ * Added after CQ support.
+ * This could be done in different way, by overflowing the HARegion queue.
+ *
+ */
+ public void /*test*/ClientUpdateMessageSerialization() throws Exception
+ {
+ // Update Value.
+ clientPut(client1, KEY1, VALUE1);
+ Log.getLogWriter().fine(">>>>>>>> after clientPut(c1, k1, v1)");
+ // Waiting in the client2 till it receives the event for the key.
+ checkFromClient(client2);
+ Log.getLogWriter().fine("after checkFromClient(c2)");
+
+ // performing check in the regionqueue of the server2
+ checkFromServer(server2, KEY1);
+ Log.getLogWriter().fine("after checkFromServer(s2, k1)");
+
+ // UPDATE.
+ clientPut(client1, KEY1, VALUE1);
+ Log.getLogWriter().fine("after clientPut 2 (c1, k1, v1)");
+ // Waiting in the client2 till it receives the event for the key.
+ checkFromClient(client2);
+ Log.getLogWriter().fine("after checkFromClient 2 (c2)");
+
+ // performing check in the regionqueue of the server2
+ checkFromServer(server2, KEY1);
+ Log.getLogWriter().fine("after checkFromServer 2 (s2, k1)");
+
+ Log.getLogWriter().info("testClientUpdateMessageSerialization() completed successfully");
+ }
+
+ private void createCache(Properties props) throws Exception
+ {
+ DistributedSystem ds = getSystem(props);
+ assertNotNull(ds);
+ ds.disconnect();
+ ds = getSystem(props);
+ cache = CacheFactory.create(ds);
+ assertNotNull(cache);
+ }
+
+ public static Integer createServerCache(Boolean isListenerPresent)
+ throws Exception
+ {
+ new HADispatcherDUnitTest("temp").createCache(new Properties());
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.DISTRIBUTED_ACK);
+ factory.setDataPolicy(DataPolicy.REPLICATE);
+ if (isListenerPresent.booleanValue() == true) {
+ CacheListener serverListener = new HAServerListener();
+ factory.setCacheListener(serverListener);
+ }
+ RegionAttributes attrs = factory.create();
+ cache.createRegion(REGION_NAME, attrs);
+ CacheServerImpl server = (CacheServerImpl)cache.addCacheServer();
+ assertNotNull(server);
+ int port = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET);
+ server.setPort(port);
+ server.setNotifyBySubscription(true);
+ server.start();
+ return new Integer(server.getPort());
+ }
+
+ public static void createClientCache(String hostName, Integer port1, Integer port2,
+ Boolean isListenerPresent) throws Exception
+ {
+ PORT1 = port1.intValue();
+ PORT2 = port2.intValue();
+ Properties props = new Properties();
+ props.setProperty(DistributionConfig.MCAST_PORT_NAME, "0");
+ props.setProperty(DistributionConfig.LOCATORS_NAME, "");
+ new HADispatcherDUnitTest("temp").createCache(props);
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.DISTRIBUTED_ACK);
+ ClientServerTestCase.configureConnectionPool(factory, hostName, new int[] {PORT1,PORT2}, true, -1, 2, null);
+ if (isListenerPresent.booleanValue() == true) {
+ CacheListener clientListener = new HAClientListener();
+ factory.setCacheListener(clientListener);
+ }
+ RegionAttributes attrs = factory.create();
+ cache.createRegion(REGION_NAME, attrs);
+ Region region = cache.getRegion(Region.SEPARATOR + REGION_NAME);
+ assertNotNull(region);
+
+ {
+ LocalRegion lr = (LocalRegion)region;
+ final PoolImpl pool = (PoolImpl)(lr.getServerProxy().getPool());
+ WaitCriterion ev = new WaitCriterion() {
+ public boolean done() {
+ return pool.getPrimary() != null;
+ }
+ public String description() {
+ return null;
+ }
+ };
+ DistributedTestCase.waitForCriterion(ev, 30 * 1000, 200, true);
+ ev = new WaitCriterion() {
+ public boolean done() {
+ return pool.getRedundants().size() >= 1;
+ }
+ public String description() {
+ return null;
+ }
+ };
+ DistributedTestCase.waitForCriterion(ev, 30 * 1000, 200, true);
+
+ assertNotNull(pool.getPrimary());
+ assertTrue("backups="+pool.getRedundants() + " expected=" + 1,
+ pool.getRedundants().size() >= 1);
+ assertEquals(PORT1, pool.getPrimaryPort());
+ }
+
+ region.registerInterest(KEY1);
+
+ // Register CQ.
+ createCQ();
+ }
+
+ private static void createCQ(){
+ QueryService cqService = null;
+ try {
+ cqService = cache.getQueryService();
+ } catch (Exception cqe) {
+ cqe.printStackTrace();
+ fail("Failed to getCQService.");
+ }
+
+ // Create CQ Attributes.
+ CqAttributesFactory cqf = new CqAttributesFactory();
+ CqListener[] cqListeners = {new CqQueryTestListener(getLogWriter())};
+ cqf.initCqListeners(cqListeners);
+ CqAttributes cqa = cqf.create();
+
+ String cqName = "CQForHARegionQueueTest";
+ String queryStr = "Select * from " + Region.SEPARATOR + REGION_NAME;
+
+ // Create CQ.
+ try {
+ CqQuery cq1 = cqService.newCq(cqName, queryStr, cqa);
+ cq1.execute();
+ } catch (Exception ex){
+ getLogWriter().info("CQService is :" + cqService);
+ ex.printStackTrace();
+ AssertionError err = new AssertionError("Failed to create/execute CQ " + cqName + " . ");
+ err.initCause(ex);
+ throw err;
+ }
+ }
+
+
+ /**
+ * This is the client listener which notifies the waiting thread when it
+ * receives the event.
+ */
+ protected static class HAClientListener extends CacheListenerAdapter implements Declarable {
+ public void afterCreate(EntryEvent event)
+ {
+ synchronized (HADispatcherDUnitTest.dummyObj) {
+ try {
+ Object value = event.getNewValue();
+ if (value.equals(HADispatcherDUnitTest.VALUE1)) {
+ HADispatcherDUnitTest.waitFlag = false;
+ HADispatcherDUnitTest.dummyObj.notifyAll();
+ }
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+
+ public void afterUpdate(EntryEvent event)
+ {
+
+ }
+
+ public void afterInvalidate(EntryEvent event)
+ {
+
+ }
+
+ public void afterDestroy(EntryEvent event)
+ {
+
+ }
+
+ public void afterRegionInvalidate(RegionEvent event)
+ {
+
+ }
+
+ public void afterRegionDestroy(RegionEvent event)
+ {
+
+ }
+
+ public void close()
+ {
+
+ }
+
+ public void init(Properties props)
+ {
+
+ }
+ public void afterRegionCreate(RegionEvent event)
+ {
+ // TODO Auto-generated method stub
+
+ }
+ public void afterRegionClear(RegionEvent event)
+ {
+ // TODO Auto-generated method stub
+
+ }
+ public void afterRegionLive(RegionEvent event)
+ {
+ // TODO NOT Auto-generated method stub, added by vrao
+
+ }
+ }
+
+ /**
+ * This is the server listener which ensures that regionqueue is properly populated
+ */
+ protected static class HAServerListener extends CacheListenerAdapter {
+ @Override
+ public void afterCreate(EntryEvent event)
+ {
+ Cache cache = event.getRegion().getCache();
+ Iterator iter = cache.getCacheServers().iterator();
+ CacheServerImpl server = (CacheServerImpl)iter.next();
+ HADispatcherDUnitTest.isObjectPresent = false;
+
+ // The event not be there in the region first time; try couple of time.
+ // This should have been replaced by listener on the HARegion and doing wait for event arrival in that.
+ while (true) {
+ Iterator iter_prox = server.getAcceptor().getCacheClientNotifier()
+ .getClientProxies().iterator();
+ while (iter_prox.hasNext()) {
+ CacheClientProxy proxy = (CacheClientProxy)iter_prox.next();
+// ClientProxyMembershipID proxyID = proxy.getProxyID();
+ HARegion regionForQueue = (HARegion)proxy.getHARegion();
+
+ Iterator itr = regionForQueue.values().iterator();
+ while (itr.hasNext()) {
+ Object obj = itr.next();
+ if (obj instanceof HAEventWrapper) {
+ Conflatable confObj = (Conflatable)obj;
+ if ((HADispatcherDUnitTest.KEY1).equals(confObj.getKeyToConflate()) ||
+ (HADispatcherDUnitTest.KEY2).equals(confObj.getKeyToConflate())) {
+ HADispatcherDUnitTest.isObjectPresent = true;
+ }
+ }
+ }
+ }
+ if (HADispatcherDUnitTest.isObjectPresent == true) {
+ break; // From while.
+ }
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException ex) {fail("interrupted");}
+ }
+ }
+
+ // this test is no longer needed since these
+ // messages are not longer Externalizable
+// /*
+// * This is for testing ClientUpdateMessage's serialization code.
+// */
+// public void afterUpdate(EntryEvent event)
+// {
+// Log.getLogWriter().info("In HAServerListener::AfterUpdate::Event=" + event);
+// Cache cache = event.getRegion().getCache();
+// Iterator iter = cache.getCacheServers().iterator();
+// BridgeServerImpl server = (BridgeServerImpl)iter.next();
+// HADispatcherDUnitTest.isObjectPresent = false;
+
+// // The event not be there in the region first time; try couple of time.
+// // This should have been replaced by listener on the HARegion and doing wait for event arrival in that.
+// while (true) {
+
+// Iterator iter_prox = server.getAcceptor().getCacheClientNotifier()
+// .getClientProxies().iterator();
+// while (iter_prox.hasNext()) {
+// CacheClientProxy proxy = (CacheClientProxy)iter_prox.next();
+// ClientProxyMembershipID proxyID = proxy.getProxyID();
+// HARegion regionForQueue = (HARegion)cache.getRegion(Region.SEPARATOR
+// + HARegionQueue.createRegionName(proxyID.toString()));
+// if (regionForQueue == null) {
+// // I observed dunit throwing an NPE here.
+// // I changed it to just keep retrying which caused dunit to hang.
+// // The queue is gone we are shutting down so just return
+// return;
+// }
+// Iterator itr = regionForQueue.values().iterator();
+// while (itr.hasNext()) {
+// Object obj = itr.next();
+// if (obj.getClass().getName().equals(
+// "com.gemstone.gemfire.internal.cache.tier.sockets.ClientUpdateMessage")) {
+// com.gemstone.gemfire.internal.cache.tier.sockets.ClientUpdateMessage clientUpdateMessage =
+// (com.gemstone.gemfire.internal.cache.tier.sockets.ClientUpdateMessage)obj;
+// try{
+// // Test for readExternal(), writeExternal().
+// ByteArrayOutputStream outStream = new ByteArrayOutputStream();
+// ObjectOutputStream out = new ObjectOutputStream(outStream);
+// clientUpdateMessage.writeExternal(out);
+// out.flush();
+// ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(outStream.toByteArray()));
+
+// com.gemstone.gemfire.internal.cache.tier.sockets.ClientUpdateMessage newClientUpdateMessage =
+// new com.gemstone.gemfire.internal.cache.tier.sockets.ClientUpdateMessageImpl();
+
+// newClientUpdateMessage.readExternal(in);
+
+// Log.getLogWriter().info("Newly constructed ClientUpdateMessage is :" + newClientUpdateMessage.toString());
+// if (!newClientUpdateMessage.hasCqs() ||
+// ((ClientUpdateMessageImpl)newClientUpdateMessage).getClientCqs() == null ||
+// ((ClientUpdateMessageImpl)newClientUpdateMessage).getClientCqs().size() != 2){
+// throw new Exception("CQ Info not present");
+// }
+// HashMap clientCQs = ((ClientUpdateMessageImpl)newClientUpdateMessage).getClientCqs();
+
+// // Try to print CQ details - debug.
+// for (Iterator cciter = clientCQs.keySet().iterator(); cciter.hasNext();) {
+// ClientProxyMembershipID proxyId = (ClientProxyMembershipID)cciter.next();
+
+// HashMap cqs = (HashMap)clientCQs.get(proxyId);
+// Log.getLogWriter().info("Client ID is :" + proxyId);
+
+// for (Iterator cqIter = cqs.keySet().iterator();cqIter.hasNext();){
+// // Add CQ Name.
+// String cq = (String)cqIter.next();
+// // Add CQ Op.
+// Log.getLogWriter().info("CQ Name :" + cq + " CQ OP :" + ((Integer)cqs.get(cq)).intValue());
+// }
+
+// }
+
+// // Test for toData(), fromData().
+// ByteArrayOutputStream dataOutStream = new ByteArrayOutputStream();
+// DataOutputStream dataout = new DataOutputStream(dataOutStream);
+// clientUpdateMessage.toData(dataout);
+// dataOutStream.flush();
+// DataInputStream datain = new DataInputStream(new ByteArrayInputStream(dataOutStream.toByteArray()));
+
+// com.gemstone.gemfire.internal.cache.tier.sockets.ClientUpdateMessage newClientUpdateMessage2 =
+// new com.gemstone.gemfire.internal.cache.tier.sockets.ClientUpdateMessageImpl();
+
+// newClientUpdateMessage2.fromData(datain);
+
+// Log.getLogWriter().info("Newly constructed ClientUpdateMessage is :" + newClientUpdateMessage2.toString());
+// if (!newClientUpdateMessage2.hasCqs() ||
+// ((ClientUpdateMessageImpl)newClientUpdateMessage2).getClientCqs() == null ||
+// ((ClientUpdateMessageImpl)newClientUpdateMessage2).getClientCqs().size() != 2){
+// throw new Exception("CQ Info not present");
+// }
+
+// } catch (Exception ex) {
+// Log.getLogWriter().info("Exception while serializing ClientUpdateMessage.", ex);
+// return;
+// }
+// HADispatcherDUnitTest.isObjectPresent = true;
+// }
+// }
+// }
+// if (HADispatcherDUnitTest.isObjectPresent == true) {
+// break; // From while.
+// }
+// try {
+// Thread.sleep(10);
+// } catch (Exception ex) {}
+// }
+// }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientToServerDeltaDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientToServerDeltaDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientToServerDeltaDUnitTest.java
new file mode 100755
index 0000000..cf434b3
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/ClientToServerDeltaDUnitTest.java
@@ -0,0 +1,1029 @@
+/*=========================================================================
+ * 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.tier.sockets;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Properties;
+
+import com.gemstone.gemfire.DeltaTestImpl;
+import com.gemstone.gemfire.cache.AttributesFactory;
+import com.gemstone.gemfire.cache.AttributesMutator;
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.CacheFactory;
+import com.gemstone.gemfire.cache.DataPolicy;
+import com.gemstone.gemfire.cache.EntryEvent;
+import com.gemstone.gemfire.cache.InterestPolicy;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.RegionAttributes;
+import com.gemstone.gemfire.cache.Scope;
+import com.gemstone.gemfire.cache.SubscriptionAttributes;
+import com.gemstone.gemfire.cache.client.PoolManager;
+import com.gemstone.gemfire.cache.client.internal.ConnectionStats;
+import com.gemstone.gemfire.cache.client.internal.PoolImpl;
+import com.gemstone.gemfire.cache.query.CqAttributes;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqEvent;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.server.CacheServer;
+import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
+import com.gemstone.gemfire.cache.util.CqListenerAdapter;
+import com.gemstone.gemfire.distributed.DistributedSystem;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.i18n.LogWriterI18n;
+import com.gemstone.gemfire.internal.AvailablePort;
+import com.gemstone.gemfire.internal.cache.CacheServerImpl;
+import com.gemstone.gemfire.internal.cache.LocalRegion;
+import com.gemstone.gemfire.internal.cache.PartitionedRegionLocalMaxMemoryDUnitTest.TestObject1;
+
+import dunit.DistributedTestCase;
+import dunit.Host;
+import dunit.VM;
+
+/**
+ * Test client to server flow for delta propogation
+ *
+ * @author aingle
+ * @since 6.1
+ */
+public class ClientToServerDeltaDUnitTest extends DistributedTestCase {
+ /*
+ * Test configuration one server one client also 2 server 2 client
+ */
+ private static Cache cache = null;
+
+ private static LogWriterI18n logger = null;
+
+ VM server = null;
+
+ VM server2 = null;
+
+ VM client = null;
+
+ VM client2 = null;
+
+ private static final String KEY1 = "DELTA_KEY_1";
+
+ private static final String REGION_NAME = "ClientToServerDeltaDunitTest_region";
+
+ private static final int NO_PUT_OPERATION = 3;
+
+ private static PoolImpl pool;
+
+ private static Object[] putDelta = { "Anil",
+ DeltaTestImpl.ERRONEOUS_STRING_FOR_FROM_DELTA, 100,
+ DeltaTestImpl.ERRONEOUS_INT_FOR_TO_DELTA };
+
+ private static int updates = 0;
+
+ private static int cqUpdates = 0;
+
+ private static int create = 0;
+
+ private static Object firstUpdate = null;
+
+ private static Object secondUpdate = null;
+
+ private static Boolean error;
+
+ private static boolean lastKeyReceived = false;
+
+ private static Region region = null;
+
+ private static CSDeltaTestImpl csDelta = null;
+
+ public static String DELTA_KEY = "DELTA_KEY";
+
+ private static final String[] CQs = new String[] {
+ "select * from /" + REGION_NAME,
+ "select * from /" + REGION_NAME + " where intVar = 0",
+ "select * from /" + REGION_NAME + " where intVar > 0",
+ "select * from /" + REGION_NAME + " where intVar < 0"};
+
+ public static String LAST_KEY = "LAST_KEY";
+
+ /** constructor */
+ public ClientToServerDeltaDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ disconnectAllFromDS();
+ final Host host = Host.getHost(0);
+ server = host.getVM(0);
+ server2 = host.getVM(1);
+ client = host.getVM(2);
+ client2 = host.getVM(3);
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ // reset all flags
+ DeltaTestImpl.resetDeltaInvokationCounters();
+ server.invoke(DeltaTestImpl.class, "resetDeltaInvokationCounters");
+ server2.invoke(DeltaTestImpl.class, "resetDeltaInvokationCounters");
+ client.invoke(DeltaTestImpl.class, "resetDeltaInvokationCounters");
+ client2.invoke(DeltaTestImpl.class, "resetDeltaInvokationCounters");
+ // close the clients first
+ client.invoke(ClientToServerDeltaDUnitTest.class, "closeCache");
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "closeCache");
+ // then close the servers
+ server.invoke(ClientToServerDeltaDUnitTest.class, "closeCache");
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "closeCache");
+ }
+
+ public void initialise(Boolean cq) {
+ initialise(Boolean.TRUE/* clone */, null, cq, Boolean.TRUE/* RI */,
+ Boolean.TRUE/* enable delta */);
+ }
+
+ public void initialise(Boolean clone, String[] queries, Boolean cq, Boolean RI, Boolean enableDelta) {
+ Integer PORT1 = ((Integer)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "createServerCache",
+ new Object[] { Boolean.TRUE, Boolean.FALSE, clone, enableDelta })).intValue();
+
+ Integer PORT2 = ((Integer)server2.invoke(
+ ClientToServerDeltaDUnitTest.class, "createServerCache", new Object[] {
+ Boolean.TRUE, Boolean.FALSE, clone, enableDelta })).intValue();
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server.getHost()), new Integer(PORT1),
+ Boolean.FALSE, Boolean.FALSE, Boolean.FALSE });
+
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server2.getHost()),
+ new Integer(PORT2), Boolean.TRUE, Boolean.FALSE, cq, queries, RI });
+ }
+
+ // Same as initialise() except listener flag is false.
+ public void initialise2(Boolean clone, String[] queries, Boolean cq, Boolean RI, Boolean enableDelta) {
+ Integer PORT1 = ((Integer)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "createServerCache",
+ new Object[] { Boolean.FALSE, Boolean.FALSE, clone, enableDelta })).intValue();
+
+ Integer PORT2 = ((Integer)server2.invoke(
+ ClientToServerDeltaDUnitTest.class, "createServerCache", new Object[] {
+ Boolean.FALSE, Boolean.FALSE, clone, enableDelta })).intValue();
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server.getHost()), new Integer(PORT1),
+ Boolean.FALSE, Boolean.FALSE, Boolean.FALSE });
+
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server2.getHost()),
+ new Integer(PORT2), Boolean.TRUE, Boolean.FALSE, cq, queries, RI });
+ }
+
+ /**
+ * This test does the following for single key (<b>failure of fromDelta-
+ * resending of full object</b>):<br>
+ * 1)Verifies that we donot loose attributes updates when delta fails<br>
+ */
+ public void testSendingofFullDeltaObjectsWhenFromDeltaFails() {
+ initialise(false);
+ // set expected value on server
+ server.invoke(ClientToServerDeltaDUnitTest.class, "setFirstSecondUpdate",
+ new Object[] { putDelta[1], (Integer)putDelta[2] });
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putWithFromDeltaERR",
+ new Object[] { KEY1 });
+
+ assertTrue("to Delta Propagation feature NOT used.", ((Boolean)client
+ .invoke(DeltaTestImpl.class, "toDeltaFeatureUsed")).booleanValue());
+ assertTrue("from Delta Propagation feature NOT used.", ((Boolean)server
+ .invoke(DeltaTestImpl.class, "fromDeltaFeatureUsed")).booleanValue());
+
+ assertFalse("Delta Propagation toDeltaFailed", ((Boolean)client.invoke(
+ DeltaTestImpl.class, "isToDeltaFailure")).booleanValue());
+ assertTrue("Delta Propagation fromDelta not Failed", ((Boolean)server
+ .invoke(DeltaTestImpl.class, "isFromDeltaFailure")).booleanValue());
+
+ boolean err = ((Boolean)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "getError")).booleanValue();
+ assertFalse("validation fails", err);
+ }
+
+ /**
+ * This test does the following for single key</br> 1)Verifies that we donot
+ * loose attributes updates when delta fails<br>
+ */
+ public void testPutForDeltaObjects() {
+ initialise(false);
+ // set expected value on server
+ server.invoke(ClientToServerDeltaDUnitTest.class, "setFirstSecondUpdate",
+ new Object[] { putDelta[0], (Integer)putDelta[2] });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "put",
+ new Object[] { KEY1 });
+
+ assertTrue("to Delta Propagation feature NOT used.", ((Boolean)client
+ .invoke(DeltaTestImpl.class, "toDeltaFeatureUsed")).booleanValue());
+ assertTrue("from Delta Propagation feature NOT used.", ((Boolean)server
+ .invoke(DeltaTestImpl.class, "fromDeltaFeatureUsed")).booleanValue());
+ assertFalse("Delta Propagation toDeltaFailed", ((Boolean)client.invoke(
+ DeltaTestImpl.class, "isToDeltaFailure")).booleanValue());
+ assertFalse("Delta Propagation fromDeltaFailed", ((Boolean)server.invoke(
+ DeltaTestImpl.class, "isFromDeltaFailure")).booleanValue());
+ boolean err = ((Boolean)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "getError")).booleanValue();
+ assertFalse("validation fails", err);
+ }
+
+ /**
+ * This test does the following for single key (<b> full cycle</b>):<br>
+ * 1)Verifies that client-server, peer-peer and server-client processing delta<br>
+ */
+ public void testClientToClientDeltaPropagation() throws Exception {
+ initialise(false);
+ // set expected value on s1,s1 and c2
+ server.invoke(ClientToServerDeltaDUnitTest.class, "setFirstSecondUpdate",
+ new Object[] { putDelta[0], (Integer)putDelta[2] });
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "setFirstSecondUpdate",
+ new Object[] { putDelta[0], (Integer)putDelta[2] });
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "setFirstSecondUpdate",
+ new Object[] { putDelta[0], (Integer)putDelta[2] });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putDelta",
+ new Object[] { KEY1 });
+
+ Thread.sleep(5000);
+
+ assertTrue("To Delta Propagation feature NOT used.", ((Boolean)client
+ .invoke(DeltaTestImpl.class, "toDeltaFeatureUsed")).booleanValue());
+ assertTrue("From Delta Propagation feature NOT used.", ((Boolean)server
+ .invoke(DeltaTestImpl.class, "fromDeltaFeatureUsed")).booleanValue());
+ // toDelta() should not be invoked
+ assertFalse("To Delta Propagation feature used.", ((Boolean)server
+ .invoke(DeltaTestImpl.class, "toDeltaFeatureUsed")).booleanValue());
+ assertTrue("From Delta Propagation feature NOT used.", ((Boolean)server2
+ .invoke(DeltaTestImpl.class, "fromDeltaFeatureUsed")).booleanValue());
+ // toDelta() should not be invoked
+ assertFalse("To Delta Propagation feature used.", ((Boolean)server2
+ .invoke(DeltaTestImpl.class, "toDeltaFeatureUsed")).booleanValue());
+ assertTrue("from Delta Propagation feature NOT used.", ((Boolean)client2
+ .invoke(DeltaTestImpl.class, "fromDeltaFeatureUsed")).booleanValue());
+
+ boolean err = ((Boolean)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "getError")).booleanValue();
+ err = ((Boolean)server2.invoke(ClientToServerDeltaDUnitTest.class,
+ "getError")).booleanValue();
+ err = ((Boolean)client2.invoke(ClientToServerDeltaDUnitTest.class,
+ "getError")).booleanValue();
+ assertFalse("validation fails", err);
+ }
+
+ private static void putDeltaForCQ(String key, Integer numOfPuts, Integer[] cqIndices, Boolean[] satisfyQuery) {
+ Region region = cache.getRegion(REGION_NAME);
+ DeltaTestImpl val = null;
+ for (int j = 0; j < numOfPuts; j++) {
+ val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
+ new TestObject1("0", 0));
+ for(int i=0; i<cqIndices.length; i++) {
+ switch(i) {
+ case 0:
+ val.setStr("CASE_0");
+ // select *
+ break;
+ case 1:
+ val.setStr("CASE_1");
+ // select where intVar = 0
+ if (satisfyQuery[i]) {
+ val.setIntVar(0);
+ }
+ else {
+ val.setIntVar(100);
+ }
+ break;
+ case 2:
+ val.setStr("CASE_2");
+ // select where intVar > 0
+ if (satisfyQuery[i]) {
+ val.setIntVar(100);
+ }
+ else {
+ val.setIntVar(-100);
+ }
+ break;
+ case 3:
+ val.setStr("CASE_3");
+ // select where intVar < 0
+ if (satisfyQuery[i]) {
+ val.setIntVar(-100);
+ }
+ else {
+ val.setIntVar(100);
+ }
+ break;
+ default:
+ break;
+ }
+ }
+ region.put(key, val);
+ }
+ }
+
+ /*
+ * put delta ; not previous deltas
+ *
+ */
+ private static void putDelta(String key) {
+ Region r = cache.getRegion(REGION_NAME);
+ DeltaTestImpl val = null;
+ for (int i = 0; i < NO_PUT_OPERATION; i++) {
+ val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
+ new TestObject1("0", 0));
+ switch (i) {
+ case 1:
+ val.setStr((String)putDelta[0]);
+ break;
+ case 2:
+ val.setIntVar(((Integer)putDelta[2]).intValue());
+ break;
+ }
+ r.put(key, val);
+ }
+ }
+
+ private static void putLastKey() {
+ Region r = cache.getRegion(REGION_NAME);
+ r.put(LAST_KEY, LAST_KEY);
+ }
+
+ private static void putLastKeyWithDelta() {
+ Region r = cache.getRegion(REGION_NAME);
+ r.put(LAST_KEY, new DeltaTestImpl());
+ }
+
+ private static void setFirstSecondUpdate(Object first, Object second) {
+ firstUpdate = first;
+ secondUpdate = second;
+ }
+
+ private static Boolean getError() {
+ return error;
+ }
+
+ /*
+ * put delta full cycle
+ *
+ */
+ private static void put(String key) {
+ Region r = cache.getRegion(REGION_NAME);
+ DeltaTestImpl val = null;
+ for (int i = 0; i < NO_PUT_OPERATION; i++) {
+ switch (i) {
+ case 0:
+ val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
+ new TestObject1("0", 0));
+ break;
+ case 1:
+ val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
+ new TestObject1("0", 0));
+ val.setStr((String)putDelta[0]);
+ break;
+ case 2:
+ val = new DeltaTestImpl(0, (String)putDelta[1], new Double(0),
+ new byte[0], new TestObject1("0", 0));
+ val.setIntVar(((Integer)putDelta[2]).intValue());
+ break;
+ }
+ r.put(key, val);
+ }
+ }
+
+ /*
+ * put delta with some times fromDelta fails to apply client sends back full
+ * object
+ */
+ private static void putWithFromDeltaERR(String key) {
+ Region r = cache.getRegion(REGION_NAME);
+ DeltaTestImpl val = null;
+ for (int i = 0; i < NO_PUT_OPERATION; i++) {
+ switch (i) {
+ case 0:
+ val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
+ new TestObject1("0", 0));
+ break;
+ case 1:
+ val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
+ new TestObject1("0", 0));
+ val.setStr((String)putDelta[1]);
+ break;
+ case 2:
+ val = new DeltaTestImpl(0, (String)putDelta[1], new Double(0),
+ new byte[0], new TestObject1("0", 0));
+ val.setIntVar(((Integer)putDelta[2]).intValue());
+ break;
+ }
+ r.put(key, val);
+ }
+ }
+
+ /*
+ * put delta with some times toDelta fails to apply; client sends back full
+ * object
+ */
+ private static void putWithTODeltaERR(String key) {
+ Region r = cache.getRegion(REGION_NAME);
+ DeltaTestImpl val = null;
+ for (int i = 0; i < NO_PUT_OPERATION; i++) {
+ switch (i) {
+ case 0:
+ val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
+ new TestObject1("0", 0));
+ break;
+ case 1:
+ val = new DeltaTestImpl(0, "0", new Double(0), new byte[0],
+ new TestObject1("0", 0));
+ val.setIntVar(((Integer)putDelta[3]).intValue());
+ break;
+ case 2:
+ val = new DeltaTestImpl(((Integer)putDelta[3]).intValue(), "0",
+ new Double(0), new byte[0], new TestObject1("0", 0));
+ val.setStr((String)putDelta[0]);
+ break;
+ }
+ r.put(key, val);
+ }
+ }
+
+ public static Integer createServerCache(Boolean attachListener,
+ Boolean isEmpty) throws Exception {
+ return createServerCache(attachListener, isEmpty, Boolean.TRUE/* clone */,
+ Boolean.TRUE/* enable delta */);
+ }
+
+ /*
+ * create server cache
+ */
+ public static Integer createServerCache(Boolean attachListener,
+ Boolean isEmpty, Boolean clone, Boolean enableDelta) throws Exception {
+ // for validation
+ updates = 0;
+ create = 0;
+ firstUpdate = null;
+ secondUpdate = null;
+ error = false;
+ Properties props = new Properties();
+ props.setProperty(DistributionConfig.DELTA_PROPAGATION_PROP_NAME,
+ enableDelta.toString());
+ new ClientToServerDeltaDUnitTest("temp").createCache(props);
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.DISTRIBUTED_ACK);
+ factory.setConcurrencyChecksEnabled(true);
+ if (isEmpty) {
+ factory.setSubscriptionAttributes(new SubscriptionAttributes(
+ InterestPolicy.ALL));
+ factory.setDataPolicy(DataPolicy.EMPTY);
+
+ }
+ else {
+ factory.setDataPolicy(DataPolicy.REPLICATE);
+ }
+ factory.setCloningEnabled(clone);
+ RegionAttributes attrs = factory.create();
+ region = cache.createRegion(REGION_NAME, attrs);
+
+ AttributesMutator am = region.getAttributesMutator();
+ if (attachListener) {
+ am.addCacheListener(new CacheListenerAdapter() {
+ public void afterCreate(EntryEvent event) {
+ create++;
+ }
+
+ public void afterUpdate(EntryEvent event) {
+ switch (updates) {
+ case 0:
+ // first delta
+ validateUpdates(event, firstUpdate, "FIRST");
+ updates++;
+ break;
+ case 1:
+ // combine delta
+ validateUpdates(event, firstUpdate, "FIRST");
+ validateUpdates(event, secondUpdate, "SECOND");
+ updates++;
+ break;
+ default:
+ break;
+ }
+ }
+ });
+ } else if (!isEmpty){
+ am.addCacheListener(new CacheListenerAdapter() {
+ public void afterCreate(EntryEvent event) {
+ switch (create) {
+ case 1:
+ validateUpdates(event, firstUpdate, "FIRST");
+ create++;
+ break;
+ case 2:
+ validateUpdates(event, secondUpdate, "SECOND");
+ create++;
+ break;
+ default:
+ create++;
+ break;
+ }
+ }
+ /* public void afterUpdate(EntryEvent event) {
+ updates++;
+ }*/
+ });
+
+ }
+
+ CacheServer server = cache.addCacheServer();
+ int port = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET);
+ server.setPort(port);
+ // ensures updates to be sent instead of invalidations
+ server.setNotifyBySubscription(true);
+ server.start();
+ return new Integer(server.getPort());
+ }
+
+ public static void createClientCache(String host, Integer port,
+ Boolean attachListener, Boolean isEmpty, Boolean isCq) throws Exception {
+ createClientCache(host, port, attachListener, isEmpty, isCq, new String[0], true);
+ }
+
+ public static void createClientCache(String host, Integer port,
+ Boolean attachListener, Boolean isEmpty, Boolean isCq, String[] cqQueryString) throws Exception {
+ createClientCache(host, port, attachListener, isEmpty, isCq, cqQueryString, true);
+ }
+
+ public static void createClientCache(String host, Integer port,
+ Boolean attachListener, Boolean isEmpty, Boolean isCq,
+ String[] cqQueryString, Boolean registerInterestAll) throws Exception {
+ createClientCache(host, port, attachListener, isEmpty, isCq, cqQueryString,
+ registerInterestAll, true);
+ }
+
+ /*
+ * create client cache
+ */
+ public static void createClientCache(String host, Integer port,
+ Boolean attachListener, Boolean isEmpty, Boolean isCq,
+ String[] cqQueryString, Boolean registerInterestAll,
+ Boolean enableSubscription) throws Exception {
+ updates = 0;
+ create = 0;
+ firstUpdate = null;
+ secondUpdate = null;
+ error = false;
+ lastKeyReceived = false;
+ Properties props = new Properties();
+ props.setProperty(DistributionConfig.MCAST_PORT_NAME, "0");
+ props.setProperty(DistributionConfig.LOCATORS_NAME, "");
+ new ClientToServerDeltaDUnitTest("temp").createCache(props);
+ pool = (PoolImpl)PoolManager.createFactory().addServer(host, port.intValue())
+ .setThreadLocalConnections(true).setMinConnections(2)
+ .setSubscriptionEnabled(enableSubscription).setSubscriptionRedundancy(0)
+ .setReadTimeout(10000).setPingInterval(1000).setSocketBufferSize(32768).create(
+ "ClientToServerDeltaDunitTestPool");
+ AttributesFactory factory = new AttributesFactory();
+ factory.setScope(Scope.DISTRIBUTED_ACK);
+ factory.setConcurrencyChecksEnabled(true);
+ if (isEmpty) {
+ factory.setSubscriptionAttributes(new SubscriptionAttributes(
+ InterestPolicy.ALL));
+ factory.setDataPolicy(DataPolicy.EMPTY);
+ }
+ else {
+ factory.setDataPolicy(DataPolicy.NORMAL);
+ }
+ factory.setPoolName(pool.getName());
+ factory.setCloningEnabled(false);
+
+ // region with empty data policy
+ RegionAttributes attrs = factory.create();
+ region = cache.createRegion(REGION_NAME, attrs);
+ if (attachListener) {
+ region.getAttributesMutator().addCacheListener(new CacheListenerAdapter() {
+ public void afterCreate(EntryEvent event) {
+ create++;
+ if (LAST_KEY.equals(event.getKey())) {
+ lastKeyReceived = true;
+ };
+ }
+
+ public void afterUpdate(EntryEvent event) {
+ switch (updates) {
+ case 0:
+ // first delta
+ validateUpdates(event, firstUpdate, "FIRST");
+ updates++;
+ break;
+ case 1:
+ // combine delta
+ validateUpdates(event, firstUpdate, "FIRST");
+ validateUpdates(event, secondUpdate, "SECOND");
+ updates++;
+ break;
+ default:
+ break;
+ }
+ }
+ });
+ }
+ if (registerInterestAll) {
+ region.registerInterest("ALL_KEYS");
+ }
+ if (isCq) {
+ CqAttributesFactory cqf = new CqAttributesFactory();
+ CqListenerAdapter cqlist = new CqListenerAdapter() {
+ public void onEvent(CqEvent cqEvent) {
+ Object key = cqEvent.getKey();
+ if (LAST_KEY.equals(key)) {
+ lastKeyReceived = true;
+ }
+ logger.fine("CQ event received for (key, value): (" + key + ", "
+ + cqEvent.getNewValue() + ")");
+ }
+
+ public void onError(CqEvent cqEvent) {
+ logger.fine("CQ error received for key: " + cqEvent.getKey());
+ }
+ };
+ cqf.addCqListener(cqlist);
+ CqAttributes cqa = cqf.create();
+ for (int i = 0; i < cqQueryString.length; i++) {
+ CqQuery cq = cache.getQueryService().newCq("Delta_Query_" + i,
+ cqQueryString[i], cqa);
+ cq.execute();
+ }
+ }
+ }
+
+ /*
+ * create cache with properties
+ */
+ private void createCache(Properties props) throws Exception {
+ DistributedSystem ds = getSystem(props);
+ cache = CacheFactory.create(ds);
+ assertNotNull(cache);
+ logger = cache.getLoggerI18n();
+ }
+
+ // to validate updates in listener
+ private static void validateUpdates(EntryEvent event, Object obj, String str) {
+ if (obj instanceof String)
+ assertTrue(str + " update missed ", ((DeltaTestImpl)event.getNewValue())
+ .getStr().equals((String)obj));
+ else if (obj instanceof Integer)
+ assertTrue(str + " update missed ", ((DeltaTestImpl)event.getNewValue())
+ .getIntVar() == (Integer)obj);
+ else
+ error = true;
+ }
+
+ /*
+ * close cache
+ */
+ public static void closeCache() {
+ if (cache != null && !cache.isClosed()) {
+ cache.close();
+ cache.getDistributedSystem().disconnect();
+ lastKeyReceived = false;
+ }
+ }
+ /**
+ * This test does the following for single key:<br>
+ * 1)Verifies that cacheless client calls toDelta <br>
+ */
+ public void testEmptyClientAsFeederToServer() {
+ Integer PORT1 = ((Integer)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "createServerCache", new Object[] { Boolean.FALSE,
+ Boolean.FALSE })).intValue();
+
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "createServerCache",
+ new Object[] { Boolean.FALSE, Boolean.FALSE });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server.getHost()), new Integer(PORT1),
+ Boolean.FALSE, Boolean.TRUE, Boolean.FALSE });
+
+/* server2.invoke(ClientToServerDeltaDUnitTest.class, "setFirstSecondUpdate",
+ new Object[] { putDelta[0], (Integer)putDelta[2] });*/
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putDelta",
+ new Object[] { KEY1 });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "checkTodeltaCounter",
+ new Object[] { new Integer(2)});
+ server.invoke(ClientToServerDeltaDUnitTest.class, "checkFromdeltaCounter");
+ server.invoke(ClientToServerDeltaDUnitTest.class, "checkTodeltaCounter",
+ new Object[] { new Integer(0)});
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "checkFromdeltaCounter");
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "checkTodeltaCounter",
+ new Object[] { new Integer(0)});
+ }
+
+ /**
+ * This test does the following for single key:<br>
+ * 1)Verifies that from delta should not called on server with empty data
+ * policy just by passed delta to data store<br>
+ */
+ public void testEmptyServerAsFeederToPeer() {
+ Integer PORT1 = ((Integer)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "createServerCache", new Object[] { Boolean.FALSE,
+ Boolean.TRUE })).intValue();
+
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "createServerCache",
+ new Object[] { Boolean.FALSE, Boolean.FALSE });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server.getHost()), new Integer(PORT1),
+ Boolean.FALSE, Boolean.TRUE, Boolean.FALSE });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putDelta",
+ new Object[] { KEY1 });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "checkTodeltaCounter",
+ new Object[] { new Integer(2)});
+
+ server.invoke(ClientToServerDeltaDUnitTest.class, "checkDeltaFeatureNotUsed");
+
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "checkTodeltaCounter",
+ new Object[] { new Integer(0)});
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "checkFromdeltaCounter");
+
+ }
+
+ /**
+ * This test does verifies that server with empty data policy sends deltas to
+ * the client which can handle deltas. Server sends full values which can not
+ * handle deltas.
+ *
+ */
+ public void testClientsConnectedToEmptyServer() {
+ Integer PORT1 = ((Integer)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "createServerCache", new Object[] { Boolean.FALSE,
+ Boolean.TRUE })).intValue();
+
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "createServerCache",
+ new Object[] { Boolean.FALSE, Boolean.FALSE });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server.getHost()), new Integer(PORT1),
+ Boolean.TRUE, Boolean.TRUE, Boolean.FALSE });
+
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server.getHost()), new Integer(PORT1),
+ Boolean.TRUE, Boolean.FALSE, Boolean.FALSE });
+
+ int deltaSent = (Integer)server2.invoke(
+ ClientToServerDeltaDUnitTest.class,"putsWhichReturnsDeltaSent");
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "waitForLastKey");
+ client.invoke(ClientToServerDeltaDUnitTest.class, "checkForDelta");
+
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "waitForLastKey");
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "checkDeltaInvoked", new Object[]{new Integer(deltaSent)});
+
+ }
+
+ /**
+ * This test does the following for single key:<br>
+ * 1)Verifies that To delta called on client should be equal to fromDeltaCounter on datastore <br>
+ */
+ public void testClientNonEmptyEmptyServerAsFeederToPeer() {
+ Integer PORT1 = ((Integer)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "createServerCache", new Object[] { Boolean.FALSE,
+ Boolean.TRUE })).intValue();
+
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "createServerCache",
+ new Object[] { Boolean.FALSE, Boolean.FALSE });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "createClientCache",
+ new Object[] { getServerHostName(server.getHost()), new Integer(PORT1),
+ Boolean.FALSE, Boolean.FALSE, Boolean.FALSE});
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putDelta",
+ new Object[] { KEY1 });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "checkTodeltaCounter",
+ new Object[] { new Integer(2)});
+
+ server.invoke(ClientToServerDeltaDUnitTest.class, "checkDeltaFeatureNotUsed");
+
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "checkTodeltaCounter",
+ new Object[] { new Integer(0)});
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "checkFromdeltaCounter");
+
+ }
+
+ /**
+ * This test has client1 connected to server1 and client2 connected to
+ * server2. client2 has a CQ registered on the server2. client1 creates a key
+ * on server1 which satisfies the CQ and it then updates the same key with a
+ * value so that it sould not satisfy the CQ. The cloning is set to false. The
+ * test ensures that the client2 gets the update event. This implies that
+ * server2 clones the old value prior to applying delta received from server1
+ * and then processes the CQ.
+ *
+ * @throws Exception
+ */
+ public void testC2CDeltaPropagationWithCQ() throws Exception {
+ initialise(false/* clone */, new String[] { CQs[1] }, true/* CQ */,
+ true/* RI */, true/* enable delta */);
+ client.invoke(ClientToServerDeltaDUnitTest.class, "createKeys",
+ new Object[] { new String[] { KEY1 } });
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putDeltaForCQ",
+ new Object[] { KEY1, new Integer(1), new Integer[] { 1 },
+ new Boolean[] { false, true, false, false } });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putLastKeyWithDelta");
+
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "waitForLastKey");
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "verifyDeltaReceived");
+ }
+
+ /**
+ * This test ensures that a server sends delta bytes to a client even if that
+ * client is not interested in that event but is getting the event only
+ * because the event satisfies a CQ which the client has registered with the
+ * server.
+ *
+ * @throws Exception
+ */
+ public void testC2CDeltaPropagationWithCQWithoutRI() throws Exception {
+ initialise(false/* clone */, new String[] { CQs[1] }, true/* CQ */,
+ false/* RI */, true/* enable delta */);
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "createKeys",
+ new Object[] { new String[] { KEY1 } });
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putDeltaForCQ",
+ new Object[] { KEY1, new Integer(1), new Integer[] { 1 },
+ new Boolean[] { false, true, false, false } });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putLastKeyWithDelta");
+
+ client2.invoke(ClientToServerDeltaDUnitTest.class, "waitForLastKey");
+ server2.invoke(ClientToServerDeltaDUnitTest.class, "verifyDeltaSent",
+ new Object[] { Integer.valueOf(1) });
+ }
+
+ public void testClientSendsFullValueToServerWhenDeltaOffAtServer() {
+ initialise2(false/* clone */, new String[] { CQs[1] }, false/* CQ */,
+ true/* RI */, false/* enable delta */);
+ // set expected value on server
+ server.invoke(ClientToServerDeltaDUnitTest.class, "setFirstSecondUpdate",
+ new Object[] { putDelta[0], (Integer)putDelta[2] });
+
+ client.invoke(ClientToServerDeltaDUnitTest.class, "putDelta",
+ new Object[] { DELTA_KEY });
+
+ assertFalse("Delta Propagation feature used at client.", (Boolean)client
+ .invoke(DeltaTestImpl.class, "toDeltaFeatureUsed"));
+ assertFalse("Delta Propagation feature used at server.", (Boolean)server
+ .invoke(DeltaTestImpl.class, "fromDeltaFeatureUsed"));
+ assertFalse(
+ "Failures at client while calculating delta. But delta-propagation is false at server.",
+ ((Boolean)client.invoke(DeltaTestImpl.class, "isToDeltaFailure"))
+ .booleanValue());
+ assertFalse(
+ "Failures at server while applying delta. But delta-propagation is false at server.",
+ (Boolean)server.invoke(DeltaTestImpl.class, "isFromDeltaFailure"));
+ }
+
+ public void testC2SDeltaPropagationWithOldValueInvalidatedAtServer()
+ throws Exception {
+ String key = "DELTA_KEY";
+ Integer port1 = (Integer)server.invoke(ClientToServerDeltaDUnitTest.class,
+ "createServerCache", new Object[] {false, false, false, true});
+ createClientCache("localhost", port1, false, false, false, null, false,
+ false);
+
+ LocalRegion region = (LocalRegion)cache.getRegion(REGION_NAME);
+ region.put(key, new DeltaTestImpl());
+
+ server.invoke(ClientToServerDeltaDUnitTest.class, "doInvalidate",
+ new Object[] {key});
+ DeltaTestImpl value = new DeltaTestImpl();
+ value.setStr("UPDATED_VALUE");
+ region.put(key, value);
+
+ assertTrue(region.getCachePerfStats().getDeltasSent() == 1);
+ assertTrue(region.getCachePerfStats().getDeltaFullValuesSent() == 1);
+ }
+
+ public static void doInvalidate(String key) {
+ Region region = cache.getRegion(REGION_NAME);
+ region.invalidate(key);
+ }
+
+ public static void verifyDeltaReceived() {
+ assertTrue("Expected 1 fromDelta() invokation but was "
+ + DeltaTestImpl.getFromDeltaInvokations(), DeltaTestImpl
+ .getFromDeltaInvokations() == 1);
+ }
+
+ public static void verifyDeltaSent(Integer deltas) {
+ CacheClientNotifier ccn = ((CacheServerImpl)cache
+ .getCacheServers().toArray()[0]).getAcceptor()
+ .getCacheClientNotifier();
+
+ int numOfDeltasSent = ((CacheClientProxy)ccn.getClientProxies().toArray()[0])
+ .getStatistics().getDeltaMessagesSent();
+ assertTrue("Expected " + deltas + " deltas to be sent but "
+ + numOfDeltasSent + " were sent.", numOfDeltasSent == deltas);
+ }
+
+ public static void checkFromdeltaCounter() {
+ assertTrue("FromDelta counters do not match, expected: "
+ + (NO_PUT_OPERATION - 1) + ", actual: "
+ + DeltaTestImpl.getFromDeltaInvokations(), DeltaTestImpl
+ .getFromDeltaInvokations() >= (NO_PUT_OPERATION - 1));
+ }
+
+ public static void checkTodeltaCounter(Integer count) {
+ assertTrue("ToDelta counters do not match, expected: " + count.intValue()
+ + ", actual: " + DeltaTestImpl.getToDeltaInvokations(), DeltaTestImpl
+ .getToDeltaInvokations() >= count.intValue());
+ }
+
+ public static void checkDeltaFeatureNotUsed() {
+ assertTrue("Delta Propagation feature used.", !(DeltaTestImpl
+ .deltaFeatureUsed()));
+ }
+
+ public static void createKeys(String[] keys) {
+ Region region = cache.getRegion(REGION_NAME);
+ for (int i = 0; i < keys.length; i++) {
+ region.create(keys[i], new DeltaTestImpl());
+ }
+ }
+
+ public static int putsWhichReturnsDeltaSent() throws Exception {
+ csDelta = new CSDeltaTestImpl();
+ region.put(DELTA_KEY, csDelta);
+ for (int i = 0; i < NO_PUT_OPERATION; i++) {
+ csDelta.setIntVar(i);
+ region.put(DELTA_KEY, csDelta);
+ }
+ region.put(LAST_KEY, "");
+
+ return csDelta.deltaSent;
+ }
+
+ public static void checkDeltaInvoked(Integer deltaSent) {
+ assertTrue("Delta applied :"
+ + ((CSDeltaTestImpl)region.get(DELTA_KEY)).getDeltaApplied()
+ + "\n Delta sent :" + deltaSent, ((CSDeltaTestImpl)region
+ .get(DELTA_KEY)).getDeltaApplied() == deltaSent);
+ }
+
+ public static void checkForDelta() {
+ assertTrue("Delta sent to EMPTY data policy region", DeltaTestImpl.getFromDeltaInvokations()==0);
+ }
+
+ public static void waitForLastKey() {
+ WaitCriterion wc = new WaitCriterion() {
+ public boolean done() {
+ return ClientToServerDeltaDUnitTest.lastKeyReceived;
+ }
+ public String description() {
+ return "Last key NOT received.";
+ }
+ };
+ DistributedTestCase.waitForCriterion(wc, 10*1000, 100, true);
+ }
+
+ static class CSDeltaTestImpl extends DeltaTestImpl {
+ int deltaSent = 0;
+ int deltaApplied = 0;
+
+ public CSDeltaTestImpl() {
+ }
+
+ public void toDelta(DataOutput out) throws IOException {
+ super.toDelta(out);
+ deltaSent++;
+ }
+
+ public void fromDelta(DataInput in) throws IOException {
+ super.fromDelta(in);
+ deltaApplied++;
+ }
+
+ public int getDeltaSent() {
+ return deltaSent;
+ }
+
+ public int getDeltaApplied() {
+ return deltaApplied;
+ }
+
+ public String toString() {
+ return "CSDeltaTestImpl[deltaApplied=" + deltaApplied + "]"
+ + super.toString();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DeltaPropagationWithCQDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DeltaPropagationWithCQDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DeltaPropagationWithCQDUnitTest.java
new file mode 100644
index 0000000..dc2f1f4
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/internal/cache/tier/sockets/DeltaPropagationWithCQDUnitTest.java
@@ -0,0 +1,336 @@
+/*=========================================================================
+ * 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.tier.sockets;
+
+import java.util.Properties;
+
+import com.gemstone.gemfire.DeltaTestImpl;
+import com.gemstone.gemfire.cache.AttributesFactory;
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.CacheFactory;
+import com.gemstone.gemfire.cache.DataPolicy;
+import com.gemstone.gemfire.cache.EntryEvent;
+import com.gemstone.gemfire.cache.GemFireCache;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.RegionAttributes;
+import com.gemstone.gemfire.cache.RegionFactory;
+import com.gemstone.gemfire.cache.RegionShortcut;
+import com.gemstone.gemfire.cache.Scope;
+import com.gemstone.gemfire.cache.client.ClientCache;
+import com.gemstone.gemfire.cache.client.ClientCacheFactory;
+import com.gemstone.gemfire.cache.client.ClientRegionFactory;
+import com.gemstone.gemfire.cache.client.ClientRegionShortcut;
+import com.gemstone.gemfire.cache.client.Pool;
+import com.gemstone.gemfire.cache.client.PoolFactory;
+import com.gemstone.gemfire.cache.client.PoolManager;
+import com.gemstone.gemfire.cache.query.CqAttributes;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqEvent;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.cache.server.CacheServer;
+import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
+import com.gemstone.gemfire.cache.util.CqListenerAdapter;
+import com.gemstone.gemfire.cache30.ClientServerTestCase;
+import com.gemstone.gemfire.distributed.DistributedSystem;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.internal.AvailablePort;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+import com.gemstone.gemfire.internal.cache.CacheServerImpl;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+
+import dunit.DistributedTestCase;
+import dunit.Host;
+import dunit.VM;
+
+/**
+ * @author ashetkar
+ *
+ */
+public class DeltaPropagationWithCQDUnitTest extends DistributedTestCase {
+
+ private static GemFireCache cache = null;
+
+ private static Pool pool = null;
+
+ private static String regionName = "CQWithInterestDUnitTest_region";
+
+ protected VM server1 = null;
+
+ protected VM server2 = null;
+
+ protected VM client1 = null;
+
+ protected VM client2 = null;
+
+ private static final String CQ1 = "SELECT * FROM " + Region.SEPARATOR
+ + regionName;
+
+ private static long totalEvents = 0;
+
+ private static long cqEvents = 0;
+
+ private static long cqErrors = 0;
+
+ /**
+ * @param name
+ */
+ public DeltaPropagationWithCQDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ super.setUp();
+ Host host = Host.getHost(0);
+ server1 = host.getVM(0);
+ server2 = host.getVM(1);
+ client1 = host.getVM(2);
+ client2 = host.getVM(3);
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ server1.invoke(DeltaPropagationWithCQDUnitTest.class, "close");
+ server2.invoke(DeltaPropagationWithCQDUnitTest.class, "close");
+ client1.invoke(DeltaPropagationWithCQDUnitTest.class, "close");
+ client2.invoke(DeltaPropagationWithCQDUnitTest.class, "close");
+ close();
+ }
+
+ public static void close() throws Exception {
+ if (cache != null && !cache.isClosed()) {
+ cache.close();
+ }
+ totalEvents = 0;
+ cqEvents = 0;
+ cqErrors = 0;
+ }
+
+ public void testCqWithRI() throws Exception {
+ // 1. setup a cache server
+ int port = (Integer)server1.invoke(DeltaPropagationWithCQDUnitTest.class,
+ "createCacheServer");
+ // 2. setup a client
+ client1
+ .invoke(DeltaPropagationWithCQDUnitTest.class, "createClientCache",
+ new Object[] {getServerHostName(server1.getHost()), port,
+ Boolean.TRUE});
+ // 3. setup another client with cqs and interest in all keys.
+ createClientCache(getServerHostName(server1.getHost()), port, true);
+ registerCQs(1, "CQWithInterestDUnitTest_cq");
+ // 4. put a key on client1
+ client1.invoke(DeltaPropagationWithCQDUnitTest.class, "doPut", new Object[] {
+ "SAMPLE_KEY", "SAMPLE_VALUE"});
+ // 5. update the key with new value, on client1
+ client1.invoke(DeltaPropagationWithCQDUnitTest.class, "doPut", new Object[] {
+ "SAMPLE_KEY", "NEW_VALUE"});
+ // 6. Wait for some time
+ WaitCriterion wc = new WaitCriterion() {
+ public boolean done() {
+ return cqEvents == 2 && cqErrors == 0;
+ }
+ public String description() {
+ return "Expected 2 cqEvents and 0 cqErrors, but found " + cqEvents
+ + " cqEvents and " + cqErrors + " cqErrors";
+ }
+ };
+ DistributedTestCase.waitForCriterion(wc, 30 * 1000, 100, true);
+
+ // 7. validate that client2 has the new value
+ assertEquals("Latest value: ", "NEW_VALUE", cache.getRegion(regionName)
+ .get("SAMPLE_KEY"));
+ }
+
+ public void testFullValueRequestsWithCqWithoutRI() throws Exception {
+ int numOfListeners = 5;
+ int numOfKeys = 10;
+ int numOfCQs = 3;
+ // 1. setup a cache server
+ int port = (Integer)server1.invoke(DeltaPropagationWithCQDUnitTest.class,
+ "createCacheServer");
+ // 2. setup a client with register interest
+ client1
+ .invoke(DeltaPropagationWithCQDUnitTest.class, "createClientCache",
+ new Object[] {getServerHostName(server1.getHost()), port,
+ Boolean.TRUE});
+ // 3. setup another client with cqs but without interest.
+ createClientCache(getServerHostName(server1.getHost()), port, false/*RI*/);
+ for (int i = 0; i < numOfCQs; i++) {
+ registerCQs(numOfListeners, "Query_"+i);
+ }
+ // 4. do delta creates
+ doPuts(numOfKeys, true);
+ // verify client2's CQ listeners see above puts
+ verifyCqListeners(numOfListeners * numOfKeys * numOfCQs);
+ // verify full value requests at server are zero
+ server1.invoke(DeltaPropagationWithCQDUnitTest.class,
+ "verifyFullValueRequestsFromClients", new Object[] {0L});
+
+ // 4. do delta updates on client1
+ client1.invoke(DeltaPropagationWithCQDUnitTest.class, "doPuts",
+ new Object[] {numOfKeys, true});
+ // verify client2's CQ listeners see above puts
+ verifyCqListeners(numOfListeners * numOfKeys * numOfCQs * 2);
+ // verify full value requests at server
+ server1.invoke(DeltaPropagationWithCQDUnitTest.class,
+ "verifyFullValueRequestsFromClients", new Object[] {10L});
+ }
+
+ public static void verifyCqListeners(final Integer events) throws Exception {
+ WaitCriterion wc = new WaitCriterion() {
+ public String description() {
+ return "Expected " + events + " listener invocations but found "
+ + (cqEvents + cqErrors);
+ }
+
+ public boolean done() {
+ return (cqEvents + cqErrors) == events;
+ }
+ };
+ DistributedTestCase.waitForCriterion(wc, 10000, 100, true);
+ }
+
+ public static void verifyFullValueRequestsFromClients(Long expected)
+ throws Exception {
+ Object[] proxies = ((CacheServerImpl)((GemFireCacheImpl)cache)
+ .getCacheServers().get(0)).getAcceptor().getCacheClientNotifier()
+ .getClientProxies().toArray();
+ long fullValueRequests = ((CacheClientProxy)proxies[0]).getStatistics()
+ .getDeltaFullMessagesSent();
+ if (fullValueRequests == 0) {
+ assertEquals("Full value requests, ", expected.longValue(),
+ ((CacheClientProxy)proxies[1]).getStatistics()
+ .getDeltaFullMessagesSent());
+ } else {
+ assertEquals("Full value requests, ", expected.longValue(),
+ fullValueRequests);
+ }
+ }
+
+ public static void doPut(Object key, Object value) throws Exception {
+ Region<Object, Object> region = cache.getRegion(regionName);
+ region.put(key, value);
+ }
+
+ public static void doPuts(Integer num, Boolean useDelta) throws Exception {
+ Region<Object, Object> region = cache.getRegion(regionName);
+ for (int i = 0; i < num; i++) {
+ if (useDelta) {
+ DeltaTestImpl delta = new DeltaTestImpl(i, "VALUE_"+i);
+ delta.setIntVar(i);
+ region.put("KEY_" + i, delta);
+ } else {
+ region.put("KEY_" + i, i);
+ }
+ }
+ }
+
+ public static Integer createCacheServer() throws Exception {
+ DeltaPropagationWithCQDUnitTest instance = new DeltaPropagationWithCQDUnitTest("temp");
+ Properties props = new Properties();
+ DistributedSystem ds = instance.getSystem(props);
+ ds.disconnect();
+ ds = instance.getSystem(props);
+ assertNotNull(ds);
+ cache = CacheFactory.create(ds);
+ assertNotNull(cache);
+// Properties props = new Properties();
+// props.setProperty(DistributionConfig.MCAST_PORT_NAME, "10333");
+// cache = new CacheFactory(props).create();
+ RegionFactory<Object, Object> rf = ((Cache)cache)
+ .createRegionFactory(RegionShortcut.REPLICATE);
+ rf.create(regionName);
+ CacheServer server = ((Cache)cache).addCacheServer();
+ server.setPort(AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET));
+ server.start();
+ return server.getPort();
+ }
+
+ public static void createClientCache(String host, Integer port, Boolean doRI)
+ throws Exception {
+ Properties props = new Properties();
+ props.setProperty(DistributionConfig.MCAST_PORT_NAME, "0");
+ props.setProperty(DistributionConfig.LOCATORS_NAME, "");
+ DeltaPropagationWithCQDUnitTest instance = new DeltaPropagationWithCQDUnitTest("temp");
+ DistributedSystem ds = instance.getSystem(props);
+ ds.disconnect();
+ ds = instance.getSystem(props);
+ assertNotNull(ds);
+ cache = CacheFactory.create(ds);
+ assertNotNull(cache);
+ AttributesFactory factory = new AttributesFactory();
+ pool = ClientServerTestCase.configureConnectionPool(factory, "localhost", new int[]{port},
+ true, 1, 2, null, 1000, 250, false, -2);
+
+ factory.setScope(Scope.LOCAL);
+// String poolName = "CQWithInterestDUnitTest_pool";
+// cache = new ClientCacheFactory(new Properties()).create();
+// ClientRegionFactory<Object, Object> rf = ((ClientCache)cache)
+// .createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY);
+// PoolFactory pf = PoolManager.createFactory().addServer(host, port)
+// .setSubscriptionEnabled(true);
+// pool = pf.create(poolName);
+// rf.setPoolName(poolName);
+// rf.addCacheListener(new CacheListenerAdapter<Object, Object>() {
+ factory.addCacheListener(new CacheListenerAdapter<Object, Object>() {
+ public void afterCreate(EntryEvent<Object, Object> event) {
+ totalEvents++;
+ }
+
+ public void afterUpdate(EntryEvent<Object, Object> event) {
+ totalEvents++;
+ }
+
+ public void afterDestroy(EntryEvent<Object, Object> event) {
+ totalEvents++;
+ }
+
+ public void afterInvalidate(EntryEvent<Object, Object> event) {
+ totalEvents++;
+ }
+ });
+ RegionAttributes attr = factory.create();
+ Region region = ((Cache)cache).createRegion(regionName, attr);
+ if (doRI) {
+ region.registerInterest("ALL_KEYS");
+ }
+ }
+
+ public static void registerCQs(Integer numOfListeners, String name) throws Exception {
+ QueryService qs = pool.getQueryService();
+
+ CqAttributesFactory caf = new CqAttributesFactory();
+
+ CqListenerAdapter[] cqListeners = new CqListenerAdapter[numOfListeners];
+ for (int i = 0; i < numOfListeners; i++) {
+ cqListeners[i] = new CqListenerAdapter() {
+ public void onEvent(CqEvent event) {
+ event.getNewValue();
+ cqEvents++;
+ }
+
+ public void onError(CqEvent event) {
+ event.getNewValue();
+ cqErrors++;
+ }
+ };
+ caf.addCqListener(cqListeners[i]);
+ }
+
+ CqQuery cQuery = qs.newCq(name, CQ1, caf.create());
+ cQuery.execute();
+
+ if (qs.getCq(name) == null) {
+ fail("Failed to get CQ " + name);
+ }
+ }
+}
+