| /* |
| * 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.cache30; |
| |
| import static java.util.concurrent.TimeUnit.SECONDS; |
| import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_NETWORK_PARTITION_DETECTION; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.test.awaitility.GeodeAwaitility.await; |
| import static org.apache.geode.test.dunit.Assert.assertEquals; |
| import static org.apache.geode.test.dunit.Assert.assertFalse; |
| import static org.apache.geode.test.dunit.Assert.assertNotNull; |
| import static org.apache.geode.test.dunit.Assert.assertNull; |
| import static org.apache.geode.test.dunit.Assert.assertTrue; |
| import static org.apache.geode.test.dunit.Assert.fail; |
| |
| import java.io.IOException; |
| import java.net.InetAddress; |
| import java.net.InetSocketAddress; |
| import java.net.Socket; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.concurrent.TimeUnit; |
| |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| |
| import org.apache.geode.InternalGemFireException; |
| import org.apache.geode.Statistics; |
| import org.apache.geode.StatisticsType; |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.CacheException; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.Scope; |
| import org.apache.geode.cache.client.ClientCache; |
| import org.apache.geode.cache.client.Pool; |
| import org.apache.geode.cache.client.PoolManager; |
| import org.apache.geode.distributed.ConfigurationProperties; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.DistributedSystem; |
| import org.apache.geode.distributed.DurableClientAttributes; |
| import org.apache.geode.distributed.internal.ServerLocation; |
| import org.apache.geode.internal.cache.tier.InternalClientMembership; |
| import org.apache.geode.internal.cache.tier.sockets.AcceptorImpl; |
| import org.apache.geode.internal.cache.tier.sockets.ServerConnection; |
| import org.apache.geode.management.membership.ClientMembership; |
| import org.apache.geode.management.membership.ClientMembershipEvent; |
| import org.apache.geode.management.membership.ClientMembershipListener; |
| import org.apache.geode.test.dunit.Assert; |
| import org.apache.geode.test.dunit.Host; |
| import org.apache.geode.test.dunit.IgnoredException; |
| import org.apache.geode.test.dunit.Invoke; |
| import org.apache.geode.test.dunit.NetworkUtils; |
| import org.apache.geode.test.dunit.SerializableCallable; |
| import org.apache.geode.test.dunit.SerializableRunnable; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.dunit.Wait; |
| import org.apache.geode.test.junit.categories.ClientServerTest; |
| |
| /** |
| * Tests the ClientMembership API including ClientMembershipListener. |
| * |
| * @since GemFire 4.2.1 |
| */ |
| @Category({ClientServerTest.class}) |
| public class ClientMembershipDUnitTest extends ClientServerTestCase { |
| |
| protected static final boolean CLIENT = true; |
| protected static final boolean SERVER = false; |
| |
| protected static final int JOINED = 0; |
| protected static final int LEFT = 1; |
| protected static final int CRASHED = 2; |
| |
| private static Properties properties; |
| |
| ServerLocation serverLocation = new ServerLocation("127.0.0.1", 0); |
| |
| @Override |
| public final void postTearDownCacheTestCase() throws Exception { |
| Invoke.invokeInEveryVM((ClientMembershipDUnitTest::cleanup)); |
| } |
| |
| public static void cleanup() { |
| properties = null; |
| InternalClientMembership.unregisterAllListeners(); |
| } |
| |
| private void waitForAcceptsInProgressToBe(final int target) throws Exception { |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| int actual = getAcceptsInProgress(); |
| return actual == getAcceptsInProgress(); |
| }); |
| } |
| |
| protected int getAcceptsInProgress() { |
| DistributedSystem distributedSystem = getCache().getDistributedSystem(); |
| StatisticsType st = distributedSystem.findType("CacheServerStats"); |
| Statistics[] s = distributedSystem.findStatisticsByType(st); |
| return s[0].getInt("acceptsInProgress"); |
| } |
| |
| protected static Socket meanSocket; |
| |
| /** |
| * test that a server times out waiting for a handshake that never arrives. |
| */ |
| @Test |
| public void testConnectionTimeout() throws Exception { |
| IgnoredException.addIgnoredException("failed accepting client connection"); |
| final Host host = Host.getHost(0); |
| final String hostName = NetworkUtils.getServerHostName(host); |
| final VM vm0 = host.getVM(0); |
| System.setProperty(AcceptorImpl.ACCEPT_TIMEOUT_PROPERTY_NAME, "1000"); |
| try { |
| final int port = startBridgeServer(0); |
| // AsyncInvocation ai = null; |
| try { |
| assertTrue(port != 0); |
| SerializableRunnable createMeanSocket = |
| new CacheSerializableRunnable("Connect to server with socket") { |
| @Override |
| public void run2() throws CacheException { |
| getCache(); // create a cache so we have stats |
| System.out.println("connecting to cache server with socket"); |
| try { |
| InetAddress addr = InetAddress.getByName(hostName); |
| meanSocket = new Socket(addr, port); |
| } catch (Exception e) { |
| throw new RuntimeException("Test failed to connect or was interrupted", e); |
| } |
| } |
| }; |
| SerializableRunnable closeMeanSocket = new CacheSerializableRunnable("close mean socket") { |
| @Override |
| public void run2() throws CacheException { |
| System.out.println("closing mean socket"); |
| try { |
| meanSocket.close(); |
| } catch (IOException ignore) { |
| } |
| } |
| }; |
| |
| assertEquals(0, getAcceptsInProgress()); |
| |
| System.out.println("creating mean socket"); |
| vm0.invoke("Connect to server with socket", () -> createMeanSocket); |
| try { |
| System.out.println("waiting to see it connect on server"); |
| waitForAcceptsInProgressToBe(1); |
| } finally { |
| System.out.println("closing mean socket"); |
| vm0.invoke("close mean socket", () -> closeMeanSocket); |
| } |
| System.out.println("waiting to see accept to go away on server"); |
| waitForAcceptsInProgressToBe(0); |
| |
| // now try it without a close. Server should timeout the mean connect |
| System.out.println("creating mean socket 2"); |
| vm0.invoke("Connect to server with socket", () -> createMeanSocket); |
| try { |
| System.out.println("waiting to see it connect on server 2"); |
| waitForAcceptsInProgressToBe(1); |
| System.out.println("waiting to see accept to go away on server without us closing"); |
| waitForAcceptsInProgressToBe(0); |
| } finally { |
| System.out.println("closing mean socket 2"); |
| vm0.invoke("close mean socket", () -> closeMeanSocket); |
| } |
| } finally { |
| stopBridgeServers(getCache()); |
| } |
| } finally { |
| System.getProperties().remove(AcceptorImpl.ACCEPT_TIMEOUT_PROPERTY_NAME); |
| } |
| } |
| |
| @Test |
| public void testSynchronousEvents() throws Exception { |
| getSystem(); |
| InternalClientMembership.setForceSynchronous(true); |
| try { |
| doTestBasicEvents(); |
| } finally { |
| InternalClientMembership.setForceSynchronous(false); |
| } |
| } |
| |
| /** |
| * Tests event notification methods on ClientMembership. |
| */ |
| @Test |
| public void testBasicEvents() throws Exception { |
| getSystem(); |
| doTestBasicEvents(); |
| } |
| |
| public void doTestBasicEvents() throws Exception { |
| final boolean[] fired = new boolean[3]; |
| final DistributedMember[] member = new DistributedMember[3]; |
| final String[] memberId = new String[3]; |
| final boolean[] isClient = new boolean[3]; |
| |
| ClientMembershipListener listener = new ClientMembershipListener() { |
| @Override |
| public void memberJoined(ClientMembershipEvent event) { |
| fired[JOINED] = true; |
| member[JOINED] = event.getMember(); |
| memberId[JOINED] = event.getMemberId(); |
| isClient[JOINED] = event.isClient(); |
| } |
| |
| @Override |
| public void memberLeft(ClientMembershipEvent event) { |
| fired[LEFT] = true; |
| member[LEFT] = event.getMember(); |
| memberId[LEFT] = event.getMemberId(); |
| isClient[LEFT] = event.isClient(); |
| } |
| |
| @Override |
| public void memberCrashed(ClientMembershipEvent event) { |
| fired[CRASHED] = true; |
| member[CRASHED] = event.getMember(); |
| memberId[CRASHED] = event.getMemberId(); |
| isClient[CRASHED] = event.isClient(); |
| } |
| }; |
| ClientMembership.registerClientMembershipListener(listener); |
| |
| // test JOIN for server |
| InternalClientMembership.notifyServerJoined(serverLocation); |
| |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED]; |
| }); |
| |
| assertTrue(fired[JOINED]); |
| assertNotNull(member[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // test JOIN for client |
| DistributedMember clientJoined = new TestDistributedMember("clientJoined"); |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED]; |
| }); |
| |
| assertTrue(fired[JOINED]); |
| assertEquals(clientJoined, member[JOINED]); |
| assertEquals(clientJoined.getId(), memberId[JOINED]); |
| assertTrue(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // test LEFT for server |
| InternalClientMembership.notifyServerLeft(serverLocation); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[LEFT]; |
| }); |
| |
| assertFalse(fired[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertTrue(fired[LEFT]); |
| assertNotNull(member[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // test LEFT for client |
| DistributedMember clientLeft = new TestDistributedMember("clientLeft"); |
| InternalClientMembership.notifyClientLeft(clientLeft); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[LEFT]; |
| }); |
| |
| assertFalse(fired[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertTrue(fired[LEFT]); |
| assertEquals(clientLeft, member[LEFT]); |
| assertEquals(clientLeft.getId(), memberId[LEFT]); |
| assertTrue(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // test CRASHED for server |
| InternalClientMembership.notifyServerCrashed(serverLocation); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[CRASHED]; |
| }); |
| |
| assertFalse(fired[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertTrue(fired[CRASHED]); |
| assertNotNull(member[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // test CRASHED for client |
| DistributedMember clientCrashed = new TestDistributedMember("clientCrashed"); |
| InternalClientMembership.notifyClientCrashed(clientCrashed); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[CRASHED]; |
| }); |
| |
| assertFalse(fired[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertTrue(fired[CRASHED]); |
| assertEquals(clientCrashed, member[CRASHED]); |
| assertEquals(clientCrashed.getId(), memberId[CRASHED]); |
| assertTrue(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| } |
| |
| /** |
| * Resets all elements of arrays used for listener testing. Boolean values are reset to false. |
| * String values are reset to null. |
| */ |
| private void resetArraysForTesting(boolean[] fired, DistributedMember[] member, String[] memberId, |
| boolean[] isClient) { |
| for (int i = 0; i < fired.length; i++) { |
| fired[i] = false; |
| member[i] = null; |
| memberId[i] = null; |
| isClient[i] = false; |
| } |
| } |
| |
| /** |
| * Tests unregisterClientMembershipListener to ensure that no further events are delivered to |
| * unregistered listeners. |
| */ |
| @Test |
| public void testUnregisterClientMembershipListener() throws Exception { |
| final boolean[] fired = new boolean[1]; |
| final DistributedMember[] member = new DistributedMember[1]; |
| final String[] memberId = new String[1]; |
| final boolean[] isClient = new boolean[1]; |
| |
| getSystem(); |
| |
| ClientMembershipListener listener = new ClientMembershipListener() { |
| @Override |
| public void memberJoined(ClientMembershipEvent event) { |
| fired[0] = true; |
| member[0] = event.getMember(); |
| memberId[0] = event.getMemberId(); |
| isClient[0] = event.isClient(); |
| } |
| |
| @Override |
| public void memberLeft(ClientMembershipEvent event) {} |
| |
| @Override |
| public void memberCrashed(ClientMembershipEvent event) {} |
| }; |
| ClientMembership.registerClientMembershipListener(listener); |
| |
| // fire event to make sure listener is registered |
| DistributedMember clientJoined = new TestDistributedMember("clientJoined"); |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED]; |
| }); |
| |
| assertTrue(fired[0]); |
| assertEquals(clientJoined, member[0]); |
| assertEquals(clientJoined.getId(), memberId[0]); |
| assertTrue(isClient[0]); |
| |
| resetArraysForTesting(fired, member, memberId, isClient); |
| assertFalse(fired[0]); |
| assertNull(memberId[0]); |
| assertFalse(isClient[0]); |
| |
| // unregister and verify listener is not notified |
| ClientMembership.unregisterClientMembershipListener(listener); |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| await().until(() -> { |
| return true; |
| }); |
| |
| assertFalse(fired[0]); |
| assertNull(member[0]); |
| assertNull(memberId[0]); |
| assertFalse(isClient[0]); |
| } |
| |
| @Test |
| public void testMultipleListeners() throws Exception { |
| final int NUM_LISTENERS = 4; |
| final boolean[] fired = new boolean[NUM_LISTENERS]; |
| final DistributedMember[] member = new DistributedMember[NUM_LISTENERS]; |
| final String[] memberId = new String[NUM_LISTENERS]; |
| final boolean[] isClient = new boolean[NUM_LISTENERS]; |
| |
| getSystem(); |
| |
| final ClientMembershipListener[] listeners = new ClientMembershipListener[NUM_LISTENERS]; |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| final int whichListener = i; |
| listeners[i] = new ClientMembershipListener() { |
| @Override |
| public void memberJoined(ClientMembershipEvent event) { |
| assertFalse(fired[whichListener]); |
| assertNull(member[whichListener]); |
| assertNull(memberId[whichListener]); |
| assertFalse(isClient[whichListener]); |
| fired[whichListener] = true; |
| member[whichListener] = event.getMember(); |
| memberId[whichListener] = event.getMemberId(); |
| isClient[whichListener] = event.isClient(); |
| } |
| |
| @Override |
| public void memberLeft(ClientMembershipEvent event) {} |
| |
| @Override |
| public void memberCrashed(ClientMembershipEvent event) {} |
| }; |
| } |
| |
| final DistributedMember clientJoined = new TestDistributedMember("clientJoined"); |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| synchronized (listeners[i]) { |
| listeners[i].wait(20); |
| } |
| assertFalse(fired[i]); |
| assertNull(member[i]); |
| assertNull(memberId[i]); |
| assertFalse(isClient[i]); |
| } |
| |
| // attempt to register same listener twice... 2nd reg should be ignored |
| // failure would cause an assertion failure in memberJoined impl |
| ClientMembership.registerClientMembershipListener(listeners[0]); |
| ClientMembership.registerClientMembershipListener(listeners[0]); |
| |
| ClientMembershipListener[] registeredListeners = |
| ClientMembership.getClientMembershipListeners(); |
| assertEquals(1, registeredListeners.length); |
| assertEquals(listeners[0], registeredListeners[0]); |
| |
| ClientMembership.registerClientMembershipListener(listeners[1]); |
| registeredListeners = ClientMembership.getClientMembershipListeners(); |
| assertEquals(2, registeredListeners.length); |
| assertEquals(listeners[0], registeredListeners[0]); |
| assertEquals(listeners[1], registeredListeners[1]); |
| |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| synchronized (listeners[1]) { |
| if (!fired[1]) { |
| listeners[1].wait(2000); |
| } |
| } |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| if (i < 2) { |
| assertTrue(fired[i]); |
| assertEquals(clientJoined, member[i]); |
| assertEquals(clientJoined.getId(), memberId[i]); |
| assertTrue(isClient[i]); |
| } else { |
| assertFalse(fired[i]); |
| assertNull(member[i]); |
| assertNull(memberId[i]); |
| assertFalse(isClient[i]); |
| } |
| } |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| ClientMembership.unregisterClientMembershipListener(listeners[0]); |
| registeredListeners = ClientMembership.getClientMembershipListeners(); |
| assertEquals(1, registeredListeners.length); |
| assertEquals(listeners[1], registeredListeners[0]); |
| |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| synchronized (listeners[1]) { |
| if (!fired[1]) { |
| listeners[1].wait(2000); |
| } |
| } |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| if (i == 1) { |
| assertTrue(fired[i]); |
| assertEquals(clientJoined, member[i]); |
| assertEquals(clientJoined.getId(), memberId[i]); |
| assertTrue(isClient[i]); |
| } else { |
| assertFalse(fired[i]); |
| assertNull(member[i]); |
| assertNull(memberId[i]); |
| assertFalse(isClient[i]); |
| } |
| } |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| ClientMembership.registerClientMembershipListener(listeners[2]); |
| ClientMembership.registerClientMembershipListener(listeners[3]); |
| registeredListeners = ClientMembership.getClientMembershipListeners(); |
| assertEquals(3, registeredListeners.length); |
| assertEquals(listeners[1], registeredListeners[0]); |
| assertEquals(listeners[2], registeredListeners[1]); |
| assertEquals(listeners[3], registeredListeners[2]); |
| |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| synchronized (listeners[3]) { |
| if (!fired[3]) { |
| listeners[3].wait(2000); |
| } |
| } |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| if (i != 0) { |
| assertTrue(fired[i]); |
| assertEquals(clientJoined, member[i]); |
| assertEquals(clientJoined.getId(), memberId[i]); |
| assertTrue(isClient[i]); |
| } else { |
| assertFalse(fired[i]); |
| assertNull(member[i]); |
| assertNull(memberId[i]); |
| assertFalse(isClient[i]); |
| } |
| } |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| ClientMembership.registerClientMembershipListener(listeners[0]); |
| registeredListeners = ClientMembership.getClientMembershipListeners(); |
| assertEquals(4, registeredListeners.length); |
| assertEquals(listeners[1], registeredListeners[0]); |
| assertEquals(listeners[2], registeredListeners[1]); |
| assertEquals(listeners[3], registeredListeners[2]); |
| assertEquals(listeners[0], registeredListeners[3]); |
| |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| synchronized (listeners[0]) { |
| if (!fired[0]) { |
| listeners[0].wait(2000); |
| } |
| } |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| assertTrue(fired[i]); |
| assertEquals(clientJoined, member[i]); |
| assertEquals(clientJoined.getId(), memberId[i]); |
| assertTrue(isClient[i]); |
| } |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| ClientMembership.unregisterClientMembershipListener(listeners[3]); |
| registeredListeners = ClientMembership.getClientMembershipListeners(); |
| assertEquals(3, registeredListeners.length); |
| assertEquals(listeners[1], registeredListeners[0]); |
| assertEquals(listeners[2], registeredListeners[1]); |
| assertEquals(listeners[0], registeredListeners[2]); |
| |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| synchronized (listeners[0]) { |
| if (!fired[0]) { |
| listeners[0].wait(2000); |
| } |
| } |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| if (i < 3) { |
| assertTrue(fired[i]); |
| assertEquals(clientJoined, member[i]); |
| assertEquals(clientJoined.getId(), memberId[i]); |
| assertTrue(isClient[i]); |
| } else { |
| assertFalse(fired[i]); |
| assertNull(member[i]); |
| assertNull(memberId[i]); |
| assertFalse(isClient[i]); |
| } |
| } |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| ClientMembership.unregisterClientMembershipListener(listeners[2]); |
| registeredListeners = ClientMembership.getClientMembershipListeners(); |
| assertEquals(2, registeredListeners.length); |
| assertEquals(listeners[1], registeredListeners[0]); |
| assertEquals(listeners[0], registeredListeners[1]); |
| |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| synchronized (listeners[0]) { |
| if (!fired[0]) { |
| listeners[0].wait(2000); |
| } |
| } |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| if (i < 2) { |
| assertTrue(fired[i]); |
| assertEquals(clientJoined, member[i]); |
| assertEquals(clientJoined.getId(), memberId[i]); |
| assertTrue(isClient[i]); |
| } else { |
| assertFalse(fired[i]); |
| assertNull(member[i]); |
| assertNull(memberId[i]); |
| assertFalse(isClient[i]); |
| } |
| } |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| ClientMembership.unregisterClientMembershipListener(listeners[1]); |
| ClientMembership.unregisterClientMembershipListener(listeners[0]); |
| registeredListeners = ClientMembership.getClientMembershipListeners(); |
| assertEquals(0, registeredListeners.length); |
| |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| synchronized (listeners[i]) { |
| listeners[i].wait(20); |
| } |
| assertFalse(fired[i]); |
| assertNull(member[i]); |
| assertNull(memberId[i]); |
| assertFalse(isClient[i]); |
| } |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| ClientMembership.registerClientMembershipListener(listeners[1]); |
| registeredListeners = ClientMembership.getClientMembershipListeners(); |
| assertEquals(1, registeredListeners.length); |
| assertEquals(listeners[1], registeredListeners[0]); |
| |
| InternalClientMembership.notifyClientJoined(clientJoined); |
| synchronized (listeners[1]) { |
| if (!fired[1]) { |
| listeners[1].wait(2000); |
| } |
| } |
| for (int i = 0; i < NUM_LISTENERS; i++) { |
| if (i == 1) { |
| assertTrue(fired[i]); |
| assertEquals(clientJoined, member[i]); |
| assertEquals(clientJoined.getId(), memberId[i]); |
| assertTrue(isClient[i]); |
| } else { |
| assertFalse(fired[i]); |
| assertNull(member[i]); |
| assertNull(memberId[i]); |
| assertFalse(isClient[i]); |
| } |
| } |
| } |
| |
| protected static int testClientMembershipEventsInClient_port; |
| |
| private static int getTestClientMembershipEventsInClient_port() { |
| return testClientMembershipEventsInClient_port; |
| } |
| |
| /** |
| * Tests notification of events in client process. Bridge clients detect server joins when the |
| * client connects to the server. If the server crashes or departs gracefully, the client will |
| * detect this as a crash. |
| */ |
| @Test |
| public void testClientMembershipEventsInClient() throws Exception { |
| properties = null; |
| getSystem(); |
| IgnoredException.addIgnoredException("IOException"); |
| final boolean[] fired = new boolean[3]; |
| final DistributedMember[] member = new DistributedMember[3]; |
| final String[] memberId = new String[3]; |
| final boolean[] isClient = new boolean[3]; |
| |
| // create and register ClientMembershipListener in controller vm... |
| ClientMembershipListener listener = new ClientMembershipListener() { |
| @Override |
| public void memberJoined(ClientMembershipEvent event) { |
| System.out.println("[testClientMembershipEventsInClient] memberJoined: " + event); |
| fired[JOINED] = true; |
| member[JOINED] = event.getMember(); |
| memberId[JOINED] = event.getMemberId(); |
| isClient[JOINED] = event.isClient(); |
| } |
| |
| @Override |
| public void memberLeft(ClientMembershipEvent event) { |
| System.out.println("[testClientMembershipEventsInClient] memberLeft: " + event); |
| } |
| |
| @Override |
| public void memberCrashed(ClientMembershipEvent event) { |
| System.out.println("[testClientMembershipEventsInClient] memberCrashed: " + event); |
| fired[CRASHED] = true; |
| member[CRASHED] = event.getMember(); |
| memberId[CRASHED] = event.getMemberId(); |
| isClient[CRASHED] = event.isClient(); |
| } |
| }; |
| ClientMembership.registerClientMembershipListener(listener); |
| |
| final VM vm0 = Host.getHost(0).getVM(0); |
| final String name = getUniqueName(); |
| final int[] ports = new int[1]; |
| |
| // create BridgeServer in vm0... |
| vm0.invoke("create cache server", () -> { |
| try { |
| System.out.println("[testClientMembershipEventsInClient] Create BridgeServer"); |
| getSystem(); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| Region region = createRegion(name, factory.create()); |
| assertNotNull(region); |
| assertNotNull(getRootRegion().getSubregion(name)); |
| testClientMembershipEventsInClient_port = startBridgeServer(0); |
| } catch (IOException e) { |
| getSystem().getLogWriter().fine(new Exception(e)); |
| fail("Failed to start CacheServer: " + e.getMessage()); |
| } |
| }); |
| |
| // gather details for later creation of ConnectionPool... |
| ports[0] = vm0.invoke("getTestClientMembershipEventsInClient_port", |
| ClientMembershipDUnitTest::getTestClientMembershipEventsInClient_port); |
| assertTrue(ports[0] != 0); |
| |
| DistributedMember serverMember = vm0.invoke("get distributed member", |
| () -> getSystem().getDistributedMember()); |
| |
| String serverMemberId = serverMember.toString(); |
| |
| System.out.println("[testClientMembershipEventsInClient] ports[0]=" + ports[0]); |
| System.out.println("[testClientMembershipEventsInClient] serverMember=" + serverMember); |
| System.out.println("[testClientMembershipEventsInClient] serverMemberId=" + serverMemberId); |
| |
| assertFalse(fired[JOINED]); |
| assertNull(member[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| |
| // sanity check... |
| System.out.println("[testClientMembershipEventsInClient] sanity check"); |
| InternalClientMembership.notifyServerJoined(serverLocation); |
| |
| await().timeout(300, SECONDS) |
| .until(() -> fired[JOINED] || fired[CRASHED]); |
| |
| assertTrue(fired[JOINED]); |
| assertNotNull(member[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // create bridge client in controller vm... |
| System.out.println("[testClientMembershipEventsInClient] create bridge client"); |
| Properties config = new Properties(); |
| config.setProperty(MCAST_PORT, "0"); |
| config.setProperty(LOCATORS, ""); |
| config.setProperty(ENABLE_NETWORK_PARTITION_DETECTION, "false"); |
| getSystem(config); |
| |
| try { |
| getCache(); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| ClientServerTestCase.configureConnectionPool(factory, |
| NetworkUtils.getServerHostName(Host.getHost(0)), ports, true, -1, -1, null); |
| createRegion(name, factory.create()); |
| assertNotNull(getRootRegion().getSubregion(name)); |
| } catch (CacheException ex) { |
| Assert.fail("While creating Region on Edge", ex); |
| } |
| |
| await().timeout(300, SECONDS) |
| .until(() -> fired[JOINED] || fired[CRASHED]); |
| |
| System.out.println("[testClientMembershipEventsInClient] assert client detected server join"); |
| |
| // first check the getCurrentServers() result |
| ClientCache clientCache = (ClientCache) getCache(); |
| Set<InetSocketAddress> servers = clientCache.getCurrentServers(); |
| assertTrue(!servers.isEmpty()); |
| |
| // now check listener results |
| assertTrue(fired[JOINED]); |
| assertNotNull(member[JOINED]); |
| assertNotNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| vm0.invoke("Stop BridgeServer", () -> stopBridgeServers(getCache())); |
| |
| await().timeout(300, SECONDS) |
| .until(() -> fired[JOINED] || fired[CRASHED]); |
| |
| System.out |
| .println("[testClientMembershipEventsInClient] assert client detected server departure"); |
| assertFalse(fired[JOINED]); |
| assertNull(member[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertTrue(fired[CRASHED]); |
| assertNotNull(member[CRASHED]); |
| assertNotNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // now test that we redisover the cache server |
| vm0.invoke("Recreate BridgeServer", () -> { |
| try { |
| System.out.println("[testClientMembershipEventsInClient] restarting BridgeServer"); |
| startBridgeServer(ports[0]); |
| } catch (IOException e) { |
| getSystem().getLogWriter().fine(new Exception(e)); |
| fail("Failed to start CacheServer on VM1: " + e.getMessage()); |
| } |
| }); |
| |
| await().timeout(300, SECONDS) |
| .until(() -> fired[JOINED] || fired[CRASHED]); |
| |
| System.out |
| .println("[testClientMembershipEventsInClient] assert client detected server recovery"); |
| assertTrue(fired[JOINED]); |
| assertNotNull(member[JOINED]); |
| assertNotNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| } |
| |
| /** |
| * Tests notification of events in server process. cache servers detect client joins when the |
| * client connects to the server. |
| */ |
| @Test |
| public void testClientMembershipEventsInServer() throws Exception { |
| final boolean[] fired = new boolean[3]; |
| final DistributedMember[] member = new DistributedMember[3]; |
| final String[] memberId = new String[3]; |
| final boolean[] isClient = new boolean[3]; |
| |
| // create and register ClientMembershipListener in controller vm... |
| ClientMembershipListener listener = new ClientMembershipListener() { |
| @Override |
| public void memberJoined(ClientMembershipEvent event) { |
| System.out.println("[testClientMembershipEventsInServer] memberJoined: " + event); |
| fired[JOINED] = true; |
| member[JOINED] = event.getMember(); |
| memberId[JOINED] = event.getMemberId(); |
| isClient[JOINED] = event.isClient(); |
| assertFalse(fired[LEFT] || fired[CRASHED]); |
| } |
| |
| @Override |
| public void memberLeft(ClientMembershipEvent event) { |
| System.out.println("[testClientMembershipEventsInServer] memberLeft: " + event); |
| fired[LEFT] = true; |
| member[LEFT] = event.getMember(); |
| memberId[LEFT] = event.getMemberId(); |
| isClient[LEFT] = event.isClient(); |
| assertFalse(fired[JOINED] || fired[CRASHED]); |
| } |
| |
| @Override |
| public void memberCrashed(ClientMembershipEvent event) { |
| System.out.println("[testClientMembershipEventsInServer] memberCrashed: " + event); |
| fired[CRASHED] = true; |
| member[CRASHED] = event.getMember(); |
| memberId[CRASHED] = event.getMemberId(); |
| isClient[CRASHED] = event.isClient(); |
| assertFalse(fired[JOINED] || fired[LEFT]); |
| } |
| }; |
| ClientMembership.registerClientMembershipListener(listener); |
| |
| final VM vm0 = Host.getHost(0).getVM(0); |
| final String name = getUniqueName(); |
| final int[] ports = new int[1]; |
| |
| // create BridgeServer in controller vm... |
| System.out.println("[testClientMembershipEventsInServer] Create BridgeServer"); |
| getSystem(); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| Region region = createRegion(name, factory.create()); |
| assertNotNull(region); |
| assertNotNull(getRootRegion().getSubregion(name)); |
| |
| ports[0] = startBridgeServer(0); |
| assertTrue(ports[0] != 0); |
| DistributedMember serverMember = getMemberId(); |
| String serverMemberId = serverMember.toString(); |
| |
| System.out.println("[testClientMembershipEventsInServer] ports[0]=" + ports[0]); |
| System.out.println("[testClientMembershipEventsInServer] serverMemberId=" + serverMemberId); |
| System.out.println("[testClientMembershipEventsInServer] serverMember=" + serverMember); |
| |
| assertFalse(fired[JOINED]); |
| assertNull(member[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| |
| // sanity check... |
| System.out.println("[testClientMembershipEventsInServer] sanity check"); |
| DistributedMember test = new TestDistributedMember("test"); |
| InternalClientMembership.notifyClientJoined(test); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED] || fired[LEFT] || fired[CRASHED]; |
| }); |
| |
| assertTrue(fired[JOINED]); |
| assertEquals(test, member[JOINED]); |
| assertEquals(test.getId(), memberId[JOINED]); |
| assertTrue(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| final Host host = Host.getHost(0); |
| SerializableCallable createConnectionPool = new SerializableCallable("Create connectionPool") { |
| @Override |
| public Object call() { |
| System.out.println("[testClientMembershipEventsInServer] create bridge client"); |
| Properties config = new Properties(); |
| config.setProperty(MCAST_PORT, "0"); |
| config.setProperty(LOCATORS, ""); |
| config.setProperty(ENABLE_NETWORK_PARTITION_DETECTION, "false"); |
| properties = config; |
| DistributedSystem s = getSystem(config); |
| AttributesFactory factory = new AttributesFactory(); |
| Pool pool = ClientServerTestCase.configureConnectionPool(factory, |
| NetworkUtils.getServerHostName(host), ports, true, -1, 2, null); |
| createRegion(name, factory.create()); |
| assertNotNull(getRootRegion().getSubregion(name)); |
| assertTrue(s == basicGetSystem()); // see geode-1078 |
| return getMemberId(); |
| } |
| }; |
| |
| // create bridge client in vm0... |
| DistributedMember clientMember = (DistributedMember) vm0.invoke(createConnectionPool); |
| String clientMemberId = clientMember.toString(); |
| |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED] || fired[LEFT] || fired[CRASHED]; |
| }); |
| |
| System.out.println("[testClientMembershipEventsInServer] assert server detected client join"); |
| assertTrue(fired[JOINED]); |
| assertEquals(member[JOINED] + " should equal " + clientMember, clientMember, member[JOINED]); |
| assertEquals(memberId[JOINED] + " should equal " + clientMemberId, clientMemberId, |
| memberId[JOINED]); |
| assertTrue(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| pauseForClientToJoin(); |
| |
| vm0.invoke(new SerializableRunnable("Stop bridge client") { |
| @Override |
| public void run() { |
| System.out.println("[testClientMembershipEventsInServer] Stop bridge client"); |
| getRootRegion().getSubregion(name).close(); |
| Map m = PoolManager.getAll(); |
| for (final Object o : m.values()) { |
| Pool p = (Pool) o; |
| p.destroy(); |
| } |
| } |
| }); |
| |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED] || fired[LEFT] || fired[CRASHED]; |
| }); |
| |
| System.out.println("[testClientMembershipEventsInServer] assert server detected client left"); |
| assertFalse(fired[JOINED]); |
| assertNull(member[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertTrue(fired[LEFT]); |
| assertEquals(clientMember, member[LEFT]); |
| assertEquals(clientMemberId, memberId[LEFT]); |
| assertTrue(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // reconnect bridge client to test for crashed event |
| clientMemberId = vm0.invoke(createConnectionPool).toString(); |
| |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED] || fired[LEFT] || fired[CRASHED]; |
| }); |
| |
| System.out |
| .println("[testClientMembershipEventsInServer] assert server detected client re-join"); |
| assertTrue(fired[JOINED]); |
| assertEquals(clientMember, member[JOINED]); |
| assertEquals(clientMemberId, memberId[JOINED]); |
| assertTrue(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertFalse(fired[CRASHED]); |
| assertNull(member[CRASHED]); |
| assertNull(memberId[CRASHED]); |
| assertFalse(isClient[CRASHED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| pauseForClientToJoin(); |
| |
| ServerConnection.setForceClientCrashEvent(true); |
| try { |
| vm0.invoke(new SerializableRunnable("Stop bridge client") { |
| @Override |
| public void run() { |
| System.out.println("[testClientMembershipEventsInServer] Stop bridge client"); |
| getRootRegion().getSubregion(name).close(); |
| Map m = PoolManager.getAll(); |
| for (final Object o : m.values()) { |
| Pool p = (Pool) o; |
| p.destroy(); |
| } |
| } |
| }); |
| |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED] || fired[LEFT] || fired[CRASHED]; |
| }); |
| |
| System.out |
| .println("[testClientMembershipEventsInServer] assert server detected client crashed"); |
| assertFalse(fired[JOINED]); |
| assertNull(member[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| assertFalse(fired[LEFT]); |
| assertNull(member[LEFT]); |
| assertNull(memberId[LEFT]); |
| assertFalse(isClient[LEFT]); |
| assertTrue(fired[CRASHED]); |
| assertEquals(clientMember, member[CRASHED]); |
| assertEquals(clientMemberId, memberId[CRASHED]); |
| assertTrue(isClient[CRASHED]); |
| } finally { |
| ServerConnection.setForceClientCrashEvent(false); |
| } |
| } |
| |
| /** |
| * The joined event fires when the first client handshake is processed. This pauses long enough to |
| * allow the rest of the client sockets to complete handshaking before making the client leave. |
| * Without doing this subsequent socket handshakes that are processed could fire join events after |
| * departure events and then a departure event again. If you see failures in |
| * testClientMembershipEventsInServer, try increasing this timeout. |
| */ |
| private void pauseForClientToJoin() { |
| Wait.pause(2000); |
| } |
| |
| /** |
| * Tests registration and event notification in conjunction with disconnecting and reconnecting to |
| * DistributedSystem. |
| */ |
| @Test |
| public void testLifecycle() throws Exception { |
| final boolean[] fired = new boolean[3]; |
| final DistributedMember[] member = new DistributedMember[3]; |
| final String[] memberId = new String[3]; |
| final boolean[] isClient = new boolean[3]; |
| |
| // create and register ClientMembershipListener in controller vm... |
| ClientMembershipListener listener = new ClientMembershipListener() { |
| @Override |
| public void memberJoined(ClientMembershipEvent event) { |
| assertFalse(fired[JOINED]); |
| assertNull(member[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| fired[JOINED] = true; |
| member[JOINED] = event.getMember(); |
| memberId[JOINED] = event.getMemberId(); |
| isClient[JOINED] = event.isClient(); |
| } |
| |
| @Override |
| public void memberLeft(ClientMembershipEvent event) {} |
| |
| @Override |
| public void memberCrashed(ClientMembershipEvent event) {} |
| }; |
| ClientMembership.registerClientMembershipListener(listener); |
| |
| // create loner in controller vm... |
| Properties config = new Properties(); |
| config.setProperty(MCAST_PORT, "0"); |
| config.setProperty(LOCATORS, ""); |
| properties = config; |
| getSystem(config); |
| |
| // assert that event is fired while connected |
| InternalClientMembership.notifyServerJoined(serverLocation); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED]; |
| }); |
| assertTrue(fired[JOINED]); |
| assertNotNull(member[JOINED]); |
| assertFalse(isClient[JOINED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // assert that event is NOT fired while disconnected |
| disconnectFromDS(); |
| |
| InternalClientMembership.notifyServerJoined(serverLocation); |
| await().until(() -> { |
| return true; |
| }); |
| |
| assertFalse(fired[JOINED]); |
| assertNull(member[JOINED]); |
| assertNull(memberId[JOINED]); |
| assertFalse(isClient[JOINED]); |
| resetArraysForTesting(fired, member, memberId, isClient); |
| |
| // assert that event is fired again after reconnecting |
| properties = config; |
| assertTrue(getSystem(config).isConnected()); |
| |
| InternalClientMembership.notifyServerJoined(serverLocation); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| return fired[JOINED]; |
| }); |
| |
| assertTrue(fired[JOINED]); |
| assertNotNull(member[JOINED]); |
| assertFalse(isClient[JOINED]); |
| } |
| |
| /** |
| * Starts up server in controller vm and 4 clients, then calls and tests |
| * ClientMembership.getConnectedClients(). |
| */ |
| @Test |
| public void testGetConnectedClients() throws Exception { |
| final String name = getUniqueName(); |
| final int[] ports = new int[1]; |
| |
| IgnoredException.addIgnoredException("ConnectException"); |
| |
| // create BridgeServer in controller vm... |
| System.out.println("[testGetConnectedClients] Create BridgeServer"); |
| getSystem(); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| Region region = createRegion(name, factory.create()); |
| assertNotNull(region); |
| assertNotNull(getRootRegion().getSubregion(name)); |
| |
| ports[0] = startBridgeServer(0); |
| assertTrue(ports[0] != 0); |
| String serverMemberId = getSystem().getDistributedMember().toString(); |
| |
| System.out.println("[testGetConnectedClients] ports[0]=" + ports[0]); |
| System.out.println("[testGetConnectedClients] serverMemberId=" + serverMemberId); |
| |
| final Host host = Host.getHost(0); |
| SerializableCallable createPool = new SerializableCallable("Create connection pool") { |
| @Override |
| public Object call() { |
| System.out.println("[testGetConnectedClients] create bridge client"); |
| properties = new Properties(); |
| properties.setProperty(MCAST_PORT, "0"); |
| properties.setProperty(LOCATORS, ""); |
| getSystem(properties); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| Pool p = ClientServerTestCase.configureConnectionPool(factory, |
| NetworkUtils.getServerHostName(host), ports, true, -1, -1, null); |
| createRegion(name, factory.create()); |
| assertNotNull(getRootRegion().getSubregion(name)); |
| assertTrue(p.getServers().size() > 0); |
| return getMemberId(); |
| } |
| }; |
| |
| // create bridge client in vm0... |
| final String[] clientMemberIdArray = new String[host.getVMCount()]; |
| |
| for (int i = 0; i < host.getVMCount(); i++) { |
| final VM vm = Host.getHost(0).getVM(i); |
| System.out.println("creating pool in vm_" + i); |
| clientMemberIdArray[i] = vm.invoke(createPool).toString(); |
| } |
| Collection clientMemberIds = Arrays.asList(clientMemberIdArray); |
| |
| { |
| final int expectedClientCount = clientMemberIds.size(); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| Map connectedClients = InternalClientMembership.getConnectedClients(false, getCache()); |
| if (connectedClients == null) { |
| return false; |
| } |
| return connectedClients.size() == expectedClientCount; |
| }); |
| } |
| |
| Map connectedClients = InternalClientMembership.getConnectedClients(false, getCache()); |
| assertNotNull(connectedClients); |
| assertEquals(clientMemberIds.size(), connectedClients.size()); |
| System.out |
| .println("connectedClients: " + connectedClients + "; clientMemberIds: " + clientMemberIds); |
| for (final Object o : connectedClients.keySet()) { |
| String connectedClient = (String) o; |
| System.out.println("[testGetConnectedClients] checking for client " + connectedClient); |
| assertTrue(clientMemberIds.contains(connectedClient)); |
| Object[] result = (Object[]) connectedClients.get(connectedClient); |
| System.out.println("[testGetConnectedClients] result: " |
| + (result == null ? "none" : result[0] + "; connections=" + result[1])); |
| } |
| } |
| |
| /** |
| * Starts up 4 server and the controller vm as a client, then calls and tests |
| * ClientMembership.getConnectedServers(). |
| */ |
| @Test |
| public void testGetConnectedServers() throws Exception { |
| final Host host = Host.getHost(0); |
| final String name = getUniqueName(); |
| final int[] ports = new int[host.getVMCount()]; |
| |
| for (int i = 0; i < host.getVMCount(); i++) { |
| final int whichVM = i; |
| final VM vm = Host.getHost(0).getVM(i); |
| vm.invoke("Create cache server", () -> { |
| // create BridgeServer in controller vm... |
| System.out.println("[testGetConnectedServers] Create BridgeServer"); |
| getSystem(); |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| Region region = createRegion(name + "_" + whichVM, factory.create()); |
| assertNotNull(region); |
| assertNotNull(getRootRegion().getSubregion(name + "_" + whichVM)); |
| region.put("KEY-1", "VAL-1"); |
| |
| try { |
| testGetConnectedServers_port = startBridgeServer(0); |
| } catch (IOException e) { |
| org.apache.geode.test.dunit.LogWriterUtils.getLogWriter() |
| .error("startBridgeServer threw IOException", e); |
| fail("startBridgeServer threw IOException " + e.getMessage()); |
| } |
| |
| assertTrue(testGetConnectedServers_port != 0); |
| |
| System.out.println("[testGetConnectedServers] port=" + ports[whichVM]); |
| System.out.println( |
| "[testGetConnectedServers] serverMemberId=" + getSystem().getDistributedMember()); |
| }); |
| ports[whichVM] = vm.invoke("getTestGetConnectedServers_port", |
| ClientMembershipDUnitTest::getTestGetConnectedServers_port); |
| assertTrue(ports[whichVM] != 0); |
| } |
| |
| System.out.println("[testGetConnectedServers] create bridge client"); |
| Properties config = new Properties(); |
| config.setProperty(MCAST_PORT, "0"); |
| config.setProperty(LOCATORS, ""); |
| properties = config; |
| getSystem(config); |
| getCache(); |
| |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| |
| for (int i = 0; i < ports.length; i++) { |
| System.out.println("[testGetConnectedServers] creating connectionpool for " |
| + NetworkUtils.getServerHostName(host) + " " + ports[i]); |
| int[] thisServerPorts = new int[] {ports[i]}; |
| configureConnectionPoolWithNameAndFactory(factory, NetworkUtils.getServerHostName(host), |
| thisServerPorts, false, -1, |
| -1, null, "pooly" + i, PoolManager.createFactory(), -1, -1, -2, |
| -1); |
| Region region = createRegion(name + "_" + i, factory.create()); |
| assertNotNull(getRootRegion().getSubregion(name + "_" + i)); |
| region.get("KEY-1"); |
| } |
| |
| final int expectedVMCount = host.getVMCount(); |
| await().timeout(300, TimeUnit.SECONDS).until(() -> { |
| if (PoolManager.getAll().size() != expectedVMCount) { |
| return false; |
| } |
| Map connectedServers = InternalClientMembership.getConnectedServers(); |
| if (connectedServers == null) { |
| return false; |
| } |
| return connectedServers.size() == expectedVMCount; |
| }); |
| |
| assertEquals(host.getVMCount(), PoolManager.getAll().size()); |
| |
| Map connectedServers = InternalClientMembership.getConnectedServers(); |
| assertNotNull(connectedServers); |
| assertEquals(host.getVMCount(), connectedServers.size()); |
| for (final Object o : connectedServers.keySet()) { |
| String connectedServer = (String) o; |
| System.out.println("[testGetConnectedServers] value for connectedServer: " |
| + connectedServers.get(connectedServer)); |
| } |
| } |
| |
| protected static int testGetConnectedServers_port; |
| |
| private static int getTestGetConnectedServers_port() { |
| return testGetConnectedServers_port; |
| } |
| |
| @Override |
| public Properties getDistributedSystemProperties() { |
| if (properties == null) { |
| properties = new Properties(); |
| properties.put(ConfigurationProperties.ENABLE_NETWORK_PARTITION_DETECTION, "false"); |
| } |
| return properties; |
| } |
| |
| /** |
| * Tests getConnectedClients(boolean onlyClientsNotifiedByThisServer) where |
| * onlyClientsNotifiedByThisServer is true. |
| */ |
| @Test |
| public void testGetNotifiedClients() throws Exception { |
| final Host host = Host.getHost(0); |
| final String name = getUniqueName(); |
| final int[] ports = new int[host.getVMCount()]; |
| |
| for (int i = 0; i < host.getVMCount(); i++) { |
| final int whichVM = i; |
| final VM vm = Host.getHost(0).getVM(i); |
| vm.invoke(new CacheSerializableRunnable("Create cache server") { |
| @Override |
| public void run2() throws CacheException { |
| // create BridgeServer in controller vm... |
| System.out.println("[testGetNotifiedClients] Create BridgeServer"); |
| getSystem(); |
| AttributesFactory factory = new AttributesFactory(); |
| Region region = createRegion(name, factory.create()); |
| assertNotNull(region); |
| assertNotNull(getRootRegion().getSubregion(name)); |
| region.put("KEY-1", "VAL-1"); |
| |
| try { |
| testGetNotifiedClients_port = startBridgeServer(0); |
| } catch (IOException e) { |
| org.apache.geode.test.dunit.LogWriterUtils.getLogWriter() |
| .error("startBridgeServer threw IOException", e); |
| fail("startBridgeServer threw IOException " + e.getMessage()); |
| } |
| |
| assertTrue(testGetNotifiedClients_port != 0); |
| |
| System.out.println("[testGetNotifiedClients] port=" + ports[whichVM]); |
| System.out.println("[testGetNotifiedClients] serverMemberId=" + getMemberId()); |
| } |
| }); |
| ports[whichVM] = vm.invoke("getTestGetNotifiedClients_port", |
| ClientMembershipDUnitTest::getTestGetNotifiedClients_port); |
| assertTrue(ports[whichVM] != 0); |
| } |
| |
| System.out.println("[testGetNotifiedClients] create bridge client"); |
| Properties config = new Properties(); |
| config.setProperty(MCAST_PORT, "0"); |
| config.setProperty(LOCATORS, ""); |
| properties = config; |
| getSystem(); |
| getCache(); |
| |
| AttributesFactory factory = new AttributesFactory(); |
| factory.setScope(Scope.LOCAL); |
| |
| System.out.println("[testGetNotifiedClients] creating connection pool"); |
| ClientServerTestCase.configureConnectionPool(factory, NetworkUtils.getServerHostName(host), |
| ports, true, -1, -1, null); |
| Region region = createRegion(name, factory.create()); |
| assertNotNull(getRootRegion().getSubregion(name)); |
| region.registerInterest("KEY-1"); |
| region.get("KEY-1"); |
| |
| final String clientMemberId = getMemberId().toString(); |
| |
| pauseForClientToJoin(); |
| |
| // assertions go here |
| int[] clientCounts = new int[host.getVMCount()]; |
| |
| // only one server vm will have that client for updating |
| for (int i = 0; i < host.getVMCount(); i++) { |
| final int whichVM = i; |
| final VM vm = Host.getHost(0).getVM(i); |
| vm.invoke("Create cache server", () -> { |
| Map clients = InternalClientMembership.getConnectedClients(true, getCache()); |
| assertNotNull(clients); |
| testGetNotifiedClients_clientCount = clients.size(); |
| // [bruce] this is not a valid assertion - the server may not use |
| // fully qualified host names while clients always use them in |
| // forming their member ID. The test needs to check InetAddresses, |
| // not strings |
| // if (testGetNotifiedClients_clientCount > 0) { |
| // // assert that the clientMemberId matches |
| // assertEquals(clientMemberId, clients.keySet().iterator().next()); |
| // } |
| }); |
| clientCounts[whichVM] = vm.invoke("getTestGetNotifiedClients_clientCount", |
| ClientMembershipDUnitTest::getTestGetNotifiedClients_clientCount); |
| } |
| |
| // only one server should have a notifier for this client... |
| int totalClientCounts = 0; |
| for (final int clientCount : clientCounts) { |
| totalClientCounts += clientCount; |
| } |
| // this assertion fails because the count is 4 |
| // assertIndexDetailsEquals(1, totalClientCounts); |
| } |
| |
| protected static int testGetNotifiedClients_port; |
| |
| private static int getTestGetNotifiedClients_port() { |
| return testGetNotifiedClients_port; |
| } |
| |
| protected static int testGetNotifiedClients_clientCount; |
| |
| private static int getTestGetNotifiedClients_clientCount() { |
| return testGetNotifiedClients_clientCount; |
| } |
| |
| // Simple DistributedMember implementation |
| static class TestDistributedMember implements DistributedMember { |
| |
| private final String host; |
| |
| public TestDistributedMember(String host) { |
| this.host = host; |
| } |
| |
| @Override |
| public String getName() { |
| return ""; |
| } |
| |
| @Override |
| public String getHost() { |
| return host; |
| } |
| |
| @Override |
| public Set getRoles() { |
| return new HashSet(); |
| } |
| |
| @Override |
| public int getProcessId() { |
| return 0; |
| } |
| |
| @Override |
| public String getId() { |
| return host; |
| } |
| |
| @Override |
| public String getUniqueId() { |
| return host; |
| } |
| |
| @Override |
| public int compareTo(DistributedMember o) { |
| if ((o == null) || !(o instanceof TestDistributedMember)) { |
| throw new InternalGemFireException("Invalidly comparing TestDistributedMember to " + o); |
| } |
| |
| TestDistributedMember tds = (TestDistributedMember) o; |
| return getHost().compareTo(tds.getHost()); |
| } |
| |
| public boolean equals(Object obj) { |
| if ((obj == null) || !(obj instanceof TestDistributedMember)) { |
| return false; |
| } |
| return compareTo((TestDistributedMember) obj) == 0; |
| } |
| |
| public int hashCode() { |
| return getHost().hashCode(); |
| } |
| |
| @Override |
| public DurableClientAttributes getDurableClientAttributes() { |
| |
| return null; |
| } |
| |
| @Override |
| public List<String> getGroups() { |
| return Collections.emptyList(); |
| } |
| } |
| } |
