| /* |
| * 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.distributed.ConfigurationProperties.DURABLE_CLIENT_ID; |
| import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_TIMEOUT; |
| 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.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.fail; |
| |
| import java.util.Collections; |
| import java.util.Iterator; |
| import java.util.Properties; |
| import java.util.Set; |
| |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.InterestResultPolicy; |
| import org.apache.geode.cache.Region; |
| 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.cache30.CacheSerializableRunnable; |
| import org.apache.geode.distributed.internal.DistributionConfig; |
| import org.apache.geode.internal.cache.CacheServerImpl; |
| import org.apache.geode.internal.cache.FilterProfile; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.PoolFactoryImpl; |
| 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.Wait; |
| import org.apache.geode.test.dunit.WaitCriterion; |
| import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase; |
| import org.apache.geode.test.junit.categories.ClientSubscriptionTest; |
| |
| /** |
| * We have 2 servers and One client which registers some keys with durable interest and some without |
| * it. We maintain queues on only One server as redundancy level is one. Following 2 tests have the |
| * two TestCase scenarios |
| * |
| * There are two Tests First Test does the follows : // Step 1: Starting the servers // Step 2: |
| * Bring Up the Client // Step 3: Client registers Interests // Step 4: Update Values on the Server |
| * for Keys // Step 5: Verify Updates on the Client // Step 6: Close Cache of the DurableClient // |
| * Step 7: Update the Values // Step 8: Re-start the Client // Step 9: Verify Updates on the Client |
| * // Step 10 : Stop all VMs |
| * |
| * For Test 2 the steps are as follows // Step 1: Starting the servers // Step 2: Bring Up the |
| * Client // Step 3: Client registers Interests // Step 4: Update Values on the Server for Keys // |
| * Step 5: Verify Updates on the Client // Step 6: Close Cache of the DurableClient // Step 7: |
| * Update the Values // Step 8: Re-start the Client // Step 9: Send Client Ready Message // Step 10: |
| * Register all Keys (K1, K2 as Non-Durable. K3, K4 as Durable) // Step 11: Unregister Some Keys |
| * (Here K1, K3) // Step 12: Modify values on the server for all the Keys // Step 13: Check the |
| * values for the ones not unregistered and the Unregistered Keys' Values should be null |
| */ |
| @Category({ClientSubscriptionTest.class}) |
| public class DurableRegistrationDUnitTest extends JUnit4DistributedTestCase { |
| |
| private VM server1VM; |
| |
| private VM server2VM; |
| |
| private VM durableClientVM; |
| |
| private String regionName; |
| |
| private int PORT1; |
| |
| private int PORT2; |
| |
| private static final String K1 = "KEY_STONE1"; |
| |
| private static final String K2 = "KEY_STONE2"; |
| |
| private static final String K3 = "KEY_STONE3"; |
| |
| private static final String K4 = "KEY_STONE4"; |
| |
| public DurableRegistrationDUnitTest() { |
| super(); |
| } |
| |
| @Override |
| public final void postSetUp() throws Exception { |
| Host host = Host.getHost(0); |
| this.server1VM = host.getVM(0); |
| this.server2VM = host.getVM(1); |
| this.durableClientVM = host.getVM(2); |
| regionName = DurableRegistrationDUnitTest.class.getName() + "_region"; |
| CacheServerTestUtil.disableShufflingOfEndpoints(); |
| } |
| |
| @Test |
| public void testSimpleDurableClient() { |
| |
| // Step 1: Starting the servers |
| PORT1 = ((Integer) this.server1VM |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| PORT2 = ((Integer) this.server2VM |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| |
| // Step 2: Bring Up the Client |
| // Start a durable client that is not kept alive on the server when it |
| // stops normally |
| final String durableClientId = getName() + "_client"; |
| |
| final int durableClientTimeout = 600; // keep the client alive for 600 |
| // seconds |
| this.durableClientVM.invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(durableClientVM.getHost()), PORT1, PORT2, true, |
| 0), |
| regionName, getClientDistributedSystemProperties(durableClientId, durableClientTimeout), |
| Boolean.TRUE)); |
| |
| // Send clientReady message |
| this.durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { |
| @Override |
| public void run2() throws CacheException { |
| CacheServerTestUtil.getCache().readyForEvents(); |
| } |
| }); |
| |
| // Step 3: Client registers Interests |
| // KEY_STONE1, KEY_STONE2 are registered as durableKeys & KEY_STONE3, |
| // KEY_STONE4 as non-durableKeys |
| |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K1, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K2, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K3, new Boolean(true))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K4, new Boolean(true))); |
| |
| // Step 4: Update Values on the Server for KEY_STONE1, KEY_STONE2, |
| // KEY_STONE3, KEY_STONE4 |
| |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K1, "Value1")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K2, "Value2")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K3, "Value3")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K4, "Value4")); |
| |
| Wait.pause(1000); |
| // Step 5: Verify Updates on the Client |
| |
| assertEquals("Value1", this.server2VM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| assertEquals("Value1", this.server1VM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| |
| assertEquals("Value1", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| assertEquals("Value2", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K2))); |
| assertEquals("Value3", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K3))); |
| assertEquals("Value4", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K4))); |
| |
| // Step 6: Close Cache of the DurableClient |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.closeCache()); |
| // pause(5000); |
| // Step 7: Update KEY_STONE1,KEY_STONE2,KEY_STONE3,KEY_STONE4 on the |
| // Server say with values PingPong1, PingPong2, PingPong3, PingPong4 |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K1, "PingPong1")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K2, "PingPong2")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K3, "PingPong3")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K4, "PingPong4")); |
| |
| // Step 8: Re-start the Client |
| this.durableClientVM |
| .invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(durableClientVM.getHost()), PORT1, PORT2, |
| true, 0), |
| regionName, getClientDistributedSystemProperties(durableClientId), Boolean.TRUE)); |
| |
| // Send clientReady message |
| this.durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { |
| @Override |
| public void run2() throws CacheException { |
| CacheServerTestUtil.getCache().readyForEvents(); |
| } |
| }); |
| |
| Wait.pause(5000); |
| |
| assertNull(this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| assertNull(this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K2))); |
| |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K1, new Boolean(false))); |
| |
| Wait.pause(5000); |
| assertNull(this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| assertNull(this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K2))); |
| |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K1, "PingPong_updated_1")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K2, "PingPong_updated_2")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K3, "PingPong_updated_3")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K4, "PingPong_updated_4")); |
| |
| Wait.pause(5000); |
| |
| // Step 9: Verify Updates on the Client |
| assertEquals("PingPong_updated_1", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| assertNull(this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K2))); |
| assertEquals("PingPong_updated_3", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K3))); |
| assertEquals("PingPong_updated_4", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K4))); |
| |
| // Step 10 : Stop all VMs |
| // Stop the durable client |
| this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| // Stop server 2 |
| this.server2VM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| // Stop server 1 |
| this.server1VM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| } |
| |
| @Test |
| public void testSimpleDurableClientWithRegistration() { |
| // Step 1: Starting the servers |
| PORT1 = ((Integer) this.server1VM |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| PORT2 = ((Integer) this.server2VM |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| |
| // Step 2: Bring Up the Client |
| // Start a durable client that is not kept alive on the server when it |
| // stops normally |
| final String durableClientId = getName() + "_client"; |
| // keep the client alive for 600 seconds |
| final int durableClientTimeout = 600; |
| this.durableClientVM.invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(durableClientVM.getHost()), PORT1, PORT2, true, |
| 0), |
| regionName, getClientDistributedSystemProperties(durableClientId, durableClientTimeout))); |
| |
| // Send clientReady message |
| this.durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { |
| @Override |
| public void run2() throws CacheException { |
| CacheServerTestUtil.getCache().readyForEvents(); |
| } |
| }); |
| |
| // Step 3: Client registers Interests |
| // KEY_STONE1, KEY_STONE2 are registered as durableKeys & KEY_STONE3, |
| // KEY_STONE4 as non-durableKeys |
| |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K1, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K2, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K3, new Boolean(true))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K4, new Boolean(true))); |
| |
| // Step 4: Update Values on the Server for KEY_STONE1, KEY_STONE2, |
| // KEY_STONE3, KEY_STONE4 |
| |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K1, "Value1")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K2, "Value2")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K3, "Value3")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K4, "Value4")); |
| |
| Wait.pause(1000); |
| // Step 5: Verify Updates on the Client |
| |
| assertEquals("Value1", this.server2VM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| assertEquals("Value1", this.server1VM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| |
| assertEquals("Value1", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| assertEquals("Value2", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K2))); |
| assertEquals("Value3", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K3))); |
| assertEquals("Value4", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K4))); |
| |
| // Step 6: Close Cache of the DurableClient |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.closeCache()); |
| // pause(5000); |
| // Step 7: Update KEY_STONE1,KEY_STONE2,KEY_STONE3,KEY_STONE4 on the |
| // Server say with values PingPong1, PingPong2, PingPong3, PingPong4 |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K1, "PingPong1")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K2, "PingPong2")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K3, "PingPong3")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K4, "PingPong4")); |
| |
| // Step 8: Re-start the Client |
| this.durableClientVM |
| .invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(durableClientVM.getHost()), PORT1, PORT2, |
| true, 0), |
| regionName, getClientDistributedSystemProperties(durableClientId), Boolean.TRUE)); |
| |
| // Step 9: Send clientReady message |
| this.durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { |
| @Override |
| public void run2() throws CacheException { |
| CacheServerTestUtil.getCache().readyForEvents(); |
| } |
| }); |
| |
| // pause(1000); |
| |
| // Step 10: Register all Keys |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K3, new Boolean(true))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K4, new Boolean(true))); |
| |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K1, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K2, new Boolean(false))); |
| |
| // Step 11: Unregister Some Keys (Here K1, K3) |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.unregisterKey(K1)); |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.unregisterKey(K3)); |
| |
| Wait.pause(5000); |
| |
| // Step 12: Modify values on the server for all the Keys |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K1, "PingPong_updated_1")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K2, "PingPong_updated_2")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K3, "PingPong_updated_3")); |
| this.server2VM.invoke(() -> DurableRegistrationDUnitTest.putValue(K4, "PingPong_updated_4")); |
| |
| Wait.pause(5000); |
| |
| // Step 13: Check the values for the ones not unregistered and the |
| // Unregistered Keys' Values should be null |
| try { |
| assertEquals("PingPong_updated_2", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K2))); |
| } catch (Exception e) { |
| fail("Prob in KEY_STONE2: " |
| + this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K2))); |
| } |
| |
| try { |
| assertEquals("PingPong_updated_4", |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K4))); |
| } catch (Exception e) { |
| fail("Prob in KEY_STONE4: " |
| + this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K4))); |
| |
| } |
| |
| try { |
| assertNull(this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| } catch (Exception e) { |
| fail("Prob in KEY_STONE1: " |
| + this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K1))); |
| |
| } |
| |
| try { |
| assertNull(this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K3))); |
| } catch (Exception e) { |
| fail("Prob in KEY_STONE3: " |
| + this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.getValue(K3))); |
| |
| } |
| |
| // Stop the durable client |
| this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| // Stop server 2 |
| this.server2VM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| // Stop server 1 |
| this.server1VM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| } |
| |
| @Test |
| public void testDurableClientWithRegistrationHA() { |
| |
| // Step 1: Start server1 |
| PORT2 = getRandomAvailableTCPPort(); |
| |
| PORT1 = ((Integer) this.server1VM |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| |
| |
| // Step 2: Bring Up the Client |
| final String durableClientId = getName() + "_client"; |
| // keep the client alive for 600 seconds |
| final int durableClientTimeout = 600; |
| this.durableClientVM.invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(durableClientVM.getHost()), PORT1, PORT2, true, |
| 1), |
| regionName, getClientDistributedSystemProperties(durableClientId, durableClientTimeout))); |
| |
| // Send clientReady message |
| this.durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { |
| @Override |
| public void run2() throws CacheException { |
| CacheServerTestUtil.getCache().readyForEvents(); |
| } |
| }); |
| |
| // Step 3: Client registers Interests |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K1, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K2, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K3, new Boolean(true))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K4, new Boolean(true))); |
| |
| // Step 4: Bring up the server2 |
| |
| this.server2VM |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true), PORT2)); |
| |
| Wait.pause(3000); |
| |
| // Check server2 got all the interests registered by the durable client. |
| server2VM.invoke(new CacheSerializableRunnable("Verify Interests.") { |
| @Override |
| public void run2() throws CacheException { |
| LogWriterUtils.getLogWriter() |
| .info("### Verifying interests registered by DurableClient. ###"); |
| CacheClientNotifier ccn = CacheClientNotifier.getInstance(); |
| CacheClientProxy p = null; |
| |
| // Get proxy for the client. |
| for (int i = 0; i < 60; i++) { |
| Iterator ps = ccn.getClientProxies().iterator(); |
| if (!ps.hasNext()) { |
| Wait.pause(1000); |
| continue; |
| } else { |
| p = (CacheClientProxy) ps.next(); |
| break; |
| } |
| } |
| |
| if (p == null) { |
| fail("Proxy initialization taking long time. Increase the wait time."); |
| } |
| |
| Iterator rs = p.getInterestRegisteredRegions().iterator(); |
| String rName = (String) rs.next(); |
| assertNotNull("Null region Name found.", rs); |
| LocalRegion r = (LocalRegion) GemFireCacheImpl.getInstance().getRegion(rName); |
| assertNotNull("Null region found.", r); |
| FilterProfile pf = r.getFilterProfile(); |
| Set intrests = Collections.EMPTY_SET; |
| |
| Set interestKeys = pf.getKeysOfInterest(p.getProxyID().getDurableId()); |
| assertNotNull("durable Interests not found for the proxy", interestKeys); |
| assertEquals("The number of durable keys registered during HARegion GII doesn't match.", |
| interestKeys.size(), 2); |
| interestKeys = pf.getKeysOfInterest(p.getProxyID()); |
| assertNotNull("non-durable Interests not found for the proxy", interestKeys); |
| assertEquals("The number of non-durable keys registered during HARegion GII doesn't match.", |
| interestKeys.size(), 2); |
| } |
| }); |
| |
| |
| // Stop the durable client |
| this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| // Stop server 2 |
| this.server2VM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| // Stop server 1 |
| this.server1VM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| } |
| |
| @Test |
| public void testDurableClientDisConnectWithRegistrationHA() { |
| |
| // Step 1: Start server1 |
| PORT2 = getRandomAvailableTCPPort(); |
| |
| PORT1 = ((Integer) this.server1VM |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true)))) |
| .intValue(); |
| |
| |
| // Step 2: Bring Up the Client |
| final String durableClientId = getName() + "_client"; |
| // keep the client alive for 600 seconds |
| final int durableClientTimeout = 600; |
| this.durableClientVM.invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(durableClientVM.getHost()), PORT1, PORT2, true, |
| 1), |
| regionName, getClientDistributedSystemProperties(durableClientId, durableClientTimeout))); |
| |
| // Send clientReady message |
| this.durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { |
| @Override |
| public void run2() throws CacheException { |
| CacheServerTestUtil.getCache().readyForEvents(); |
| } |
| }); |
| |
| // Step 3: Client registers Interests |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K1, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K2, new Boolean(false))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K3, new Boolean(true))); |
| this.durableClientVM |
| .invoke(() -> DurableRegistrationDUnitTest.registerKey(K4, new Boolean(true))); |
| |
| // Close Cache of the DurableClient |
| this.durableClientVM.invoke(() -> DurableRegistrationDUnitTest.closeCache()); |
| |
| Wait.pause(2000); |
| |
| // Re-start the Client |
| this.durableClientVM |
| .invoke(() -> CacheServerTestUtil.createCacheClient( |
| getClientPool(NetworkUtils.getServerHostName(durableClientVM.getHost()), PORT1, PORT2, |
| true, 1), |
| regionName, getClientDistributedSystemProperties(durableClientId), Boolean.TRUE)); |
| |
| // Send clientReady message |
| this.durableClientVM.invoke(new CacheSerializableRunnable("Send clientReady") { |
| @Override |
| public void run2() throws CacheException { |
| CacheServerTestUtil.getCache().readyForEvents(); |
| } |
| }); |
| |
| // Step 4: Bring up the server2 |
| this.server2VM |
| .invoke(() -> CacheServerTestUtil.createCacheServer(regionName, new Boolean(true), PORT2)); |
| |
| Wait.pause(3000); |
| |
| // Check server2 got all the interests registered by the durable client. |
| server2VM.invoke(new CacheSerializableRunnable("Verify Interests.") { |
| @Override |
| public void run2() throws CacheException { |
| LogWriterUtils.getLogWriter() |
| .info("### Verifying interests registered by DurableClient. ###"); |
| CacheClientNotifier ccn = CacheClientNotifier.getInstance(); |
| CacheClientProxy p = null; |
| |
| // Get proxy for the client. |
| for (int i = 0; i < 60; i++) { |
| Iterator ps = ccn.getClientProxies().iterator(); |
| if (!ps.hasNext()) { |
| Wait.pause(1000); |
| continue; |
| } else { |
| p = (CacheClientProxy) ps.next(); |
| break; |
| } |
| } |
| |
| if (p == null) { |
| fail("Proxy initialization taking long time. Increase the wait time."); |
| } |
| |
| Iterator rs = p.getInterestRegisteredRegions().iterator(); |
| String rName = (String) rs.next(); |
| assertNotNull("Null region Name found.", rs); |
| LocalRegion r = (LocalRegion) GemFireCacheImpl.getInstance().getRegion(rName); |
| assertNotNull("Null region found.", r); |
| FilterProfile pf = r.getFilterProfile(); |
| Set intrests = Collections.EMPTY_SET; |
| |
| Set interestKeys = pf.getKeysOfInterest(p.getProxyID().getDurableId()); |
| assertNotNull("durable Interests not found for the proxy", interestKeys); |
| assertEquals("The number of durable keys registered during HARegion GII doesn't match.", |
| interestKeys.size(), 2); |
| interestKeys = pf.getKeysOfInterest(p.getProxyID()); |
| assertNull("non-durable Interests found for the proxy", interestKeys); |
| } |
| }); |
| |
| |
| // Stop the durable client |
| this.durableClientVM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| // Stop server 2 |
| this.server2VM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| // Stop server 1 |
| this.server1VM.invoke(() -> CacheServerTestUtil.closeCache()); |
| |
| } |
| |
| private static void unregisterAllKeys() { |
| // Get the region |
| Region region = CacheServerTestUtil.getCache() |
| .getRegion(DurableRegistrationDUnitTest.class.getName() + "_region"); |
| // Region region = |
| // CacheServerTestUtil.getCache().getRegion(DurableClientSampleDUnitTest.regionName); |
| assertNotNull(region); |
| region.unregisterInterest(K1); |
| region.unregisterInterest(K2); |
| region.unregisterInterest(K3); |
| region.unregisterInterest(K4); |
| |
| } |
| |
| private static void registerKeys() throws Exception { |
| try { |
| // Get the region |
| Region region = CacheServerTestUtil.getCache() |
| .getRegion(DurableRegistrationDUnitTest.class.getName() + "_region"); |
| // Region region = |
| // CacheServerTestUtil.getCache().getRegion(DurableClientSampleDUnitTest.regionName); |
| assertNotNull(region); |
| |
| region.registerInterest(K1, InterestResultPolicy.KEYS_VALUES, false); |
| region.registerInterest(K2, InterestResultPolicy.KEYS_VALUES, false); |
| region.registerInterest(K3, InterestResultPolicy.KEYS_VALUES, true); |
| region.registerInterest(K4, InterestResultPolicy.KEYS_VALUES, true); |
| |
| assertNotNull(region.getInterestList()); |
| |
| } catch (Exception ex) { |
| Assert.fail("failed while registering interest in registerKey function", ex); |
| } |
| } |
| |
| private static String getValue(String key) { |
| Region r = CacheServerTestUtil.getCache() |
| .getRegion(DurableRegistrationDUnitTest.class.getName() + "_region"); |
| // Region r = CacheServerTestUtil.getCache().getRegion(regionName); |
| assertNotNull(r); |
| // String value = (String)r.get(key); |
| // String value = (String)r.getEntry(key).getValue(); |
| Region.Entry re = r.getEntry(key); |
| |
| if (re == null) { |
| return null; |
| } else { |
| return (String) re.getValue(); |
| } |
| } |
| |
| private static void registerKey(String key, boolean isDurable) throws Exception { |
| try { |
| // Get the region |
| Region region = CacheServerTestUtil.getCache() |
| .getRegion(DurableRegistrationDUnitTest.class.getName() + "_region"); |
| // Region region = |
| // CacheServerTestUtil.getCache().getRegion(regionName); |
| assertNotNull(region); |
| region.registerInterest(key, InterestResultPolicy.NONE, isDurable); |
| } catch (Exception ex) { |
| Assert.fail("failed while registering interest in registerKey function", ex); |
| } |
| } |
| |
| private static void unregisterKey(String key) throws Exception { |
| try { |
| // Get the region |
| Region region = CacheServerTestUtil.getCache() |
| .getRegion(DurableRegistrationDUnitTest.class.getName() + "_region"); |
| // Region region = |
| // CacheServerTestUtil.getCache().getRegion(regionName); |
| assertNotNull(region); |
| region.unregisterInterest(key); |
| } catch (Exception ex) { |
| Assert.fail("failed while registering interest in registerKey function", ex); |
| } |
| } |
| |
| private static void putValue(String key, String value) { |
| try { |
| Region r = CacheServerTestUtil.getCache() |
| .getRegion(DurableRegistrationDUnitTest.class.getName() + "_region"); |
| // Region r = CacheServerTestUtil.getCache().getRegion(regionName); |
| assertNotNull(r); |
| if (r.getEntry(key) != null) { |
| r.put(key, value); |
| } else { |
| r.create(key, value); |
| } |
| assertEquals(value, r.getEntry(key).getValue()); |
| } catch (Exception e) { |
| |
| fail("Put in Server has some fight"); |
| |
| } |
| } |
| |
| private Pool getClientPool(String host, int server1Port, int server2Port, |
| boolean establishCallbackConnection, int redundancyLevel) { |
| PoolFactory pf = PoolManager.createFactory(); |
| pf.addServer(host, server1Port).addServer(host, server2Port) |
| .setSubscriptionEnabled(establishCallbackConnection) |
| .setSubscriptionRedundancy(redundancyLevel); |
| return ((PoolFactoryImpl) pf).getPoolAttributes(); |
| } |
| |
| private Properties getClientDistributedSystemProperties(String durableClientId) { |
| return getClientDistributedSystemProperties(durableClientId, |
| DistributionConfig.DEFAULT_DURABLE_CLIENT_TIMEOUT); |
| } |
| |
| private static void checkNumberOfClientProxies(final int expected) { |
| WaitCriterion ev = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return expected == getNumberOfClientProxies(); |
| } |
| |
| @Override |
| public String description() { |
| return null; |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(ev); |
| } |
| |
| protected static int getNumberOfClientProxies() { |
| return getBridgeServer().getAcceptor().getCacheClientNotifier().getClientProxies().size(); |
| } |
| |
| private Properties getClientDistributedSystemProperties(String durableClientId, |
| int durableClientTimeout) { |
| Properties properties = new Properties(); |
| properties.setProperty(MCAST_PORT, "0"); |
| properties.setProperty(LOCATORS, ""); |
| properties.setProperty(DURABLE_CLIENT_ID, durableClientId); |
| properties.setProperty(DURABLE_CLIENT_TIMEOUT, String.valueOf(durableClientTimeout)); |
| return properties; |
| } |
| |
| private static CacheClientProxy getClientProxy() { |
| // Get the CacheClientNotifier |
| CacheClientNotifier notifier = getBridgeServer().getAcceptor().getCacheClientNotifier(); |
| |
| // Get the CacheClientProxy or not (if proxy set is empty) |
| CacheClientProxy proxy = null; |
| Iterator i = notifier.getClientProxies().iterator(); |
| if (i.hasNext()) { |
| proxy = (CacheClientProxy) i.next(); |
| } |
| return proxy; |
| } |
| |
| private static CacheServerImpl getBridgeServer() { |
| CacheServerImpl bridgeServer = |
| (CacheServerImpl) CacheServerTestUtil.getCache().getCacheServers().iterator().next(); |
| assertNotNull(bridgeServer); |
| return bridgeServer; |
| } |
| |
| public static void closeCache() { |
| Cache cache = CacheServerTestUtil.getCache(); |
| if (cache != null && !cache.isClosed()) { |
| cache.close(true); |
| cache.getDistributedSystem().disconnect(); |
| } |
| } |
| |
| public String getRegionName() { |
| return regionName; |
| } |
| |
| public void setRegionName(String regionName) { |
| this.regionName = regionName; |
| } |
| |
| @Override |
| public final void preTearDown() throws Exception { |
| CacheServerTestUtil.resetDisableShufflingOfEndpointsFlag(); |
| } |
| } |
