geode-core/src/distributedTest/java/org/apache/geode/internal/cache/tier/sockets/ClientInterestNotifyDUnitTest.java - geode - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
  * agreements. See the NOTICE file distributed with this work for additional information regarding
  * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance with the License. You may obtain a
  * copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software distributed under the License
  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  * or implied. See the License for the specific language governing permissions and limitations under
  * the License.
  */
 package org.apache.geode.internal.cache.tier.sockets;

 import static org.apache.geode.cache.Region.SEPARATOR;
 import static org.apache.geode.distributed.ConfigurationProperties.DELTA_PROPAGATION;
 import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
 import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
 import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;

 import java.util.Iterator;
 import java.util.Properties;

 import org.junit.Test;
 import org.junit.experimental.categories.Category;

 import org.apache.geode.cache.AttributesFactory;
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.CacheFactory;
 import org.apache.geode.cache.CacheWriterException;
 import org.apache.geode.cache.DataPolicy;
 import org.apache.geode.cache.EntryEvent;
 import org.apache.geode.cache.InterestResultPolicy;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionAttributes;
 import org.apache.geode.cache.Scope;
 import org.apache.geode.cache.client.Pool;
 import org.apache.geode.cache.client.PoolFactory;
 import org.apache.geode.cache.client.PoolManager;
 import org.apache.geode.cache.server.CacheServer;
 import org.apache.geode.cache.util.CacheListenerAdapter;
 import org.apache.geode.distributed.DistributedSystem;
 import org.apache.geode.internal.cache.CacheServerImpl;
 import org.apache.geode.internal.cache.GemFireCacheImpl;
 import org.apache.geode.test.awaitility.GeodeAwaitility;
 import org.apache.geode.test.dunit.Assert;
 import org.apache.geode.test.dunit.Host;
 import org.apache.geode.test.dunit.LogWriterUtils;
 import org.apache.geode.test.dunit.NetworkUtils;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.WaitCriterion;
 import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
 import org.apache.geode.test.junit.categories.ClientServerTest;

 /**
  * This test verifies the per-client notify-by-subscription (NBS) override functionality along with
  * register interest new receiveValues flag. Taken from the existing ClientConflationDUnitTest.java
  * and modified.
  *
  * @since GemFire 6.0.3
  */
 @Category({ClientServerTest.class})
 public class ClientInterestNotifyDUnitTest extends JUnit4DistributedTestCase {

   class EventListener extends CacheListenerAdapter {
     public EventListener(String name) {
       m_name = name;
     }

     private String m_name = null;
     private int m_creates = 0;
     private int m_updates = 0;
     private int m_invalidates = 0;
     private int m_destroys = 0;

     @Override
     public void afterCreate(EntryEvent event) {
       m_creates++;
     }

     @Override
     public void afterUpdate(EntryEvent event) {
       m_updates++;
     }

     @Override
     public void afterInvalidate(EntryEvent event) {
       m_invalidates++;
     }

     @Override
     public void afterDestroy(EntryEvent event) {
       m_destroys++;
     }

     public void reset() {
       m_creates = 0;
       m_updates = 0;
       m_invalidates = 0;
       m_destroys = 0;
     }

     // validate expected event counts with actual counted
     public void validate(int creates, int updates, int invalidates, int destroys) {
       // Wait for the last destroy event to arrive.
       try {
         await().until(() -> {
           return (destroys == m_destroys);
         });
       } catch (Exception ex) {
         // The event is not received in a given wait time.
         // The check will be done below to report the valid reason.
       }

       GemFireCacheImpl.getInstance().getLogger()
           .info(m_name + ": creates: expected=" + creates + ", actual=" + m_creates);
       GemFireCacheImpl.getInstance().getLogger()
           .info(m_name + ": updates: expected=" + updates + ", actual=" + m_updates);
       GemFireCacheImpl.getInstance().getLogger()
           .info(m_name + ": invalidates: expected=" + invalidates + ", actual=" + m_invalidates);
       GemFireCacheImpl.getInstance().getLogger()
           .info(m_name + ": destroys: expected=" + destroys + ", actual=" + m_destroys);

       assertEquals("Creates :", creates, m_creates);
       assertEquals("Updates :", updates, m_updates);
       assertEquals("Invalidates :", invalidates, m_invalidates);
       assertEquals("Destroys :", destroys, m_destroys);
     }
   }

   // server is on master controller
   VM vm0 = null; // feeder
   VM vm1 = null; // client1
   /*
    * VM vm2 = null; // client2 VM vm3 = null; // client3
    */
   private static Cache cacheServer = null;
   private int PORT;
   private static int poolNameCounter = 0;
   // Region 1 only does interest registrations with receiveValues flag set to true
   private static final String REGION_NAME1 = "ClientInterestNotifyDUnitTest_region1";
   // Region 2 only does interest registrations with receiveValues flag set to false
   private static final String REGION_NAME2 = "ClientInterestNotifyDUnitTest_region2";
   // Region 3 does NOT register any interest
   private static final String REGION_NAME3 = "ClientInterestNotifyDUnitTest_region3";

   @Override
   public final void postSetUp() throws Exception {
     disconnectAllFromDS();

     final Host host = Host.getHost(0);
     vm0 = host.getVM(0);
     vm1 = host.getVM(1);
   }

   private Cache createCache(Properties props) throws Exception {
     props.setProperty(MCAST_PORT, "0");
     props.setProperty(LOCATORS, "");
     DistributedSystem ds = DistributedSystem.connect(props);
     Cache cache = CacheFactory.create(ds);
     if (cache == null) {
       throw new Exception("CacheFactory.create() returned null ");
     }
     return cache;
   }

   @Test
   public void testInterestNotify() throws Exception {
     performSteps();
   }

   private void performSteps() throws Exception {

     // Server is created with notify-by-subscription (NBS) set to false.
     PORT = createServerCache();

     Host host = Host.getHost(0);
     // Create a feeder.
     vm0.invoke(() -> ClientInterestNotifyDUnitTest
         .createClientCacheFeeder(NetworkUtils.getServerHostName(host), new Integer(PORT)));

     // Client 1 overrides NBS to true.
     // Client 2 "overrides" NSB to false.
     // Client 3 uses the default NBS which is false on the server.

     vm1.invoke(() -> ClientInterestNotifyDUnitTest
         .createClientCache(NetworkUtils.getServerHostName(host), new Integer(PORT), "ClientOn"));

     // Feeder doFeed does one put on one key for each of the 3 regions so
     // that the following client RI with ALL_KEYS and KEYS_VALUE result works.

     vm0.invoke(() -> ClientInterestNotifyDUnitTest.doFeed());

     // RI on ALL_KEYS with InterestResultPolicy KEYS_VALUES.

     vm1.invoke(() -> ClientInterestNotifyDUnitTest.registerInterest());

     // Get key for region 3 for all clients to check no unwanted notifications
     // arrive on client 1 region 3 since we do not register interest on any
     // client but notifications should arrive for client 2 and client 3.

     vm1.invoke(() -> ClientInterestNotifyDUnitTest.getEntries());

     // Feeder doEntryOps does 2 puts, 1 invalidate and 1 destroy on a
     // single key for each of the 3 regions.

     vm0.invoke(() -> ClientInterestNotifyDUnitTest.doEntryOps());

     waitForQueuesToDrain();

     // Unregister interest to check it works and no extra notifications received.

     vm1.invoke(() -> ClientInterestNotifyDUnitTest.unregisterInterest());

     // Feeder doEntryOps again does 2 puts, 1 invalidate and 1 destroy on a
     // single key for each of the 3 regions while no interest on the clients.

     vm0.invoke(() -> ClientInterestNotifyDUnitTest.doEntryOps());

     assertAllQueuesEmpty(); // since no client has registered interest

     // Re-register interest on all clients except for region 3 again.

     vm1.invoke(() -> ClientInterestNotifyDUnitTest.registerInterest());

     // Feeder doEntryOps again does 2 puts, 1 invalidate and 1 destroy on a
     // single key for each of the 3 regions after clients re-register interest.

     vm0.invoke(() -> ClientInterestNotifyDUnitTest.doEntryOps());

     waitForQueuesToDrain();

     // doValidation on all listeners:

     // Client invokes listeners as follows:
     // 1. For update notification:
     // a. afterCreate() if the key does not exist in the local cache.
     // b. afterUpdate() if the key exists even if the entry is invalid.
     // 2. For invalidate notification:
     // a. afterInvalidate() only if the key/entry exists and is not already invalid.
     // 3. For destroy notification:
     // a. afterDestroy() only if the key/entry is not already destroyed.

     // RI is done for ALL_KEYS with InterestResultPolicy KEYS_VALUES.

     // For region 1 we set the RI receiveValues param to true so
     // that it behaves like notify-by-subscription is set to true.
     // For region 2 we set the RI receiveValues param to false so
     // that it behaves like notify-by-subscription is set to false but
     // only for matching interested keys.
     // For region 3 we do not register any interest to check that it
     // does not get any events other than those expected from the
     // server's notify-by-subscription setting which may be overridden.

     // Region 3 calls a single afterCreate() because of the single get() done.
     // This is to verify client 1 does not receive any unwanted event traffic.
     // Region 3 of clients 2 and 3 receive an invalidate and a destroy each
     // because NBS is set to false for those clients.

     vm1.invoke(() -> ClientInterestNotifyDUnitTest.doValidation(REGION_NAME1, 1, 3, 2, 2));
     vm1.invoke(() -> ClientInterestNotifyDUnitTest.doValidation(REGION_NAME2, 0, 0, 1, 1));
     vm1.invoke(() -> ClientInterestNotifyDUnitTest.doValidation(REGION_NAME3, 1, 0, 0, 0));

   }

   /**
    * create properties for a loner VM
    */
   private static Properties createProperties1(/* String nbs */) {
     Properties props = new Properties();
     props.setProperty(MCAST_PORT, "0");
     props.setProperty(LOCATORS, "");
     return props;
   }


   private static void createPool2(String host, AttributesFactory factory, Integer port) {
     PoolFactory pf = PoolManager.createFactory();
     pf.addServer(host, port.intValue()).setSubscriptionEnabled(true)
         .setReadTimeout(10000).setSocketBufferSize(32768).setPingInterval(1000).setMinConnections(3)
         .setSubscriptionRedundancy(-1);
     Pool pool = pf.create("superpoolish" + (poolNameCounter++));
     factory.setPoolName(pool.getName());
   }

   /**
    * Do validation based on feeder events received by the client
    */
   public static void doValidation(String region, int creates, int updates, int invalidates,
       int destroys) {
     Cache cacheClient = GemFireCacheImpl.getInstance();
     EventListener listener = null;
     listener = (EventListener) cacheClient.getRegion(region).getAttributes().getCacheListeners()[0];
     listener.validate(creates, updates, invalidates, destroys);
   }

   /**
    * create client with 3 regions each with a unique listener
    *
    */
   public static void createClientCache(String host, Integer port, /* String nbs, */
       String name) throws Exception {
     ClientInterestNotifyDUnitTest test = new ClientInterestNotifyDUnitTest();
     Cache cacheClient = test.createCache(createProperties1(/* nbs */));
     AttributesFactory factory = new AttributesFactory();
     factory.setScope(Scope.LOCAL);
     factory.setConcurrencyChecksEnabled(false);
     createPool2(host, factory, port);
     factory.setCacheListener(test.new EventListener(name + REGION_NAME1));
     RegionAttributes attrs = factory.create();
     cacheClient.createRegion(REGION_NAME1, attrs);

     factory = new AttributesFactory();
     factory.setScope(Scope.LOCAL);
     factory.setConcurrencyChecksEnabled(false);
     createPool2(host, factory, port);
     factory.setCacheListener(test.new EventListener(name + REGION_NAME2));
     attrs = factory.create();
     cacheClient.createRegion(REGION_NAME2, attrs);

     factory = new AttributesFactory();
     factory.setScope(Scope.LOCAL);
     factory.setConcurrencyChecksEnabled(false);
     createPool2(host, factory, port);
     factory.setCacheListener(test.new EventListener(name + REGION_NAME3));
     attrs = factory.create();
     cacheClient.createRegion(REGION_NAME3, attrs);
   }

   public static void createClientCacheFeeder(String host, Integer port) throws Exception {
     ClientInterestNotifyDUnitTest test = new ClientInterestNotifyDUnitTest();
     Cache cacheFeeder = test.createCache(createProperties1(
     /* DistributionConfig.NOTIFY_BY_SUBSCRIPTION_OVERRIDE_PROP_VALUE_DEFAULT */));
     AttributesFactory factory = new AttributesFactory();
     factory.setScope(Scope.LOCAL);
     factory.setConcurrencyChecksEnabled(false);
     createPool2(host, factory, port);
     RegionAttributes attrs = factory.create();
     cacheFeeder.createRegion(REGION_NAME1, attrs);
     cacheFeeder.createRegion(REGION_NAME2, attrs);
     cacheFeeder.createRegion(REGION_NAME3, attrs);
   }

   /**
    * Assert all queues are empty to aid later assertion for listener event counts.
    */

   // NOTE: replaced with waitForQueuesToDrain() using waitcriterion to avoid
   // occasional failures in precheckins and cruisecontrol.

   public static void assertAllQueuesEmpty() {
     Iterator servers = cacheServer.getCacheServers().iterator();
     assertTrue("No servers found!", servers.hasNext());
     while (servers.hasNext()) {
       Iterator proxies = ((CacheServerImpl) servers.next()).getAcceptor().getCacheClientNotifier()
           .getClientProxies().iterator();
       assertTrue("No proxies found!", proxies.hasNext());
       while (proxies.hasNext()) {
         int qsize = ((CacheClientProxy) proxies.next()).getQueueSize();
         assertTrue("Queue size expected to be zero but is " + qsize, qsize == 0);
       }
     }
   }

   public static void waitForQueuesToDrain() {
     WaitCriterion wc = new WaitCriterion() {
       String excuse;

       @Override
       public boolean done() {
         // assume a single cache server as configured in this test
         CacheServerImpl bridgeServer =
             (CacheServerImpl) cacheServer.getCacheServers().iterator().next();
         if (bridgeServer == null) {
           excuse = "No Cache Server";
           return false;
         }
         Iterator proxies =
             bridgeServer.getAcceptor().getCacheClientNotifier().getClientProxies().iterator();
         if (!proxies.hasNext()) {
           excuse = "No CacheClientProxy";
           return false;
         }
         while (proxies.hasNext()) {
           CacheClientProxy proxy = (CacheClientProxy) proxies.next();
           if (proxy == null) {
             excuse = "No CacheClientProxy";
             return false;
           }
           // Verify the queue size
           int sz = proxy.getQueueSize();
           if (0 != sz) {
             excuse = "Queue did not drain. Expected size = 0, actual = " + sz + "for " + proxy;
             return false;
           }
         }
         return true;
       }

       @Override
       public String description() {
         return excuse;
       }
     };
     GeodeAwaitility.await().untilAsserted(wc);
   }

   /**
    * create a server cache and start the server
    *
    */
   public static Integer createServerCache() throws Exception {
     ClientInterestNotifyDUnitTest test = new ClientInterestNotifyDUnitTest();
     Properties props = new Properties();
     props.setProperty(DELTA_PROPAGATION, "false");
     cacheServer = test.createCache(props);
     AttributesFactory factory = new AttributesFactory();
     factory.setScope(Scope.DISTRIBUTED_ACK);
     factory.setDataPolicy(DataPolicy.REPLICATE);
     factory.setConcurrencyChecksEnabled(false);
     RegionAttributes attrs = factory.create();
     cacheServer.createRegion(REGION_NAME1, attrs);
     cacheServer.createRegion(REGION_NAME2, attrs);
     cacheServer.createRegion(REGION_NAME3, attrs);
     CacheServer server = cacheServer.addCacheServer();
     int port = getRandomAvailableTCPPort();
     server.setPort(port);
     server.setNotifyBySubscription(true);
     server.setSocketBufferSize(32768);
     server.start();
     return new Integer(server.getPort());
   }

   /**
    * close the client cache
    *
    */
   public static void closeCache() {
     Cache cacheClient = GemFireCacheImpl.getInstance();
     if (cacheClient != null && !cacheClient.isClosed()) {
       cacheClient.close();
       cacheClient.getDistributedSystem().disconnect();
     }
   }

   /**
    * close the server cache
    *
    */
   public static void closeCacheServer() {
     if (cacheServer != null && !cacheServer.isClosed()) {
       cacheServer.close();
       cacheServer.getDistributedSystem().disconnect();
     }
   }

   /**
    * register interest with the server on ALL_KEYS
    *
    */
   public static void registerInterest() {
     try {
       Cache cacheClient = GemFireCacheImpl.getInstance();
       Region region1 = cacheClient.getRegion(SEPARATOR + REGION_NAME1);
       Region region2 = cacheClient.getRegion(SEPARATOR + REGION_NAME2);
       // We intentionally do not register interest in region 3 to check no events recvd.
       Region region3 = cacheClient.getRegion(SEPARATOR + REGION_NAME3);
       assertTrue(region1 != null);
       assertTrue(region2 != null);
       assertTrue(region3 != null);
       region1.registerInterest("ALL_KEYS", InterestResultPolicy.KEYS_VALUES, false, true);
       region2.registerInterest("ALL_KEYS", InterestResultPolicy.KEYS_VALUES, false, false);
       // We intentionally do not register interest in region 3 to check no events recvd.
       // region3.registerInterestBlah();
     } catch (CacheWriterException e) {
       fail("test failed due to " + e);
     }
   }


   /**
    * register interest with the server on ALL_KEYS
    *
    */

   public static void unregisterInterest() {
     try {
       Cache cacheClient = GemFireCacheImpl.getInstance();
       Region region1 = cacheClient.getRegion(SEPARATOR + REGION_NAME1);
       Region region2 = cacheClient.getRegion(SEPARATOR + REGION_NAME2);
       region1.unregisterInterest("ALL_KEYS");
       region2.unregisterInterest("ALL_KEYS");
     } catch (CacheWriterException e) {
       fail("test failed due to " + e);
     }
   }

   /**
    * Do 2 puts, 1 invalidate and 1 destroy for a key "key-1"
    */
   public static void doEntryOps() {
     try {
       LogWriterUtils.getLogWriter().info("Putting entries...");
       Cache cacheClient = GemFireCacheImpl.getInstance();
       Region r1 = cacheClient.getRegion(SEPARATOR + REGION_NAME1);
       Region r2 = cacheClient.getRegion(SEPARATOR + REGION_NAME2);
       Region r3 = cacheClient.getRegion(SEPARATOR + REGION_NAME3);
       r1.put("key-1", "11");
       r2.put("key-1", "11");
       r3.put("key-1", "11");
       r1.put("key-1", "22");
       r2.put("key-1", "22");
       r3.put("key-1", "22");
       r1.invalidate("key-1");
       r2.invalidate("key-1");
       r3.invalidate("key-1");
       r1.destroy("key-1");
       r2.destroy("key-1");
       r3.destroy("key-1");
     } catch (Exception ex) {
       ex.printStackTrace();
       Assert.fail("failed while region doing ops", ex);
     }
   }

   /**
    * Do initial puts
    */
   public static void doFeed() {
     try {
       LogWriterUtils.getLogWriter().info("Putting entries...");
       Cache cacheClient = GemFireCacheImpl.getInstance();
       Region r1 = cacheClient.getRegion(SEPARATOR + REGION_NAME1);
       Region r2 = cacheClient.getRegion(SEPARATOR + REGION_NAME2);
       Region r3 = cacheClient.getRegion(SEPARATOR + REGION_NAME3);
       r1.put("key-1", "00");
       r2.put("key-1", "00");
       r3.put("key-1", "00");
     } catch (Exception ex) {
       ex.printStackTrace();
       Assert.fail("failed while region doing ops", ex);
     }
   }

   /**
    * Get entries on all clients' region 3 since it does not register interest.
    */
   public static void getEntries() {
     try {
       LogWriterUtils.getLogWriter().info("Getting entries...");
       Cache cacheClient = GemFireCacheImpl.getInstance();
       Region r3 = cacheClient.getRegion(SEPARATOR + REGION_NAME3);
       r3.get("key-1");
     } catch (Exception ex) {
       ex.printStackTrace();
       Assert.fail("failed while region doing ops", ex);
     }
   }

   /**
    * close the caches in tearDown
    */
   @Override
   public final void preTearDown() throws Exception {
     vm0.invoke(() -> ClientInterestNotifyDUnitTest.closeCache());
     vm1.invoke(() -> ClientInterestNotifyDUnitTest.closeCache());
     closeCacheServer();
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more contributor license
	* agreements. See the NOTICE file distributed with this work for additional information regarding
	* copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance with the License. You may obtain a
	* copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software distributed under the License
	* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
	* or implied. See the License for the specific language governing permissions and limitations under
	* the License.
	*/
	package org.apache.geode.internal.cache.tier.sockets;

	import static org.apache.geode.cache.Region.SEPARATOR;
	import static org.apache.geode.distributed.ConfigurationProperties.DELTA_PROPAGATION;
	import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
	import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
	import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertTrue;
	import static org.junit.Assert.fail;

	import java.util.Iterator;
	import java.util.Properties;

	import org.junit.Test;
	import org.junit.experimental.categories.Category;

	import org.apache.geode.cache.AttributesFactory;
	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.CacheFactory;
	import org.apache.geode.cache.CacheWriterException;
	import org.apache.geode.cache.DataPolicy;
	import org.apache.geode.cache.EntryEvent;
	import org.apache.geode.cache.InterestResultPolicy;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionAttributes;
	import org.apache.geode.cache.Scope;
	import org.apache.geode.cache.client.Pool;
	import org.apache.geode.cache.client.PoolFactory;
	import org.apache.geode.cache.client.PoolManager;
	import org.apache.geode.cache.server.CacheServer;
	import org.apache.geode.cache.util.CacheListenerAdapter;
	import org.apache.geode.distributed.DistributedSystem;
	import org.apache.geode.internal.cache.CacheServerImpl;
	import org.apache.geode.internal.cache.GemFireCacheImpl;
	import org.apache.geode.test.awaitility.GeodeAwaitility;
	import org.apache.geode.test.dunit.Assert;
	import org.apache.geode.test.dunit.Host;
	import org.apache.geode.test.dunit.LogWriterUtils;
	import org.apache.geode.test.dunit.NetworkUtils;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.WaitCriterion;
	import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
	import org.apache.geode.test.junit.categories.ClientServerTest;

	/**
	* This test verifies the per-client notify-by-subscription (NBS) override functionality along with
	* register interest new receiveValues flag. Taken from the existing ClientConflationDUnitTest.java
	* and modified.
	*
	* @since GemFire 6.0.3
	*/
	@Category({ClientServerTest.class})
	public class ClientInterestNotifyDUnitTest extends JUnit4DistributedTestCase {

	class EventListener extends CacheListenerAdapter {
	public EventListener(String name) {
	m_name = name;
	}

	private String m_name = null;
	private int m_creates = 0;
	private int m_updates = 0;
	private int m_invalidates = 0;
	private int m_destroys = 0;

	@Override
	public void afterCreate(EntryEvent event) {
	m_creates++;
	}

	@Override
	public void afterUpdate(EntryEvent event) {
	m_updates++;
	}

	@Override
	public void afterInvalidate(EntryEvent event) {
	m_invalidates++;
	}

	@Override
	public void afterDestroy(EntryEvent event) {
	m_destroys++;
	}

	public void reset() {
	m_creates = 0;
	m_updates = 0;
	m_invalidates = 0;
	m_destroys = 0;
	}

	// validate expected event counts with actual counted
	public void validate(int creates, int updates, int invalidates, int destroys) {
	// Wait for the last destroy event to arrive.
	try {
	await().until(() -> {
	return (destroys == m_destroys);
	});
	} catch (Exception ex) {
	// The event is not received in a given wait time.
	// The check will be done below to report the valid reason.
	}

	GemFireCacheImpl.getInstance().getLogger()
	.info(m_name + ": creates: expected=" + creates + ", actual=" + m_creates);
	GemFireCacheImpl.getInstance().getLogger()
	.info(m_name + ": updates: expected=" + updates + ", actual=" + m_updates);
	GemFireCacheImpl.getInstance().getLogger()
	.info(m_name + ": invalidates: expected=" + invalidates + ", actual=" + m_invalidates);
	GemFireCacheImpl.getInstance().getLogger()
	.info(m_name + ": destroys: expected=" + destroys + ", actual=" + m_destroys);

	assertEquals("Creates :", creates, m_creates);
	assertEquals("Updates :", updates, m_updates);
	assertEquals("Invalidates :", invalidates, m_invalidates);
	assertEquals("Destroys :", destroys, m_destroys);
	}
	}

	// server is on master controller
	VM vm0 = null; // feeder
	VM vm1 = null; // client1
	/*
	* VM vm2 = null; // client2 VM vm3 = null; // client3
	*/
	private static Cache cacheServer = null;
	private int PORT;
	private static int poolNameCounter = 0;
	// Region 1 only does interest registrations with receiveValues flag set to true
	private static final String REGION_NAME1 = "ClientInterestNotifyDUnitTest_region1";
	// Region 2 only does interest registrations with receiveValues flag set to false
	private static final String REGION_NAME2 = "ClientInterestNotifyDUnitTest_region2";
	// Region 3 does NOT register any interest
	private static final String REGION_NAME3 = "ClientInterestNotifyDUnitTest_region3";

	@Override
	public final void postSetUp() throws Exception {
	disconnectAllFromDS();

	final Host host = Host.getHost(0);
	vm0 = host.getVM(0);
	vm1 = host.getVM(1);
	}

	private Cache createCache(Properties props) throws Exception {
	props.setProperty(MCAST_PORT, "0");
	props.setProperty(LOCATORS, "");
	DistributedSystem ds = DistributedSystem.connect(props);
	Cache cache = CacheFactory.create(ds);
	if (cache == null) {
	throw new Exception("CacheFactory.create() returned null ");
	}
	return cache;
	}

	@Test
	public void testInterestNotify() throws Exception {
	performSteps();
	}

	private void performSteps() throws Exception {

	// Server is created with notify-by-subscription (NBS) set to false.
	PORT = createServerCache();

	Host host = Host.getHost(0);
	// Create a feeder.
	vm0.invoke(() -> ClientInterestNotifyDUnitTest
	.createClientCacheFeeder(NetworkUtils.getServerHostName(host), new Integer(PORT)));

	// Client 1 overrides NBS to true.
	// Client 2 "overrides" NSB to false.
	// Client 3 uses the default NBS which is false on the server.

	vm1.invoke(() -> ClientInterestNotifyDUnitTest
	.createClientCache(NetworkUtils.getServerHostName(host), new Integer(PORT), "ClientOn"));

	// Feeder doFeed does one put on one key for each of the 3 regions so
	// that the following client RI with ALL_KEYS and KEYS_VALUE result works.

	vm0.invoke(() -> ClientInterestNotifyDUnitTest.doFeed());

	// RI on ALL_KEYS with InterestResultPolicy KEYS_VALUES.

	vm1.invoke(() -> ClientInterestNotifyDUnitTest.registerInterest());

	// Get key for region 3 for all clients to check no unwanted notifications
	// arrive on client 1 region 3 since we do not register interest on any
	// client but notifications should arrive for client 2 and client 3.

	vm1.invoke(() -> ClientInterestNotifyDUnitTest.getEntries());

	// Feeder doEntryOps does 2 puts, 1 invalidate and 1 destroy on a
	// single key for each of the 3 regions.

	vm0.invoke(() -> ClientInterestNotifyDUnitTest.doEntryOps());

	waitForQueuesToDrain();

	// Unregister interest to check it works and no extra notifications received.

	vm1.invoke(() -> ClientInterestNotifyDUnitTest.unregisterInterest());

	// Feeder doEntryOps again does 2 puts, 1 invalidate and 1 destroy on a
	// single key for each of the 3 regions while no interest on the clients.

	vm0.invoke(() -> ClientInterestNotifyDUnitTest.doEntryOps());

	assertAllQueuesEmpty(); // since no client has registered interest

	// Re-register interest on all clients except for region 3 again.

	vm1.invoke(() -> ClientInterestNotifyDUnitTest.registerInterest());

	// Feeder doEntryOps again does 2 puts, 1 invalidate and 1 destroy on a
	// single key for each of the 3 regions after clients re-register interest.

	vm0.invoke(() -> ClientInterestNotifyDUnitTest.doEntryOps());

	waitForQueuesToDrain();

	// doValidation on all listeners:

	// Client invokes listeners as follows:
	// 1. For update notification:
	// a. afterCreate() if the key does not exist in the local cache.
	// b. afterUpdate() if the key exists even if the entry is invalid.
	// 2. For invalidate notification:
	// a. afterInvalidate() only if the key/entry exists and is not already invalid.
	// 3. For destroy notification:
	// a. afterDestroy() only if the key/entry is not already destroyed.

	// RI is done for ALL_KEYS with InterestResultPolicy KEYS_VALUES.

	// For region 1 we set the RI receiveValues param to true so
	// that it behaves like notify-by-subscription is set to true.
	// For region 2 we set the RI receiveValues param to false so
	// that it behaves like notify-by-subscription is set to false but
	// only for matching interested keys.
	// For region 3 we do not register any interest to check that it
	// does not get any events other than those expected from the
	// server's notify-by-subscription setting which may be overridden.

	// Region 3 calls a single afterCreate() because of the single get() done.
	// This is to verify client 1 does not receive any unwanted event traffic.
	// Region 3 of clients 2 and 3 receive an invalidate and a destroy each
	// because NBS is set to false for those clients.

	vm1.invoke(() -> ClientInterestNotifyDUnitTest.doValidation(REGION_NAME1, 1, 3, 2, 2));
	vm1.invoke(() -> ClientInterestNotifyDUnitTest.doValidation(REGION_NAME2, 0, 0, 1, 1));
	vm1.invoke(() -> ClientInterestNotifyDUnitTest.doValidation(REGION_NAME3, 1, 0, 0, 0));

	}

	/**
	* create properties for a loner VM
	*/
	private static Properties createProperties1(/* String nbs */) {
	Properties props = new Properties();
	props.setProperty(MCAST_PORT, "0");
	props.setProperty(LOCATORS, "");
	return props;
	}


	private static void createPool2(String host, AttributesFactory factory, Integer port) {
	PoolFactory pf = PoolManager.createFactory();
	pf.addServer(host, port.intValue()).setSubscriptionEnabled(true)
	.setReadTimeout(10000).setSocketBufferSize(32768).setPingInterval(1000).setMinConnections(3)
	.setSubscriptionRedundancy(-1);
	Pool pool = pf.create("superpoolish" + (poolNameCounter++));
	factory.setPoolName(pool.getName());
	}

	/**
	* Do validation based on feeder events received by the client
	*/
	public static void doValidation(String region, int creates, int updates, int invalidates,
	int destroys) {
	Cache cacheClient = GemFireCacheImpl.getInstance();
	EventListener listener = null;
	listener = (EventListener) cacheClient.getRegion(region).getAttributes().getCacheListeners()[0];
	listener.validate(creates, updates, invalidates, destroys);
	}

	/**
	* create client with 3 regions each with a unique listener
	*
	*/
	public static void createClientCache(String host, Integer port, /* String nbs, */
	String name) throws Exception {
	ClientInterestNotifyDUnitTest test = new ClientInterestNotifyDUnitTest();
	Cache cacheClient = test.createCache(createProperties1(/* nbs */));
	AttributesFactory factory = new AttributesFactory();
	factory.setScope(Scope.LOCAL);
	factory.setConcurrencyChecksEnabled(false);
	createPool2(host, factory, port);
	factory.setCacheListener(test.new EventListener(name + REGION_NAME1));
	RegionAttributes attrs = factory.create();
	cacheClient.createRegion(REGION_NAME1, attrs);

	factory = new AttributesFactory();
	factory.setScope(Scope.LOCAL);
	factory.setConcurrencyChecksEnabled(false);
	createPool2(host, factory, port);
	factory.setCacheListener(test.new EventListener(name + REGION_NAME2));
	attrs = factory.create();
	cacheClient.createRegion(REGION_NAME2, attrs);

	factory = new AttributesFactory();
	factory.setScope(Scope.LOCAL);
	factory.setConcurrencyChecksEnabled(false);
	createPool2(host, factory, port);
	factory.setCacheListener(test.new EventListener(name + REGION_NAME3));
	attrs = factory.create();
	cacheClient.createRegion(REGION_NAME3, attrs);
	}

	public static void createClientCacheFeeder(String host, Integer port) throws Exception {
	ClientInterestNotifyDUnitTest test = new ClientInterestNotifyDUnitTest();
	Cache cacheFeeder = test.createCache(createProperties1(
	/* DistributionConfig.NOTIFY_BY_SUBSCRIPTION_OVERRIDE_PROP_VALUE_DEFAULT */));
	AttributesFactory factory = new AttributesFactory();
	factory.setScope(Scope.LOCAL);
	factory.setConcurrencyChecksEnabled(false);
	createPool2(host, factory, port);
	RegionAttributes attrs = factory.create();
	cacheFeeder.createRegion(REGION_NAME1, attrs);
	cacheFeeder.createRegion(REGION_NAME2, attrs);
	cacheFeeder.createRegion(REGION_NAME3, attrs);
	}

	/**
	* Assert all queues are empty to aid later assertion for listener event counts.
	*/

	// NOTE: replaced with waitForQueuesToDrain() using waitcriterion to avoid
	// occasional failures in precheckins and cruisecontrol.

	public static void assertAllQueuesEmpty() {
	Iterator servers = cacheServer.getCacheServers().iterator();
	assertTrue("No servers found!", servers.hasNext());
	while (servers.hasNext()) {
	Iterator proxies = ((CacheServerImpl) servers.next()).getAcceptor().getCacheClientNotifier()
	.getClientProxies().iterator();
	assertTrue("No proxies found!", proxies.hasNext());
	while (proxies.hasNext()) {
	int qsize = ((CacheClientProxy) proxies.next()).getQueueSize();
	assertTrue("Queue size expected to be zero but is " + qsize, qsize == 0);
	}
	}
	}

	public static void waitForQueuesToDrain() {
	WaitCriterion wc = new WaitCriterion() {
	String excuse;

	@Override
	public boolean done() {
	// assume a single cache server as configured in this test
	CacheServerImpl bridgeServer =
	(CacheServerImpl) cacheServer.getCacheServers().iterator().next();
	if (bridgeServer == null) {
	excuse = "No Cache Server";
	return false;
	}
	Iterator proxies =
	bridgeServer.getAcceptor().getCacheClientNotifier().getClientProxies().iterator();
	if (!proxies.hasNext()) {
	excuse = "No CacheClientProxy";
	return false;
	}
	while (proxies.hasNext()) {
	CacheClientProxy proxy = (CacheClientProxy) proxies.next();
	if (proxy == null) {
	excuse = "No CacheClientProxy";
	return false;
	}
	// Verify the queue size
	int sz = proxy.getQueueSize();
	if (0 != sz) {
	excuse = "Queue did not drain. Expected size = 0, actual = " + sz + "for " + proxy;
	return false;
	}
	}
	return true;
	}

	@Override
	public String description() {
	return excuse;
	}
	};
	GeodeAwaitility.await().untilAsserted(wc);
	}

	/**
	* create a server cache and start the server
	*
	*/
	public static Integer createServerCache() throws Exception {
	ClientInterestNotifyDUnitTest test = new ClientInterestNotifyDUnitTest();
	Properties props = new Properties();
	props.setProperty(DELTA_PROPAGATION, "false");
	cacheServer = test.createCache(props);
	AttributesFactory factory = new AttributesFactory();
	factory.setScope(Scope.DISTRIBUTED_ACK);
	factory.setDataPolicy(DataPolicy.REPLICATE);
	factory.setConcurrencyChecksEnabled(false);
	RegionAttributes attrs = factory.create();
	cacheServer.createRegion(REGION_NAME1, attrs);
	cacheServer.createRegion(REGION_NAME2, attrs);
	cacheServer.createRegion(REGION_NAME3, attrs);
	CacheServer server = cacheServer.addCacheServer();
	int port = getRandomAvailableTCPPort();
	server.setPort(port);
	server.setNotifyBySubscription(true);
	server.setSocketBufferSize(32768);
	server.start();
	return new Integer(server.getPort());
	}

	/**
	* close the client cache
	*
	*/
	public static void closeCache() {
	Cache cacheClient = GemFireCacheImpl.getInstance();
	if (cacheClient != null && !cacheClient.isClosed()) {
	cacheClient.close();
	cacheClient.getDistributedSystem().disconnect();
	}
	}

	/**
	* close the server cache
	*
	*/
	public static void closeCacheServer() {
	if (cacheServer != null && !cacheServer.isClosed()) {
	cacheServer.close();
	cacheServer.getDistributedSystem().disconnect();
	}
	}

	/**
	* register interest with the server on ALL_KEYS
	*
	*/
	public static void registerInterest() {
	try {
	Cache cacheClient = GemFireCacheImpl.getInstance();
	Region region1 = cacheClient.getRegion(SEPARATOR + REGION_NAME1);
	Region region2 = cacheClient.getRegion(SEPARATOR + REGION_NAME2);
	// We intentionally do not register interest in region 3 to check no events recvd.
	Region region3 = cacheClient.getRegion(SEPARATOR + REGION_NAME3);
	assertTrue(region1 != null);
	assertTrue(region2 != null);
	assertTrue(region3 != null);
	region1.registerInterest("ALL_KEYS", InterestResultPolicy.KEYS_VALUES, false, true);
	region2.registerInterest("ALL_KEYS", InterestResultPolicy.KEYS_VALUES, false, false);
	// We intentionally do not register interest in region 3 to check no events recvd.
	// region3.registerInterestBlah();
	} catch (CacheWriterException e) {
	fail("test failed due to " + e);
	}
	}


	/**
	* register interest with the server on ALL_KEYS
	*
	*/

	public static void unregisterInterest() {
	try {
	Cache cacheClient = GemFireCacheImpl.getInstance();
	Region region1 = cacheClient.getRegion(SEPARATOR + REGION_NAME1);
	Region region2 = cacheClient.getRegion(SEPARATOR + REGION_NAME2);
	region1.unregisterInterest("ALL_KEYS");
	region2.unregisterInterest("ALL_KEYS");
	} catch (CacheWriterException e) {
	fail("test failed due to " + e);
	}
	}

	/**
	* Do 2 puts, 1 invalidate and 1 destroy for a key "key-1"
	*/
	public static void doEntryOps() {
	try {
	LogWriterUtils.getLogWriter().info("Putting entries...");
	Cache cacheClient = GemFireCacheImpl.getInstance();
	Region r1 = cacheClient.getRegion(SEPARATOR + REGION_NAME1);
	Region r2 = cacheClient.getRegion(SEPARATOR + REGION_NAME2);
	Region r3 = cacheClient.getRegion(SEPARATOR + REGION_NAME3);
	r1.put("key-1", "11");
	r2.put("key-1", "11");
	r3.put("key-1", "11");
	r1.put("key-1", "22");
	r2.put("key-1", "22");
	r3.put("key-1", "22");
	r1.invalidate("key-1");
	r2.invalidate("key-1");
	r3.invalidate("key-1");
	r1.destroy("key-1");
	r2.destroy("key-1");
	r3.destroy("key-1");
	} catch (Exception ex) {
	ex.printStackTrace();
	Assert.fail("failed while region doing ops", ex);
	}
	}

	/**
	* Do initial puts
	*/
	public static void doFeed() {
	try {
	LogWriterUtils.getLogWriter().info("Putting entries...");
	Cache cacheClient = GemFireCacheImpl.getInstance();
	Region r1 = cacheClient.getRegion(SEPARATOR + REGION_NAME1);
	Region r2 = cacheClient.getRegion(SEPARATOR + REGION_NAME2);
	Region r3 = cacheClient.getRegion(SEPARATOR + REGION_NAME3);
	r1.put("key-1", "00");
	r2.put("key-1", "00");
	r3.put("key-1", "00");
	} catch (Exception ex) {
	ex.printStackTrace();
	Assert.fail("failed while region doing ops", ex);
	}
	}

	/**
	* Get entries on all clients' region 3 since it does not register interest.
	*/
	public static void getEntries() {
	try {
	LogWriterUtils.getLogWriter().info("Getting entries...");
	Cache cacheClient = GemFireCacheImpl.getInstance();
	Region r3 = cacheClient.getRegion(SEPARATOR + REGION_NAME3);
	r3.get("key-1");
	} catch (Exception ex) {
	ex.printStackTrace();
	Assert.fail("failed while region doing ops", ex);
	}
	}

	/**
	* close the caches in tearDown
	*/
	@Override
	public final void preTearDown() throws Exception {
	vm0.invoke(() -> ClientInterestNotifyDUnitTest.closeCache());
	vm1.invoke(() -> ClientInterestNotifyDUnitTest.closeCache());
	closeCacheServer();
	}
	}