geode-core/src/test/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifierTest.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.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.ArgumentMatchers.isA;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;

 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.net.Socket;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;

 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;

 import org.apache.geode.cache.Operation;
 import org.apache.geode.cache.query.internal.cq.ServerCQ;
 import org.apache.geode.internal.cache.DistributedRegion;
 import org.apache.geode.internal.cache.FilterProfile;
 import org.apache.geode.internal.cache.FilterRoutingInfo;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.cache.InternalCacheEvent;
 import org.apache.geode.internal.cache.RegionQueueException;
 import org.apache.geode.internal.statistics.StatisticsClock;
 import org.apache.geode.test.fake.Fakes;

 public class CacheClientNotifierTest {
   @Before
   public void setup() {
     // Perform cleanup on any singletons received from previous test runs, since the
     // CacheClientNotifier is a static and previous tests may not have cleaned up properly.
     shutdownExistingCacheClientNotifier();
   }

   @After
   public void tearDown() {
     shutdownExistingCacheClientNotifier();
   }

   @Test
   public void eventsInClientRegistrationQueueAreSentToClientAfterRegistrationIsComplete()
       throws IOException, ClassNotFoundException, NoSuchMethodException, IllegalAccessException,
       InvocationTargetException, InterruptedException, ExecutionException {
     InternalCache internalCache = Fakes.cache();
     CacheServerStats cacheServerStats = mock(CacheServerStats.class);
     Socket socket = mock(Socket.class);
     ConnectionListener connectionListener = mock(ConnectionListener.class);
     ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
     CacheClientProxy cacheClientProxy = mock(CacheClientProxy.class);
     ClientUpdateMessageImpl clientUpdateMessage = mock(ClientUpdateMessageImpl.class);
     ClientRegistrationMetadata clientRegistrationMetadata = mock(ClientRegistrationMetadata.class);
     StatisticsClock statisticsClock = mock(StatisticsClock.class);

     when(clientRegistrationMetadata.getClientProxyMembershipID()).thenReturn(
         clientProxyMembershipID);

     CacheClientNotifier cacheClientNotifier = CacheClientNotifier.getInstance(internalCache,
         new ClientRegistrationEventQueueManager(), statisticsClock, cacheServerStats, 0, 0,
         connectionListener, null, false);
     final CacheClientNotifier cacheClientNotifierSpy = spy(cacheClientNotifier);

     CountDownLatch waitForEventDispatchCountdownLatch = new CountDownLatch(1);
     CountDownLatch waitForRegistrationCountdownLatch = new CountDownLatch(1);
     ExecutorService registerAndNotifyExecutor = Executors.newFixedThreadPool(2);

     try {
       // We stub out the CacheClientNotifier.registerClientInternal() to do some "work" until
       // a new event is received and queued, as triggered by the waitForEventDispatchCountdownLatch
       doAnswer((i) -> {
         when(cacheClientProxy.getProxyID()).thenReturn(clientProxyMembershipID);
         cacheClientNotifierSpy.addClientProxy(cacheClientProxy);
         waitForRegistrationCountdownLatch.countDown();
         waitForEventDispatchCountdownLatch.await();
         return null;
       }).when(cacheClientNotifierSpy).registerClientInternal(clientRegistrationMetadata, socket,
           false, 0, true);
       List<Callable<Void>> registerAndNotifyTasks = new ArrayList<>();

       // In one thread, we register the new client which should create the temporary client
       // registration event queue. Events will be passed to that queue while registration is
       // underway. Once registration is complete, the queue is drained and the event is processed
       // as normal.
       registerAndNotifyTasks.add(() -> {
         cacheClientNotifierSpy.registerClient(clientRegistrationMetadata, socket, false, 0, true);
         return null;
       });

       // In a second thread, we mock the arrival of a new event. We want to ensure this event
       // goes into the temporary client registration event queue. To do that, we wait on the
       // waitForRegistrationCountdownLatch until registration is underway and the temp queue is
       // created. Once it is, we process the event and notify clients, which should add the event
       // to the temp queue. Finally, we resume registration and after it is complete, we verify
       // that the event was drained, processed, and "delivered" (note message delivery is mocked
       // and results in a no-op).
       registerAndNotifyTasks.add(() -> {
         try {
           waitForRegistrationCountdownLatch.await();

           InternalCacheEvent internalCacheEvent =
               createMockInternalCacheEvent(clientProxyMembershipID, clientUpdateMessage,
                   cacheClientNotifierSpy);

           CacheClientNotifier.notifyClients(internalCacheEvent, clientUpdateMessage);
         } finally {
           // Ensure we always countdown so if the test fails it won't hang due to
           // awaiting on this countdown latch.
           waitForEventDispatchCountdownLatch.countDown();
         }
         return null;
       });

       final List<Future<Void>> futures =
           registerAndNotifyExecutor.invokeAll(registerAndNotifyTasks);

       for (final Future future : futures) {
         future.get();
       }
     } finally {
       // To prevent not cleaning up test resources in case an unexpected exception occurs
       waitForEventDispatchCountdownLatch.countDown();
       waitForRegistrationCountdownLatch.countDown();
       registerAndNotifyExecutor.shutdownNow();
       cacheClientNotifier.shutdown(0);
     }

     verify(cacheClientProxy, times(1)).deliverMessage(isA(HAEventWrapper.class));
   }

   @Test
   public void clientRegistrationFailsQueueStillDrained()
       throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException,
       IllegalAccessException, IOException {
     InternalCache internalCache = Fakes.cache();
     CacheServerStats cacheServerStats = mock(CacheServerStats.class);
     Socket socket = mock(Socket.class);
     ConnectionListener connectionListener = mock(ConnectionListener.class);
     ClientRegistrationMetadata clientRegistrationMetadata = mock(ClientRegistrationMetadata.class);
     StatisticsClock statisticsClock = mock(StatisticsClock.class);
     ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
     ClientRegistrationEventQueueManager clientRegistrationEventQueueManager =
         mock(ClientRegistrationEventQueueManager.class);
     ClientRegistrationEventQueueManager.ClientRegistrationEventQueue clientRegistrationEventQueue =
         mock(ClientRegistrationEventQueueManager.ClientRegistrationEventQueue.class);

     when(clientRegistrationMetadata.getClientProxyMembershipID()).thenReturn(
         clientProxyMembershipID);
     when(clientRegistrationEventQueueManager.create(eq(clientProxyMembershipID), any(), any()))
         .thenReturn(clientRegistrationEventQueue);

     CacheClientNotifier cacheClientNotifier = CacheClientNotifier.getInstance(internalCache,
         clientRegistrationEventQueueManager, statisticsClock, cacheServerStats, 0, 0,
         connectionListener, null, false);
     CacheClientNotifier cacheClientNotifierSpy = spy(cacheClientNotifier);

     doAnswer((i) -> {
       throw new RegionQueueException();
     }).when(cacheClientNotifierSpy).registerClientInternal(clientRegistrationMetadata, socket,
         false, 0, true);

     assertThatThrownBy(() -> cacheClientNotifierSpy.registerClient(clientRegistrationMetadata,
         socket, false, 0, true))
             .isInstanceOf(IOException.class);

     verify(clientRegistrationEventQueueManager, times(1)).create(
         eq(clientProxyMembershipID), any(), any());

     verify(clientRegistrationEventQueueManager, times(1)).drain(
         eq(clientRegistrationEventQueue),
         eq(cacheClientNotifierSpy));
   }

   private InternalCacheEvent createMockInternalCacheEvent(
       final ClientProxyMembershipID clientProxyMembershipID,
       final ClientUpdateMessageImpl clientUpdateMessage,
       final CacheClientNotifier cacheClientNotifierSpy) {
     InternalCacheEvent internalCacheEvent = mock(InternalCacheEvent.class);

     DistributedRegion region = mock(DistributedRegion.class);
     when(internalCacheEvent.getRegion()).thenReturn(region);

     FilterRoutingInfo.FilterInfo filterInfo = mock(FilterRoutingInfo.FilterInfo.class);
     final Long cqId = 0L;
     final String cqName = "testCQ";

     HashMap cqs = new HashMap<Long, Integer>() {
       {
         put(cqId, 123);
       }
     };
     when(filterInfo.getCQs()).thenReturn(cqs);
     when(internalCacheEvent.getLocalFilterInfo()).thenReturn(
         filterInfo);
     when(internalCacheEvent.getOperation()).thenReturn(mock(Operation.class));

     FilterProfile filterProfile = mock(FilterProfile.class);
     when(filterProfile.getRealCqID(cqId)).thenReturn(cqName);

     ServerCQ serverCQ = mock(ServerCQ.class);
     when(serverCQ.getClientProxyId()).thenReturn(clientProxyMembershipID);
     when(filterProfile.getCq(cqName)).thenReturn(serverCQ);
     when(region.getFilterProfile()).thenReturn(filterProfile);

     FilterRoutingInfo filterRoutingInfo = mock(FilterRoutingInfo.class);
     when(filterRoutingInfo.getLocalFilterInfo()).thenReturn(filterInfo);
     when(filterProfile.getFilterRoutingInfoPart2(null, internalCacheEvent))
         .thenReturn(filterRoutingInfo);
     doReturn(clientUpdateMessage).when(cacheClientNotifierSpy)
         .constructClientMessage(internalCacheEvent);

     return internalCacheEvent;
   }

   @Test
   public void testSingletonHasClientProxiesFalseNoCCN() {
     assertFalse(CacheClientNotifier.singletonHasClientProxies());
   }

   @Test
   public void testSingletonHasClientProxiesFalseNoProxy() {
     InternalCache internalCache = Fakes.cache();

     CacheClientNotifier ccn =
         CacheClientNotifier.getInstance(internalCache,
             mock(ClientRegistrationEventQueueManager.class),
             mock(StatisticsClock.class),
             mock(CacheServerStats.class), 10, 10, mock(ConnectionListener.class), null, true);

     assertFalse(CacheClientNotifier.singletonHasClientProxies());
     ccn.shutdown(111);

   }

   @Test
   public void testSingletonHasClientProxiesTrue() {
     InternalCache internalCache = Fakes.cache();
     CacheClientProxy proxy = mock(CacheClientProxy.class);

     CacheClientNotifier ccn =
         CacheClientNotifier.getInstance(internalCache,
             mock(ClientRegistrationEventQueueManager.class),
             mock(StatisticsClock.class),
             mock(CacheServerStats.class), 10, 10, mock(ConnectionListener.class), null, true);

     when(proxy.getProxyID()).thenReturn(mock(ClientProxyMembershipID.class));
     ccn.addClientProxy(proxy);

     // check ClientProxy Map is not empty
     assertTrue(CacheClientNotifier.singletonHasClientProxies());

     when(proxy.getAcceptorId()).thenReturn(Long.valueOf(111));
     ccn.shutdown(111);
   }

   @Test
   public void testSingletonHasInitClientProxiesTrue() {
     InternalCache internalCache = Fakes.cache();
     CacheClientProxy proxy = mock(CacheClientProxy.class);

     CacheClientNotifier ccn =
         CacheClientNotifier.getInstance(internalCache,
             mock(ClientRegistrationEventQueueManager.class),
             mock(StatisticsClock.class),
             mock(CacheServerStats.class), 10, 10, mock(ConnectionListener.class), null, true);

     when(proxy.getProxyID()).thenReturn(mock(ClientProxyMembershipID.class));
     ccn.addClientInitProxy(proxy);

     // check InitClientProxy Map is not empty
     assertTrue(CacheClientNotifier.singletonHasClientProxies());

     ccn.addClientProxy(proxy);

     // check ClientProxy Map is not empty
     assertTrue(CacheClientNotifier.singletonHasClientProxies());

     when(proxy.getAcceptorId()).thenReturn(Long.valueOf(111));
     ccn.shutdown(111);
   }

   private void shutdownExistingCacheClientNotifier() {
     CacheClientNotifier cacheClientNotifier = CacheClientNotifier.getInstance();
     if (cacheClientNotifier != null) {
       cacheClientNotifier.shutdown(0);
     }
   }
 }
	/*
	* 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.assertj.core.api.Assertions.assertThatThrownBy;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertTrue;
	import static org.mockito.ArgumentMatchers.any;
	import static org.mockito.ArgumentMatchers.eq;
	import static org.mockito.ArgumentMatchers.isA;
	import static org.mockito.Mockito.doAnswer;
	import static org.mockito.Mockito.doReturn;
	import static org.mockito.Mockito.mock;
	import static org.mockito.Mockito.spy;
	import static org.mockito.Mockito.times;
	import static org.mockito.Mockito.verify;
	import static org.mockito.Mockito.when;

	import java.io.IOException;
	import java.lang.reflect.InvocationTargetException;
	import java.net.Socket;
	import java.util.ArrayList;
	import java.util.HashMap;
	import java.util.List;
	import java.util.concurrent.Callable;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.ExecutionException;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Executors;
	import java.util.concurrent.Future;

	import org.junit.After;
	import org.junit.Before;
	import org.junit.Test;

	import org.apache.geode.cache.Operation;
	import org.apache.geode.cache.query.internal.cq.ServerCQ;
	import org.apache.geode.internal.cache.DistributedRegion;
	import org.apache.geode.internal.cache.FilterProfile;
	import org.apache.geode.internal.cache.FilterRoutingInfo;
	import org.apache.geode.internal.cache.InternalCache;
	import org.apache.geode.internal.cache.InternalCacheEvent;
	import org.apache.geode.internal.cache.RegionQueueException;
	import org.apache.geode.internal.statistics.StatisticsClock;
	import org.apache.geode.test.fake.Fakes;

	public class CacheClientNotifierTest {
	@Before
	public void setup() {
	// Perform cleanup on any singletons received from previous test runs, since the
	// CacheClientNotifier is a static and previous tests may not have cleaned up properly.
	shutdownExistingCacheClientNotifier();
	}

	@After
	public void tearDown() {
	shutdownExistingCacheClientNotifier();
	}

	@Test
	public void eventsInClientRegistrationQueueAreSentToClientAfterRegistrationIsComplete()
	throws IOException, ClassNotFoundException, NoSuchMethodException, IllegalAccessException,
	InvocationTargetException, InterruptedException, ExecutionException {
	InternalCache internalCache = Fakes.cache();
	CacheServerStats cacheServerStats = mock(CacheServerStats.class);
	Socket socket = mock(Socket.class);
	ConnectionListener connectionListener = mock(ConnectionListener.class);
	ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
	CacheClientProxy cacheClientProxy = mock(CacheClientProxy.class);
	ClientUpdateMessageImpl clientUpdateMessage = mock(ClientUpdateMessageImpl.class);
	ClientRegistrationMetadata clientRegistrationMetadata = mock(ClientRegistrationMetadata.class);
	StatisticsClock statisticsClock = mock(StatisticsClock.class);

	when(clientRegistrationMetadata.getClientProxyMembershipID()).thenReturn(
	clientProxyMembershipID);

	CacheClientNotifier cacheClientNotifier = CacheClientNotifier.getInstance(internalCache,
	new ClientRegistrationEventQueueManager(), statisticsClock, cacheServerStats, 0, 0,
	connectionListener, null, false);
	final CacheClientNotifier cacheClientNotifierSpy = spy(cacheClientNotifier);

	CountDownLatch waitForEventDispatchCountdownLatch = new CountDownLatch(1);
	CountDownLatch waitForRegistrationCountdownLatch = new CountDownLatch(1);
	ExecutorService registerAndNotifyExecutor = Executors.newFixedThreadPool(2);

	try {
	// We stub out the CacheClientNotifier.registerClientInternal() to do some "work" until
	// a new event is received and queued, as triggered by the waitForEventDispatchCountdownLatch
	doAnswer((i) -> {
	when(cacheClientProxy.getProxyID()).thenReturn(clientProxyMembershipID);
	cacheClientNotifierSpy.addClientProxy(cacheClientProxy);
	waitForRegistrationCountdownLatch.countDown();
	waitForEventDispatchCountdownLatch.await();
	return null;
	}).when(cacheClientNotifierSpy).registerClientInternal(clientRegistrationMetadata, socket,
	false, 0, true);
	List<Callable<Void>> registerAndNotifyTasks = new ArrayList<>();

	// In one thread, we register the new client which should create the temporary client
	// registration event queue. Events will be passed to that queue while registration is
	// underway. Once registration is complete, the queue is drained and the event is processed
	// as normal.
	registerAndNotifyTasks.add(() -> {
	cacheClientNotifierSpy.registerClient(clientRegistrationMetadata, socket, false, 0, true);
	return null;
	});

	// In a second thread, we mock the arrival of a new event. We want to ensure this event
	// goes into the temporary client registration event queue. To do that, we wait on the
	// waitForRegistrationCountdownLatch until registration is underway and the temp queue is
	// created. Once it is, we process the event and notify clients, which should add the event
	// to the temp queue. Finally, we resume registration and after it is complete, we verify
	// that the event was drained, processed, and "delivered" (note message delivery is mocked
	// and results in a no-op).
	registerAndNotifyTasks.add(() -> {
	try {
	waitForRegistrationCountdownLatch.await();

	InternalCacheEvent internalCacheEvent =
	createMockInternalCacheEvent(clientProxyMembershipID, clientUpdateMessage,
	cacheClientNotifierSpy);

	CacheClientNotifier.notifyClients(internalCacheEvent, clientUpdateMessage);
	} finally {
	// Ensure we always countdown so if the test fails it won't hang due to
	// awaiting on this countdown latch.
	waitForEventDispatchCountdownLatch.countDown();
	}
	return null;
	});

	final List<Future<Void>> futures =
	registerAndNotifyExecutor.invokeAll(registerAndNotifyTasks);

	for (final Future future : futures) {
	future.get();
	}
	} finally {
	// To prevent not cleaning up test resources in case an unexpected exception occurs
	waitForEventDispatchCountdownLatch.countDown();
	waitForRegistrationCountdownLatch.countDown();
	registerAndNotifyExecutor.shutdownNow();
	cacheClientNotifier.shutdown(0);
	}

	verify(cacheClientProxy, times(1)).deliverMessage(isA(HAEventWrapper.class));
	}

	@Test
	public void clientRegistrationFailsQueueStillDrained()
	throws ClassNotFoundException, NoSuchMethodException, InvocationTargetException,
	IllegalAccessException, IOException {
	InternalCache internalCache = Fakes.cache();
	CacheServerStats cacheServerStats = mock(CacheServerStats.class);
	Socket socket = mock(Socket.class);
	ConnectionListener connectionListener = mock(ConnectionListener.class);
	ClientRegistrationMetadata clientRegistrationMetadata = mock(ClientRegistrationMetadata.class);
	StatisticsClock statisticsClock = mock(StatisticsClock.class);
	ClientProxyMembershipID clientProxyMembershipID = mock(ClientProxyMembershipID.class);
	ClientRegistrationEventQueueManager clientRegistrationEventQueueManager =
	mock(ClientRegistrationEventQueueManager.class);
	ClientRegistrationEventQueueManager.ClientRegistrationEventQueue clientRegistrationEventQueue =
	mock(ClientRegistrationEventQueueManager.ClientRegistrationEventQueue.class);

	when(clientRegistrationMetadata.getClientProxyMembershipID()).thenReturn(
	clientProxyMembershipID);
	when(clientRegistrationEventQueueManager.create(eq(clientProxyMembershipID), any(), any()))
	.thenReturn(clientRegistrationEventQueue);

	CacheClientNotifier cacheClientNotifier = CacheClientNotifier.getInstance(internalCache,
	clientRegistrationEventQueueManager, statisticsClock, cacheServerStats, 0, 0,
	connectionListener, null, false);
	CacheClientNotifier cacheClientNotifierSpy = spy(cacheClientNotifier);

	doAnswer((i) -> {
	throw new RegionQueueException();
	}).when(cacheClientNotifierSpy).registerClientInternal(clientRegistrationMetadata, socket,
	false, 0, true);

	assertThatThrownBy(() -> cacheClientNotifierSpy.registerClient(clientRegistrationMetadata,
	socket, false, 0, true))
	.isInstanceOf(IOException.class);

	verify(clientRegistrationEventQueueManager, times(1)).create(
	eq(clientProxyMembershipID), any(), any());

	verify(clientRegistrationEventQueueManager, times(1)).drain(
	eq(clientRegistrationEventQueue),
	eq(cacheClientNotifierSpy));
	}

	private InternalCacheEvent createMockInternalCacheEvent(
	final ClientProxyMembershipID clientProxyMembershipID,
	final ClientUpdateMessageImpl clientUpdateMessage,
	final CacheClientNotifier cacheClientNotifierSpy) {
	InternalCacheEvent internalCacheEvent = mock(InternalCacheEvent.class);

	DistributedRegion region = mock(DistributedRegion.class);
	when(internalCacheEvent.getRegion()).thenReturn(region);

	FilterRoutingInfo.FilterInfo filterInfo = mock(FilterRoutingInfo.FilterInfo.class);
	final Long cqId = 0L;
	final String cqName = "testCQ";

	HashMap cqs = new HashMap<Long, Integer>() {
	{
	put(cqId, 123);
	}
	};
	when(filterInfo.getCQs()).thenReturn(cqs);
	when(internalCacheEvent.getLocalFilterInfo()).thenReturn(
	filterInfo);
	when(internalCacheEvent.getOperation()).thenReturn(mock(Operation.class));

	FilterProfile filterProfile = mock(FilterProfile.class);
	when(filterProfile.getRealCqID(cqId)).thenReturn(cqName);

	ServerCQ serverCQ = mock(ServerCQ.class);
	when(serverCQ.getClientProxyId()).thenReturn(clientProxyMembershipID);
	when(filterProfile.getCq(cqName)).thenReturn(serverCQ);
	when(region.getFilterProfile()).thenReturn(filterProfile);

	FilterRoutingInfo filterRoutingInfo = mock(FilterRoutingInfo.class);
	when(filterRoutingInfo.getLocalFilterInfo()).thenReturn(filterInfo);
	when(filterProfile.getFilterRoutingInfoPart2(null, internalCacheEvent))
	.thenReturn(filterRoutingInfo);
	doReturn(clientUpdateMessage).when(cacheClientNotifierSpy)
	.constructClientMessage(internalCacheEvent);

	return internalCacheEvent;
	}

	@Test
	public void testSingletonHasClientProxiesFalseNoCCN() {
	assertFalse(CacheClientNotifier.singletonHasClientProxies());
	}

	@Test
	public void testSingletonHasClientProxiesFalseNoProxy() {
	InternalCache internalCache = Fakes.cache();

	CacheClientNotifier ccn =
	CacheClientNotifier.getInstance(internalCache,
	mock(ClientRegistrationEventQueueManager.class),
	mock(StatisticsClock.class),
	mock(CacheServerStats.class), 10, 10, mock(ConnectionListener.class), null, true);

	assertFalse(CacheClientNotifier.singletonHasClientProxies());
	ccn.shutdown(111);

	}

	@Test
	public void testSingletonHasClientProxiesTrue() {
	InternalCache internalCache = Fakes.cache();
	CacheClientProxy proxy = mock(CacheClientProxy.class);

	CacheClientNotifier ccn =
	CacheClientNotifier.getInstance(internalCache,
	mock(ClientRegistrationEventQueueManager.class),
	mock(StatisticsClock.class),
	mock(CacheServerStats.class), 10, 10, mock(ConnectionListener.class), null, true);

	when(proxy.getProxyID()).thenReturn(mock(ClientProxyMembershipID.class));
	ccn.addClientProxy(proxy);

	// check ClientProxy Map is not empty
	assertTrue(CacheClientNotifier.singletonHasClientProxies());

	when(proxy.getAcceptorId()).thenReturn(Long.valueOf(111));
	ccn.shutdown(111);
	}

	@Test
	public void testSingletonHasInitClientProxiesTrue() {
	InternalCache internalCache = Fakes.cache();
	CacheClientProxy proxy = mock(CacheClientProxy.class);

	CacheClientNotifier ccn =
	CacheClientNotifier.getInstance(internalCache,
	mock(ClientRegistrationEventQueueManager.class),
	mock(StatisticsClock.class),
	mock(CacheServerStats.class), 10, 10, mock(ConnectionListener.class), null, true);

	when(proxy.getProxyID()).thenReturn(mock(ClientProxyMembershipID.class));
	ccn.addClientInitProxy(proxy);

	// check InitClientProxy Map is not empty
	assertTrue(CacheClientNotifier.singletonHasClientProxies());

	ccn.addClientProxy(proxy);

	// check ClientProxy Map is not empty
	assertTrue(CacheClientNotifier.singletonHasClientProxies());

	when(proxy.getAcceptorId()).thenReturn(Long.valueOf(111));
	ccn.shutdown(111);
	}

	private void shutdownExistingCacheClientNotifier() {
	CacheClientNotifier cacheClientNotifier = CacheClientNotifier.getInstance();
	if (cacheClientNotifier != null) {
	cacheClientNotifier.shutdown(0);
	}
	}
	}