| /* |
| * 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.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.test.awaitility.GeodeAwaitility.await; |
| import static org.assertj.core.api.Assertions.assertThat; |
| import static org.assertj.core.api.Assertions.assertThatThrownBy; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertNotEquals; |
| import static org.junit.Assert.assertNotNull; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.DataInputStream; |
| import java.lang.reflect.InvocationTargetException; |
| import java.lang.reflect.Method; |
| import java.util.Arrays; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.GemFireConfigException; |
| import org.apache.geode.GemFireIOException; |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheException; |
| 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.PartitionAttributesFactory; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionAttributes; |
| import org.apache.geode.cache.RegionShortcut; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.client.ClientCache; |
| import org.apache.geode.cache.client.ClientCacheFactory; |
| import org.apache.geode.cache.client.ClientRegionShortcut; |
| import org.apache.geode.cache.client.NoAvailableServersException; |
| 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.client.internal.Connection; |
| import org.apache.geode.cache.client.internal.Op; |
| import org.apache.geode.cache.client.internal.PoolImpl; |
| import org.apache.geode.cache.client.internal.QueueConnectionImpl; |
| import org.apache.geode.cache.client.internal.RegisterInterestTracker; |
| import org.apache.geode.cache.server.CacheServer; |
| import org.apache.geode.cache.util.CacheListenerAdapter; |
| import org.apache.geode.cache30.CacheSerializableRunnable; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.DistributedSystem; |
| import org.apache.geode.distributed.DistributedSystemDisconnectedException; |
| import org.apache.geode.distributed.internal.DistributionConfig; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.AvailablePort; |
| import org.apache.geode.internal.OSProcess; |
| import org.apache.geode.internal.cache.CachePerfStats; |
| import org.apache.geode.internal.cache.CacheServerImpl; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.LocalRegion; |
| import org.apache.geode.internal.cache.ha.ThreadIdentifier; |
| import org.apache.geode.internal.logging.LogService; |
| import org.apache.geode.test.dunit.DistributedTestUtils; |
| import org.apache.geode.test.dunit.Host; |
| import org.apache.geode.test.dunit.IgnoredException; |
| 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.cache.internal.JUnit4CacheTestCase; |
| import org.apache.geode.test.dunit.internal.DUnitLauncher; |
| import org.apache.geode.test.junit.categories.ClientServerTest; |
| import org.apache.geode.test.version.VersionManager; |
| |
| /** |
| * Tests client server corner cases between Region and Pool |
| */ |
| @Category({ClientServerTest.class}) |
| public class ClientServerMiscDUnitTestBase extends JUnit4CacheTestCase { |
| |
| protected static PoolImpl pool = null; |
| |
| protected static Connection conn = null; |
| |
| static Cache static_cache; |
| |
| static int PORT1; |
| |
| private static final String k1 = "k1"; |
| |
| private static final String k2 = "k2"; |
| |
| static final String server_k1 = "server-k1"; |
| |
| static final String server_k2 = "server-k2"; |
| |
| static final String REGION_NAME1 = "ClientServerMiscDUnitTest_region1"; |
| |
| static final String REGION_NAME2 = "ClientServerMiscDUnitTest_region2"; |
| |
| private static final String PR_REGION_NAME = "ClientServerMiscDUnitTest_PRregion"; |
| |
| private static Host host; |
| |
| protected static VM server1; |
| |
| protected static VM server2; |
| |
| private static RegionAttributes attrs; |
| |
| // variables for concurrent map API test |
| Properties props = new Properties(); |
| private final int putRange_1Start = 1; |
| private final int putRange_1End = 5; |
| private final int putRange_2Start = 6; |
| private final int putRange_2End = 10; |
| |
| |
| String testVersion; // version for client caches for backward-compatibility |
| // testing |
| |
| public ClientServerMiscDUnitTestBase() { |
| testVersion = VersionManager.CURRENT_VERSION; |
| } |
| |
| @Override |
| public final void postSetUp() { |
| host = Host.getHost(0); |
| server1 = host.getVM(VersionManager.CURRENT_VERSION, 2); |
| server2 = host.getVM(VersionManager.CURRENT_VERSION, 3); |
| } |
| |
| int initServerCache(boolean notifyBySub) { |
| return initServerCache(notifyBySub, false); |
| } |
| |
| int initServerCache2() { |
| return initServerCache2(false); |
| } |
| |
| private int initServerCache(boolean notifyBySub, boolean isHA) { |
| return initServerCache(notifyBySub, server1, isHA); |
| } |
| |
| private int initServerCache2(boolean isHA) { |
| return initServerCache(true, server2, isHA); |
| } |
| |
| int initServerCache(boolean notifyBySub, VM vm, boolean isHA) { |
| return vm.invoke(() -> createServerCache(notifyBySub, getMaxThreads(), isHA)); |
| } |
| |
| @Test |
| public void testConcurrentOperationsWithDRandPR() { |
| int port1 = initServerCache(true); // vm0 |
| int port2 = initServerCache2(); // vm1 |
| String serverName = NetworkUtils.getServerHostName(); |
| host.getVM(testVersion, 0).invoke(() -> createClientCacheV(serverName, port1)); |
| host.getVM(testVersion, 1).invoke(() -> createClientCacheV(serverName, port2)); |
| LogService.getLogger() |
| .info("Testing concurrent map operations from a client with a distributed region"); |
| concurrentMapTest(host.getVM(testVersion, 0), "/" + REGION_NAME1); |
| // TODO add verification in vm1 |
| LogService.getLogger() |
| .info("Testing concurrent map operations from a client with a partitioned region"); |
| concurrentMapTest(host.getVM(testVersion, 0), "/" + PR_REGION_NAME); |
| // TODO add verification in vm1 |
| } |
| |
| /** |
| * When a client's subscription thread connects to a server it should receive the server's |
| * pingInterval setting. This is used by the client to set a read-timeout in order to avoid |
| * hanging should the server's machine crash. |
| */ |
| @Test |
| public void testClientReceivesPingIntervalSetting() { |
| VM clientVM = Host.getHost(0).getVM(testVersion, 0); |
| |
| final int port = initServerCache(true); |
| final String host = NetworkUtils.getServerHostName(); |
| |
| clientVM.invoke("create client cache and verify", () -> { |
| createClientCacheAndVerifyPingIntervalIsSet(host, port); |
| }); |
| } |
| |
| void createClientCacheAndVerifyPingIntervalIsSet(String host, int port) throws Exception { |
| PoolImpl pool = null; |
| try { |
| Properties props = new Properties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, ""); |
| |
| createCache(props); |
| |
| pool = (PoolImpl) PoolManager.createFactory().addServer(host, port) |
| .setSubscriptionEnabled(true).setReadTimeout(1000) |
| .setSocketBufferSize(32768).setMinConnections(1).setSubscriptionRedundancy(-1) |
| .setPingInterval(2000).create("test pool"); |
| |
| Region<Object, Object> region = cache.createRegionFactory(RegionShortcut.LOCAL) |
| .setPoolName("test pool").create(REGION_NAME1); |
| region.registerInterest(".*"); |
| |
| /** get the subscription connection and verify that it has the correct timeout setting */ |
| QueueConnectionImpl primaryConnection = (QueueConnectionImpl) pool.getPrimaryConnection(); |
| int pingInterval = ((CacheClientUpdater) primaryConnection.getUpdater()) |
| .getServerQueueStatus().getPingInterval(); |
| assertNotEquals(0, pingInterval); |
| assertEquals(CacheClientNotifier.getClientPingInterval(), pingInterval); |
| } finally { |
| cache.close(); |
| } |
| |
| } |
| |
| @Test |
| public void testConcurrentOperationsWithDRandPRandEmptyClient() { |
| int port1 = initServerCache(true); // vm0 |
| int port2 = initServerCache2(); // vm1 |
| String serverName = NetworkUtils.getServerHostName(); |
| host.getVM(testVersion, 0).invoke(() -> createEmptyClientCache(serverName, port1)); |
| host.getVM(testVersion, 1).invoke(() -> createClientCacheV(serverName, port2)); |
| LogService.getLogger() |
| .info("Testing concurrent map operations from a client with a distributed region"); |
| concurrentMapTest(host.getVM(testVersion, 0), "/" + REGION_NAME1); |
| // TODO add verification in vm1 |
| LogService.getLogger() |
| .info("Testing concurrent map operations from a client with a partitioned region"); |
| concurrentMapTest(host.getVM(testVersion, 0), "/" + PR_REGION_NAME); |
| // TODO add verification in vm1 |
| } |
| |
| /** |
| * Do putIfAbsent(), replace(Object, Object), replace(Object, Object, Object), remove(Object, |
| * Object) operations |
| */ |
| private void concurrentMapTest(final VM clientVM, final String rName) { |
| |
| // String exceptionStr = ""; |
| clientVM.invoke(new CacheSerializableRunnable("doConcurrentMapOperations") { |
| @Override |
| public void run2() throws CacheException { |
| Cache cache = getCache(); |
| final Region pr = cache.getRegion(rName); |
| assertNotNull(rName + " not created", pr); |
| boolean isEmpty = pr.getAttributes().getDataPolicy() == DataPolicy.EMPTY; |
| |
| // test successful putIfAbsent |
| for (int i = putRange_1Start; i <= putRange_1End; i++) { |
| Object putResult = pr.putIfAbsent(Integer.toString(i), Integer.toString(i)); |
| assertNull("Expected null, but got " + putResult + " for key " + i, putResult); |
| } |
| int size; |
| if (!isEmpty) { |
| size = pr.size(); |
| assertEquals("Size doesn't return expected value", putRange_1End, size); |
| assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty()); |
| } |
| |
| // test unsuccessful putIfAbsent |
| for (int i = putRange_1Start; i <= putRange_1End; i++) { |
| Object putResult = pr.putIfAbsent(Integer.toString(i), Integer.toString(i + 1)); |
| assertEquals("for i=" + i, Integer.toString(i), putResult); |
| assertEquals("for i=" + i, Integer.toString(i), pr.get(Integer.toString(i))); |
| } |
| if (!isEmpty) { |
| size = pr.size(); |
| assertEquals("Size doesn't return expected value", putRange_1End, size); |
| assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty()); |
| } |
| |
| // test successful replace(key, oldValue, newValue) |
| for (int i = putRange_1Start; i <= putRange_1End; i++) { |
| boolean replaceSucceeded = |
| pr.replace(Integer.toString(i), Integer.toString(i), "replaced" + i); |
| assertTrue("for i=" + i, replaceSucceeded); |
| assertEquals("for i=" + i, "replaced" + i, pr.get(Integer.toString(i))); |
| } |
| if (!isEmpty) { |
| size = pr.size(); |
| assertEquals("Size doesn't return expected value", putRange_1End, size); |
| assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty()); |
| } |
| |
| // test unsuccessful replace(key, oldValue, newValue) |
| for (int i = putRange_1Start; i <= putRange_2End; i++) { |
| boolean replaceSucceeded = pr.replace(Integer.toString(i), Integer.toString(i), // wrong |
| // expected |
| // old |
| // value |
| "not" + i); |
| assertFalse("for i=" + i, replaceSucceeded); |
| assertEquals("for i=" + i, i <= putRange_1End ? "replaced" + i : null, |
| pr.get(Integer.toString(i))); |
| } |
| if (!isEmpty) { |
| size = pr.size(); |
| assertEquals("Size doesn't return expected value", putRange_1End, size); |
| assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty()); |
| } |
| |
| // test successful replace(key, value) |
| for (int i = putRange_1Start; i <= putRange_1End; i++) { |
| Object replaceResult = pr.replace(Integer.toString(i), "twice replaced" + i); |
| assertEquals("for i=" + i, "replaced" + i, replaceResult); |
| assertEquals("for i=" + i, "twice replaced" + i, pr.get(Integer.toString(i))); |
| } |
| if (!isEmpty) { |
| size = pr.size(); |
| assertEquals("Size doesn't return expected value", putRange_1End, size); |
| assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty()); |
| } |
| |
| // test unsuccessful replace(key, value) |
| for (int i = putRange_2Start; i <= putRange_2End; i++) { |
| Object replaceResult = pr.replace(Integer.toString(i), "thrice replaced" + i); |
| assertNull("for i=" + i, replaceResult); |
| assertNull("for i=" + i, pr.get(Integer.toString(i))); |
| } |
| if (!isEmpty) { |
| size = pr.size(); |
| assertEquals("Size doesn't return expected value", putRange_1End, size); |
| assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty()); |
| } |
| |
| // test unsuccessful remove(key, value) |
| for (int i = putRange_1Start; i <= putRange_2End; i++) { |
| boolean removeResult = pr.remove(Integer.toString(i), Integer.toString(-i)); |
| assertFalse("for i=" + i, removeResult); |
| assertEquals("for i=" + i, i <= putRange_1End ? "twice replaced" + i : null, |
| pr.get(Integer.toString(i))); |
| } |
| if (!isEmpty) { |
| size = pr.size(); |
| assertEquals("Size doesn't return expected value", putRange_1End, size); |
| assertFalse("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty()); |
| } |
| |
| // test successful remove(key, value) |
| for (int i = putRange_1Start; i <= putRange_1End; i++) { |
| boolean removeResult = pr.remove(Integer.toString(i), "twice replaced" + i); |
| assertTrue("for i=" + i, removeResult); |
| assertNull("for i=" + i, pr.get(Integer.toString(i))); |
| } |
| if (!isEmpty) { |
| size = pr.size(); |
| assertEquals("Size doesn't return expected value", 0, size); |
| pr.localClear(); |
| assertTrue("isEmpty doesnt return proper state of the PartitionedRegion", pr.isEmpty()); |
| } |
| |
| if (!isEmpty) { |
| // bug #42169 - entry not updated on server when locally destroyed on client |
| String key42169 = "key42169"; |
| pr.put(key42169, "initialValue42169"); |
| pr.localDestroy(key42169); |
| boolean success = pr.replace(key42169, "initialValue42169", "newValue42169"); |
| assertTrue("expected replace to succeed", success); |
| pr.destroy(key42169); |
| pr.put(key42169, "secondRound"); |
| pr.localDestroy(key42169); |
| Object result = pr.putIfAbsent(key42169, null); |
| assertEquals("expected putIfAbsent to fail", result, "secondRound"); |
| pr.destroy(key42169); |
| } |
| |
| if (isEmpty) { |
| String key41265 = "key41265"; |
| boolean success = pr.remove(key41265, null); |
| assertFalse("expected remove to fail because key does not exist", success); |
| } |
| |
| // test null values |
| |
| // putIfAbsent with null value creates invalid entry |
| Object oldValue = pr.putIfAbsent("keyForNull", null); |
| assertNull(oldValue); |
| if (!isEmpty) { |
| assertTrue(pr.containsKey("keyForNull")); |
| assertTrue(!pr.containsValueForKey("keyForNull")); |
| } |
| |
| // replace allows null value for oldValue, meaning invalidated entry |
| assertTrue(pr.replace("keyForNull", null, "no longer invalid")); |
| |
| // replace does not allow null value for new value |
| assertThatThrownBy(() -> pr.replace("keyForNull", "no longer invalid", null)) |
| .isExactlyInstanceOf(NullPointerException.class); |
| |
| // other variant of replace does not allow null value for new value |
| assertThatThrownBy(() -> pr.replace("keyForNull", null)) |
| .isExactlyInstanceOf(NullPointerException.class); |
| |
| // replace with null oldvalue matches invalidated entry |
| pr.putIfAbsent("otherKeyForNull", null); |
| CachePerfStats stats = ((GemFireCacheImpl) pr.getCache()).getCachePerfStats(); |
| |
| Number puts = getNumPuts(stats); |
| boolean success = pr.replace("otherKeyForNull", null, "no longer invalid"); |
| assertTrue(success); |
| |
| Number newputs = getNumPuts(stats); |
| assertEquals("stats not updated properly or replace malfunctioned", newputs.longValue(), |
| puts.longValue() + 1); |
| |
| } |
| |
| public Number getNumPuts(CachePerfStats stats) { |
| Method getPutsMethod = null; |
| try { |
| getPutsMethod = stats.getClass().getMethod("getPuts"); |
| } catch (NoSuchMethodException e) { |
| fail(e.getMessage()); |
| } |
| |
| Number puts = null; |
| try { |
| puts = (Number) getPutsMethod.invoke(stats); |
| } catch (IllegalAccessException e) { |
| fail(e.getMessage()); |
| } catch (InvocationTargetException e) { |
| fail(e.getMessage()); |
| } |
| |
| return puts; |
| } |
| }); |
| } |
| |
| /** |
| * Test two regions: notify by subscription is true. For region1 the interest list is empty , for |
| * region 2 the intetest list is all keys. If an update/create is made on region1 , the client |
| * should not receive any. If the create/update is on region2 , the client should receive the |
| * update. |
| */ |
| @Test |
| public void testForTwoRegionHavingDifferentInterestList() { |
| // start server first |
| PORT1 = initServerCache(true); |
| int serverPort = PORT1; |
| VM client1 = Host.getHost(0).getVM(testVersion, 1); |
| String hostname = NetworkUtils.getServerHostName(); |
| client1.invoke("create client1 cache", () -> { |
| createClientCache(hostname, serverPort); |
| populateCache(); |
| registerInterest(); |
| }); |
| |
| server1.invoke("putting entries in server1", () -> put()); |
| |
| client1.invoke(() -> verifyUpdates()); |
| } |
| |
| /** |
| * Test two regions: notify by subscription is true. Both the regions have registered interest in |
| * all the keys. Now close region1 on the client. The region1 should get removed from the interest |
| * list on CCP at server. Any update on region1 on server should not get pushed to the client. |
| * Ensure that the message related is not added to the client's queue at all ( which is diferent |
| * from not receiving a callbak on the client). If an update on region2 is made on the server , |
| * then client should receive the calback |
| */ |
| @Test |
| public void testForTwoRegionHavingALLKEYSInterest() throws Exception { |
| // start server first |
| PORT1 = initServerCache(true); |
| createClientCache(NetworkUtils.getServerHostName(), PORT1); |
| populateCache(); |
| registerInterestInBothTheRegions(); |
| closeRegion1(); |
| Wait.pause(6000); |
| server1.invoke(() -> ClientServerMiscDUnitTestBase.verifyInterestListOnServer()); |
| server1.invoke(() -> ClientServerMiscDUnitTestBase.put()); |
| // pause(5000); |
| verifyUpdatesOnRegion2(); |
| } |
| |
| /** |
| * Test two regions: notify by subscription is true. Both the regions have registered interest in |
| * all the keys. Close both the regions. When the last region is closed , it should close the |
| * ConnectionProxy on the client , close all the server connection threads on the server & remove |
| * the CacheClientProxy from the CacheClient notifier |
| */ |
| @Test |
| public void testRegionClose() throws Exception { |
| // start server first |
| PORT1 = initServerCache(true); |
| pool = (PoolImpl) createClientCache(NetworkUtils.getServerHostName(), PORT1); |
| populateCache(); |
| registerInterestInBothTheRegions(); |
| closeBothRegions(); |
| |
| assertFalse(pool.isDestroyed()); |
| pool.destroy(); |
| assertTrue(pool.isDestroyed()); |
| server1.invoke(() -> ClientServerMiscDUnitTestBase.verifyNoCacheClientProxyOnServer()); |
| |
| } |
| |
| /** |
| * Test two regions: notify by subscription is true. Both the regions have registered interest in |
| * all the keys. Destroy region1 on the client. It should reach the server , kill the region on |
| * the server , propagate it to the interested clients , but it should keep CacheClient Proxy |
| * alive. Destroy Region2 . It should reach server , close conenction proxy , destroy the region2 |
| * on the server , remove the cache client proxy from the cache client notifier & propagate it to |
| * the clients. Then create third region and verify that no CacheClientProxy is created on server |
| */ |
| @Test |
| public void testCCPDestroyOnLastDestroyRegion() throws Exception { |
| PORT1 = initServerCache(true); |
| PoolImpl pool = |
| (PoolImpl) createClientCache(NetworkUtils.getServerHostName(), PORT1); |
| destroyRegion1(); |
| // pause(5000); |
| server1.invoke( |
| () -> ClientServerMiscDUnitTestBase.verifyCacheClientProxyOnServer(REGION_NAME1)); |
| Connection conn = pool.acquireConnection(); |
| assertNotNull(conn); |
| assertEquals(1, pool.getConnectedServerCount()); |
| assertFalse(pool.isDestroyed()); |
| destroyRegion2(); |
| assertFalse(pool.isDestroyed()); |
| destroyPRRegion(); |
| assertFalse(pool.isDestroyed()); |
| pool.destroy(); |
| assertTrue(pool.isDestroyed()); |
| // pause(5000); |
| server1.invoke(() -> ClientServerMiscDUnitTestBase.verifyNoCacheClientProxyOnServer()); |
| |
| assertThatThrownBy(() -> getCache().createRegion(REGION_NAME2, attrs)) |
| .isExactlyInstanceOf(IllegalStateException.class); |
| } |
| |
| /** |
| * Test two regions:If notify by subscription is false , both the regions should receive |
| * invalidates for the updates on server in their respective regions |
| * |
| */ |
| @Test |
| public void testInvalidatesPropagateOnTwoRegions() throws Exception { |
| // start server first |
| PORT1 = initServerCache(false); |
| createClientCache(NetworkUtils.getServerHostName(), PORT1); |
| registerInterestForInvalidatesInBothTheRegions(); |
| populateCache(); |
| server1.invoke(() -> ClientServerMiscDUnitTestBase.put()); |
| // pause(5000); |
| verifyInvalidatesOnBothRegions(); |
| |
| } |
| |
| /** |
| * Test for bug 43407, where LRU in the client caused an entry to be evicted with DESTROY(), then |
| * the client invalidated the entry and did a get(). After the get() the entry was not seen to be |
| * in the client's cache. This turned out to be expected behavior, but we now have this test to |
| * guarantee that the product behaves as expected. |
| */ |
| @Test |
| public void testGetInClientCreatesEntry() throws Exception { |
| // start server first |
| PORT1 = initServerCache(false); |
| createClientCache(NetworkUtils.getServerHostName(), PORT1); |
| registerInterestForInvalidatesInBothTheRegions(); |
| Region region = static_cache.getRegion(REGION_NAME1); |
| populateCache(); |
| region.put("invalidationKey", "invalidationValue"); |
| |
| region.localDestroy("invalidationKey"); |
| assertThat(region.containsKey("invalidationKey")).isFalse(); |
| |
| region.invalidate("invalidationKey"); |
| assertThat(region.containsKey("invalidationKey")).isTrue(); |
| |
| Object value = region.get("invalidationKey"); |
| assertThat(value).isNull(); |
| assertThat(region.containsKeyOnServer("invalidationKey")).isTrue(); |
| } |
| |
| /** |
| * GEODE-478 - large payloads are rejected by client->server |
| */ |
| @Test |
| public void testLargeMessageIsRejected() throws Exception { |
| PORT1 = initServerCache(false); |
| createClientCache(NetworkUtils.getServerHostName(), PORT1); |
| Region region = static_cache.getRegion(REGION_NAME1); |
| Op operation = new Op() { |
| @Override |
| public Object attempt(Connection cnx) throws Exception { |
| throw new MessageTooLargeException("message is too big"); |
| } |
| }; |
| try { |
| ((LocalRegion) region).getServerProxy().getPool().execute(operation); |
| } catch (GemFireIOException e) { |
| assertTrue(e.getCause() instanceof MessageTooLargeException); |
| return; |
| } |
| fail("expected an exception to be thrown"); |
| } |
| |
| /** |
| * Create cache, create pool, notify-by-subscription=false, create a region and on client and on |
| * server. Do not attach pool to region , populate some entries on region both on client and |
| * server. Update the entries on server the client. The client should not have entry invalidate. |
| */ |
| @Test |
| public void testInvalidatesPropagateOnRegionHavingNoPool() { |
| // start server first |
| PORT1 = initServerCache(false); |
| Properties props = new Properties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, ""); |
| new ClientServerMiscDUnitTestBase().createCache(props); |
| String host = NetworkUtils.getServerHostName(); |
| PoolImpl p = |
| (PoolImpl) PoolManager.createFactory().addServer(host, PORT1).setSubscriptionEnabled(true) |
| .setReadTimeout(1000).setSocketBufferSize(32768) |
| .setMinConnections(3).setSubscriptionRedundancy(-1).setPingInterval(2000) |
| // .setRetryAttempts(5) |
| // .setRetryInterval(2000) |
| .create("testInvalidatesPropagateOnRegionHavingNoPool"); |
| |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| // factory.setPoolName(p.getName()); |
| |
| attrs = factory.create(); |
| final Region region1 = getCache().createRegion(REGION_NAME1, attrs); |
| final Region region2 = getCache().createRegion(REGION_NAME2, attrs); |
| assertNotNull(region1); |
| assertNotNull(region2); |
| pool = p; |
| conn = pool.acquireConnection(); |
| assertNotNull(conn); |
| |
| populateCache(); |
| server1.invoke(() -> ClientServerMiscDUnitTestBase.put()); |
| |
| await().until(() -> { |
| Object val = region1.getEntry(k1).getValue(); |
| return k1.equals(val); |
| }); |
| |
| await().until(() -> { |
| Object val = region1.getEntry(k2).getValue(); |
| return k2.equals(val); |
| }); |
| |
| await().until(() -> { |
| Object val = region2.getEntry(k1).getValue(); |
| return k1.equals(val); |
| }); |
| |
| await().until(() -> { |
| Object val = region2.getEntry(k2).getValue(); |
| return k2.equals(val); |
| }); |
| |
| // assertIndexDetailsEquals(region2.getEntry(k2).getValue(), k2); |
| } |
| |
| /** |
| * Create proxy before cache creation, create cache, create two regions, attach same bridge writer |
| * to both of the regions Region interests AL_KEYS on both the regions, |
| * notify-by-subscription=true . The CCP should have both the regions in interest list. |
| * |
| */ |
| |
| @Test |
| public void testProxyCreationBeforeCacheCreation() { |
| Properties props = new Properties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, ""); |
| DistributedSystem ds = getSystem(props); |
| assertNotNull(ds); |
| ds.disconnect(); |
| ds = getSystem(props); |
| PORT1 = initServerCache(true); |
| String host = NetworkUtils.getServerHostName(); |
| Pool p = PoolManager.createFactory().addServer(host, PORT1).setSubscriptionEnabled(true) |
| .setSubscriptionRedundancy(-1) |
| // .setRetryAttempts(5) |
| .create("testProxyCreationBeforeCacheCreationPool"); |
| |
| Cache cache = getCache(); |
| assertNotNull(cache); |
| |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| factory.setPoolName(p.getName()); |
| |
| RegionAttributes myAttrs = factory.create(); |
| Region region1 = cache.createRegion(REGION_NAME1, myAttrs); |
| Region region2 = cache.createRegion(REGION_NAME2, myAttrs); |
| assertNotNull(region1); |
| assertNotNull(region2); |
| // region1.registerInterest(CacheClientProxy.ALL_KEYS); |
| region2.registerInterest("ALL_KEYS"); |
| Wait.pause(6000); |
| server1.invoke(() -> ClientServerMiscDUnitTestBase.verifyInterestListOnServer()); |
| |
| } |
| |
| /** |
| * |
| * bug 35380: Cycling a DistributedSystem with an initialized pool causes interest registration |
| * NPE |
| * |
| * Test Scenario: |
| * |
| * Create a DistributedSystem (DS1). Create a pool, initialize (creates a proxy with DS1 memberid) |
| * Disconnect DS1. Create a DistributedSystem (DS2). Create a Region with pool, it attempts to |
| * register interest using DS2 memberid, gets NPE. |
| * |
| */ |
| @Test |
| public void testSystemCanBeCycledWithAnInitializedPool() { |
| // work around GEODE-477 |
| IgnoredException.addIgnoredException("Connection reset"); |
| Properties props = new Properties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, ""); |
| DistributedSystem ds = getSystem(props); |
| assertNotNull(ds); |
| |
| PORT1 = initServerCache(true); |
| String host = NetworkUtils.getServerHostName(); |
| Pool p = PoolManager.createFactory().addServer(host, PORT1).setSubscriptionEnabled(true) |
| .setSubscriptionRedundancy(-1) |
| // .setRetryAttempts(5) |
| .create("testBug35380Pool"); |
| |
| Cache cache = getCache(); |
| |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| factory.setPoolName(p.getName()); |
| |
| RegionAttributes myAttrs = factory.create(); |
| Region region1 = cache.createRegion(REGION_NAME1, myAttrs); |
| Region region2 = cache.createRegion(REGION_NAME2, myAttrs); |
| assertNotNull(region1); |
| assertNotNull(region2); |
| |
| region2.registerInterest("ALL_KEYS"); |
| |
| ds.disconnect(); |
| Properties prop = new Properties(); |
| prop.setProperty(MCAST_PORT, "0"); |
| prop.setProperty(LOCATORS, ""); |
| ds = getSystem(prop); |
| |
| final Cache cacheForLamda = getCache(); |
| assertNotNull(cacheForLamda); |
| |
| AttributesFactory factory1 = new AttributesFactory(); |
| factory1.setScope(Scope.DISTRIBUTED_ACK); |
| // reuse writer from prev DS |
| factory1.setPoolName(p.getName()); |
| |
| final RegionAttributes attrs1 = factory1.create(); |
| assertThatThrownBy(() -> cacheForLamda.createRegion(REGION_NAME1, attrs1)) |
| .isInstanceOfAny(IllegalStateException.class, DistributedSystemDisconnectedException.class); |
| } |
| |
| @Test(expected = GemFireConfigException.class) |
| public void clientIsPreventedFromConnectingToLocatorAsServer() { |
| IgnoredException.addIgnoredException("Improperly configured client detected"); |
| ClientCacheFactory clientCacheFactory = new ClientCacheFactory(); |
| clientCacheFactory.addPoolServer("localhost", DistributedTestUtils.getDUnitLocatorPort()); |
| clientCacheFactory.setPoolSubscriptionEnabled(true); |
| getClientCache(clientCacheFactory); |
| Region region = ((ClientCache) cache).createClientRegionFactory(ClientRegionShortcut.PROXY) |
| .create(REGION_NAME1); |
| region.registerInterest(k1); |
| } |
| |
| |
| private void createCache(Properties props) { |
| createCacheV(props); |
| } |
| |
| private Cache createCacheV(Properties props) { |
| DistributedSystem ds = getSystem(props); |
| assertNotNull(ds); |
| ds.disconnect(); |
| ds = getSystem(props); |
| Cache cache = getCache(); |
| assertNotNull(cache); |
| return cache; |
| } |
| |
| public static void createClientCacheV(String h, int port) { |
| _createClientCache(h, false, -1, port); |
| } |
| |
| public static void createEmptyClientCache(String h, int... ports) { |
| _createClientCache(h, false, -1, ports); |
| } |
| |
| public static Pool createClientCache(String h, int... ports) { |
| return _createClientCache(h, false, -1, ports); |
| } |
| |
| public static Pool createClientCache(String h, int subscriptionAckInterval, boolean empty, |
| int... ports) { |
| return _createClientCache(h, empty, subscriptionAckInterval, ports); |
| } |
| |
| private static PoolFactory addServers(PoolFactory factory, String h, int... ports) { |
| for (int port : ports) { |
| factory.addServer(h, port); |
| } |
| return factory; |
| } |
| |
| public static Pool _createClientCache(String h, boolean empty, int subscriptionAckInterval, |
| int... ports) { |
| Properties props = new Properties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, ""); |
| props.setProperty(LOG_LEVEL, DUnitLauncher.logLevel); |
| Cache cache = new ClientServerMiscDUnitTestBase().createCacheV(props); |
| ClientServerMiscDUnitTestBase.static_cache = cache; |
| System.setProperty(DistributionConfig.GEMFIRE_PREFIX + "bridge.disableShufflingOfEndpoints", |
| "true"); |
| PoolImpl p; |
| try { |
| PoolFactory poolFactory = PoolManager.createFactory(); |
| addServers(poolFactory, h, ports).setSubscriptionEnabled(true) |
| .setReadTimeout(5000).setSocketBufferSize(32768).setMinConnections(3) |
| .setSubscriptionRedundancy(1).setPingInterval(2000); |
| // .setRetryAttempts(5) |
| // .setRetryInterval(2000) |
| if (subscriptionAckInterval > 0) { |
| poolFactory.setSubscriptionAckInterval(subscriptionAckInterval); |
| } |
| p = (PoolImpl) poolFactory.create("ClientServerMiscDUnitTestPool"); |
| |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| if (empty) { |
| factory.setDataPolicy(DataPolicy.EMPTY); |
| } |
| factory.setPoolName(p.getName()); |
| |
| attrs = factory.create(); |
| } finally { |
| System.getProperties() |
| .remove(DistributionConfig.GEMFIRE_PREFIX + "bridge.disableShufflingOfEndpoints"); |
| } |
| |
| Region region1 = cache.createRegion(REGION_NAME1, attrs); |
| Region region2 = cache.createRegion(REGION_NAME2, attrs); |
| Region prRegion = cache.createRegion(PR_REGION_NAME, attrs); |
| assertNotNull(region1); |
| assertNotNull(region2); |
| assertNotNull(prRegion); |
| pool = p; |
| |
| await().until(() -> { |
| try { |
| conn = pool.acquireConnection(); |
| return conn != null; |
| } catch (NoAvailableServersException e) { |
| return false; |
| } |
| }); |
| |
| return p; |
| } |
| |
| static class MemberIDVerifier extends CacheListenerAdapter { |
| boolean memberIDNotReceived = true; |
| boolean eventReceived = false; |
| |
| @Override |
| public void afterCreate(EntryEvent event) { |
| eventReceived(event); |
| } |
| |
| @Override |
| public void afterUpdate(EntryEvent event) { |
| eventReceived(event); |
| } |
| |
| private void eventReceived(EntryEvent event) { |
| eventReceived = true; |
| DistributedMember memberID = event.getDistributedMember(); |
| memberIDNotReceived = (memberID == null); |
| // System.out.println("received event " + event); |
| } |
| |
| public void reset() { |
| memberIDNotReceived = true; |
| eventReceived = false; |
| } |
| } |
| |
| public static void dumpPoolIdentifiers() throws Exception { |
| // duplicate events were received, so let's look at the thread identifiers we have |
| PoolImpl pool = (PoolImpl) PoolManager.find("ClientServerMiscDUnitTestPool"); |
| |
| Map seqMap = pool.getThreadIdToSequenceIdMap(); |
| for (Object o : seqMap.keySet()) { |
| ThreadIdentifier tid = (ThreadIdentifier) o; |
| byte[] memberBytes = tid.getMembershipID(); |
| dumpMemberId(tid, memberBytes); |
| } |
| } |
| |
| public static void dumpMemberId(Object holder, byte[] memberBytes) throws Exception { |
| byte[] newBytes = new byte[memberBytes.length + 17]; |
| System.arraycopy(memberBytes, 0, newBytes, 0, memberBytes.length); |
| ByteArrayInputStream bais = new ByteArrayInputStream(newBytes); |
| DataInputStream dataIn = new DataInputStream(bais); |
| InternalDistributedMember memberId = InternalDistributedMember.readEssentialData(dataIn); |
| String sb = "<" + Thread.currentThread().getName() + "> " + holder |
| + " is " + memberId + " byte count = " + memberBytes.length |
| + " bytes = " + Arrays.toString(memberBytes); |
| System.out.println(sb); |
| } |
| |
| |
| public static Integer createServerCache(Boolean notifyBySubscription, Integer maxThreads, |
| boolean isHA) throws Exception { |
| Cache cache = new ClientServerMiscDUnitTestBase().createCacheV(new Properties()); |
| unsetSlowDispatcherFlag(); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.DISTRIBUTED_ACK); |
| factory.setEnableConflation(true); |
| factory.setDataPolicy(DataPolicy.REPLICATE); |
| factory.setConcurrencyChecksEnabled(true); |
| RegionAttributes myAttrs = factory.create(); |
| Region r1 = cache.createRegion(REGION_NAME1, myAttrs); |
| Region r2 = cache.createRegion(REGION_NAME2, myAttrs); |
| factory = new AttributesFactory(); |
| factory.setDataPolicy(DataPolicy.PARTITION); |
| if (isHA) { |
| PartitionAttributesFactory paf = new PartitionAttributesFactory().setRedundantCopies(1); |
| factory.setPartitionAttributes(paf.create()); |
| } |
| RegionAttributes prAttrs = factory.create(); |
| |
| Region pr = cache.createRegion(PR_REGION_NAME, prAttrs); |
| assertNotNull(r1); |
| assertNotNull(r2); |
| assertNotNull(pr); |
| |
| CacheServer server = cache.addCacheServer(); |
| int port = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET); |
| r1.getCache().getDistributedSystem().getLogWriter().info("Starting server on port " + port); |
| server.setPort(port); |
| server.setMaxThreads(maxThreads); |
| server.setNotifyBySubscription(notifyBySubscription); |
| server.start(); |
| r1.getCache().getDistributedSystem().getLogWriter() |
| .info("Started server on port " + server.getPort()); |
| return server.getPort(); |
| |
| } |
| |
| protected int getMaxThreads() { |
| return 0; |
| } |
| |
| public static void registerInterest() { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r); |
| r.registerInterest("ALL_KEYS"); |
| r.getAttributesMutator().addCacheListener(new MemberIDVerifier()); |
| } |
| |
| private static void registerInterestForInvalidatesInBothTheRegions() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| assertNotNull(r1); |
| Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r2); |
| r1.registerInterestForAllKeys(InterestResultPolicy.KEYS, false, false); |
| r2.registerInterestForAllKeys(InterestResultPolicy.KEYS, false, false); |
| } catch (CacheWriterException e) { |
| e.printStackTrace(); |
| fail("Test failed due to CacheWriterException during registerInterestnBothRegions" + e); |
| } |
| } |
| |
| private static void registerInterestInBothTheRegions() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| assertNotNull(r1); |
| Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r2); |
| r1.registerInterestForAllKeys(); |
| r2.registerInterestForAllKeys(); |
| } catch (CacheWriterException e) { |
| e.printStackTrace(); |
| fail("Test failed due to CacheWriterException during registerInterestnBothRegions" + e); |
| } |
| } |
| |
| private static void closeRegion1() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region<String, String> r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| assertNotNull(r1); |
| r1.close(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail("Test failed due to Exception during closeRegion1" + e); |
| } |
| } |
| |
| private static void closeBothRegions() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| Region pr = cache.getRegion(Region.SEPARATOR + PR_REGION_NAME); |
| assertNotNull(r1); |
| assertNotNull(r2); |
| assertNotNull(pr); |
| r1.close(); |
| r2.close(); |
| pr.close(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail("Test failed due to Exception during closeBothRegions" + e); |
| } |
| } |
| |
| private static void destroyRegion1() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| assertNotNull(r1); |
| r1.destroyRegion(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail("Test failed due to Exception during closeBothRegions" + e); |
| } |
| } |
| |
| private static void destroyRegion2() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r2); |
| r2.destroyRegion(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| fail("Test failed due to Exception during closeBothRegions" + e); |
| } |
| } |
| |
| private static void destroyPRRegion() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r2 = cache.getRegion(Region.SEPARATOR + PR_REGION_NAME); |
| assertNotNull(r2); |
| r2.destroyRegion(); |
| } catch (Exception e) { |
| // e.printStackTrace(); |
| fail("Test failed due to Exception during closeBothRegions" + e); |
| } |
| } |
| |
| private static void verifyInterestListOnServer() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| assertEquals("More than one BridgeServer", 1, cache.getCacheServers().size()); |
| CacheServerImpl bs = (CacheServerImpl) cache.getCacheServers().iterator().next(); |
| assertNotNull(bs); |
| assertNotNull(bs.getAcceptor()); |
| assertNotNull(bs.getAcceptor().getCacheClientNotifier()); |
| for (CacheClientProxy ccp : bs.getAcceptor().getCacheClientNotifier().getClientProxies()) { |
| // CCP should not contain region1 |
| Set<String> akr = ccp.cils[RegisterInterestTracker.interestListIndex].regions; |
| assertNotNull(akr); |
| assertTrue(!akr.contains(Region.SEPARATOR + REGION_NAME1)); |
| // CCP should contain region2 |
| assertTrue(akr.contains(Region.SEPARATOR + REGION_NAME2)); |
| assertEquals(1, akr.size()); |
| } |
| } catch (Exception ex) { |
| ex.printStackTrace(); |
| fail("while setting verifyInterestListOnServer " + ex); |
| } |
| } |
| |
| private static void verifyNoCacheClientProxyOnServer() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| assertEquals("More than one BridgeServer", 1, cache.getCacheServers().size()); |
| CacheServerImpl cacheServer = (CacheServerImpl) cache.getCacheServers().iterator().next(); |
| assertNotNull(cacheServer); |
| assertNotNull(cacheServer.getAcceptor()); |
| final CacheClientNotifier ccn = cacheServer.getAcceptor().getCacheClientNotifier(); |
| |
| assertNotNull(ccn); |
| await() |
| .until(() -> ccn.getClientProxies().size() == 0); |
| } catch (Exception ex) { |
| System.out.println("The size of the client proxies != 0"); |
| OSProcess.printStacks(0); |
| throw ex; |
| } |
| } |
| |
| private static void verifyCacheClientProxyOnServer(String regionName) { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| assertNull(cache.getRegion(Region.SEPARATOR + regionName)); |
| verifyCacheClientProxyOnServer(); |
| |
| // assertIndexDetailsEquals(1, |
| // bs.getAcceptor().getCacheClientNotifier().getClientProxies().size()); |
| } catch (Exception ex) { |
| ex.printStackTrace(); |
| fail("while setting verifyNoCacheClientProxyOnServer " + ex); |
| } |
| } |
| |
| private static void verifyCacheClientProxyOnServer() { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| assertEquals("More than one BridgeServer", 1, cache.getCacheServers().size()); |
| CacheServerImpl bs = (CacheServerImpl) cache.getCacheServers().iterator().next(); |
| assertNotNull(bs); |
| assertNotNull(bs.getAcceptor()); |
| final CacheClientNotifier ccn = bs.getAcceptor().getCacheClientNotifier(); |
| |
| assertNotNull(ccn); |
| |
| await() |
| .until(() -> ccn.getClientProxies().size() == 1); |
| } |
| |
| public static void populateCache() { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r1); |
| assertNotNull(r2); |
| |
| if (!r1.containsKey(k1)) |
| r1.create(k1, k1); |
| if (!r1.containsKey(k2)) |
| r1.create(k2, k2); |
| if (!r2.containsKey(k1)) |
| r2.create(k1, k1); |
| if (!r2.containsKey(k2)) |
| r2.create(k2, k2); |
| |
| assertEquals(r1.getEntry(k1).getValue(), k1); |
| assertEquals(r1.getEntry(k2).getValue(), k2); |
| assertEquals(r2.getEntry(k1).getValue(), k1); |
| assertEquals(r2.getEntry(k2).getValue(), k2); |
| } |
| |
| public static void put() { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r1); |
| assertNotNull(r2); |
| |
| r1.put(k1, server_k1); |
| r1.put(k2, server_k2); |
| |
| r2.put(k1, server_k1); |
| r2.put(k2, server_k2); |
| |
| assertEquals(r1.getEntry(k1).getValue(), server_k1); |
| assertEquals(r1.getEntry(k2).getValue(), server_k2); |
| assertEquals(r2.getEntry(k1).getValue(), server_k1); |
| assertEquals(r2.getEntry(k2).getValue(), server_k2); |
| } |
| |
| static void putForClient() { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| if (r2 == null) { |
| r2 = cache.createRegionFactory(RegionShortcut.REPLICATE).create(REGION_NAME2); |
| } |
| |
| r2.put(k1, "client2_k1"); |
| r2.put(k2, "client2_k2"); |
| } |
| |
| private static void verifyUpdates() { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| final Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| final Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r1); |
| assertNotNull(r2); |
| |
| // no interest registered in region1 - it should hold client values, which are |
| // the same as the keys |
| await().until(() -> { |
| Object val = r1.getEntry(k1).getValue(); |
| return k1.equals(val); |
| }); |
| |
| await().until(() -> { |
| Object val = r1.getEntry(k2).getValue(); |
| return k2.equals(val); |
| }); |
| |
| // interest was registered in region2 - it should contain server values |
| await().until(() -> { |
| Object val = r2.getEntry(k1).getValue(); |
| return server_k1.equals(val); |
| }); |
| |
| await().until(() -> { |
| Object val = r2.getEntry(k2).getValue(); |
| return server_k2.equals(val); |
| }); |
| |
| // events should have contained a memberID |
| MemberIDVerifier verifier = (MemberIDVerifier) ((LocalRegion) r2).getCacheListener(); |
| assertTrue("client should have received a listener event", verifier.eventReceived); |
| assertFalse("client received an update but the event had no member id", |
| verifier.memberIDNotReceived); |
| verifier.reset(); |
| |
| } |
| |
| private static void verifyInvalidatesOnBothRegions() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| final Region r1 = cache.getRegion(Region.SEPARATOR + REGION_NAME1); |
| final Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r1); |
| assertNotNull(r2); |
| |
| await() |
| .until(() -> r1.getEntry(k1).getValue() == null); |
| |
| await() |
| .until(() -> r1.getEntry(k2).getValue() == null); |
| |
| await() |
| .until(() -> r2.getEntry(k1).getValue() == null); |
| |
| await() |
| .until(() -> r2.getEntry(k2).getValue() == null); |
| |
| } catch (Exception ex) { |
| fail("failed while verifyInvalidatesOnBothRegions()" + ex); |
| } |
| } |
| |
| private static void verifyUpdatesOnRegion2() { |
| try { |
| Cache cache = new ClientServerMiscDUnitTestBase().getCache(); |
| final Region r2 = cache.getRegion(Region.SEPARATOR + REGION_NAME2); |
| assertNotNull(r2); |
| |
| await() |
| .until(() -> server_k1.equals(r2.getEntry(k1).getValue())); |
| |
| await() |
| .until(() -> server_k2.equals(r2.getEntry(k2).getValue())); |
| |
| // assertIndexDetailsEquals(server_k2, r2.getEntry(k2).getValue()); |
| } catch (Exception ex) { |
| fail("failed while verifyUpdatesOnRegion2()" + ex); |
| } |
| } |
| |
| /** |
| * set the boolean for starting the dispatcher thread a bit later to FALSE. This is just a |
| * precaution in case any test set it to true and did not unset it on completion. |
| * |
| */ |
| private static void unsetSlowDispatcherFlag() { |
| CacheClientProxy.isSlowStartForTesting = false; |
| } |
| |
| @Test |
| public void testOnSeverMethodsWithProxyClient() throws Exception { |
| testOnServerMethods(false, false); |
| } |
| |
| @Test |
| public void testOnSeverMethodsWithCachingProxyClient() throws Exception { |
| testOnServerMethods(true, false); |
| } |
| |
| @Test |
| public void testOnSeverMethodsWithProxyClientHA() throws Exception { |
| testOnServerMethods(false, true); |
| } |
| |
| @Test |
| public void testOnSeverMethodsWithCachingProxyClientHA() throws Exception { |
| testOnServerMethods(true, true); |
| } |
| |
| private void testOnServerMethods(boolean isCachingProxy, boolean isHA) throws Exception { |
| int port1 = initServerCache(true, isHA); // vm0 |
| int port2 = initServerCache2(isHA); // vm1 |
| String serverName = NetworkUtils.getServerHostName(); |
| if (isCachingProxy) { |
| createClientCache(serverName, port1, port2); |
| } else { |
| createEmptyClientCache(serverName, port1, port2); |
| } |
| if (isHA) { |
| // add another server for HA scenario |
| initServerCache(true, host.getVM(VersionManager.CURRENT_VERSION, 1), true); |
| } |
| String rName = "/" + REGION_NAME1; |
| String prName = "/" + PR_REGION_NAME; |
| |
| verifyIsEmptyOnServer(rName, true); |
| verifyIsEmptyOnServer(prName, true); |
| |
| int size = 10; |
| putIntoRegion(rName, size, isCachingProxy); |
| verifySizeOnServer(rName, size); |
| |
| putIntoRegion(prName, size, isCachingProxy); |
| if (isHA) { |
| server1.invoke(() -> closeMyCache()); |
| } |
| verifySizeOnServer(prName, size); |
| |
| verifyIsEmptyOnServer(rName, false); |
| verifyIsEmptyOnServer(prName, false); |
| |
| destroyEntries(rName, size); |
| destroyEntries(prName, size); |
| |
| verifyIsEmptyOnServer(rName, true); |
| verifyIsEmptyOnServer(prName, true); |
| verifySizeOnServer(rName, 0); |
| verifySizeOnServer(prName, 0); |
| } |
| |
| private void putIntoRegion(String regionName, int size, boolean isCachingProxy) { |
| Cache cache = getCache(); |
| final Region region = cache.getRegion(regionName); |
| for (int i = 0; i < size; i++) { |
| region.put(i, i); |
| } |
| |
| if (isCachingProxy) { |
| for (int i = 0; i < size; i++) { |
| region.localDestroy(i, i); |
| } |
| } |
| } |
| |
| private void destroyEntries(String regionName, int size) { |
| Cache cache = getCache(); |
| final Region region = cache.getRegion(regionName); |
| |
| for (int i = 0; i < size; i++) { |
| region.destroy(i); |
| } |
| } |
| |
| private void verifySizeOnServer(String regionName, int expectedSize) { |
| Cache cache = getCache(); |
| final Region region = cache.getRegion(regionName); |
| int actualSize = region.sizeOnServer(); |
| assertEquals( |
| "sizeOnServer returns unexpected " + actualSize + " instead of expected " + expectedSize, |
| expectedSize, actualSize); |
| } |
| |
| private void verifyIsEmptyOnServer(String regionName, boolean expected) { |
| Cache cache = getCache(); |
| final Region region = cache.getRegion(regionName); |
| boolean isEmptyOnServer = region.isEmptyOnServer(); |
| assertEquals("isEmptyOnServer returns unexpected " + isEmptyOnServer + " instead of expected " |
| + expected, expected, isEmptyOnServer); |
| } |
| |
| private void closeMyCache() { |
| Cache cache = getCache(); |
| cache.close(); |
| } |
| |
| } |
