From 91aefe84aa2eb9d30cccee1a4e707dd7ea799ced Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 7 Oct 2014 17:26:27 +0100 Subject: [PATCH] removed Jozsef Bartok --- .../openhft/chronicle/map/NodeDiscovery.java | 53 +++++++++++++------ .../chronicle/map/NodeDiscoveryTest.java | 18 ++----- 2 files changed, 42 insertions(+), 29 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 8ac91c32f..b16c2164e 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -25,19 +25,35 @@ import static java.util.concurrent.TimeUnit.SECONDS; import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES; import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG; import static net.openhft.chronicle.map.Replicators.tcp; +import static net.openhft.chronicle.map.UdpReplicationConfig.simple; /** * @author Rob Austin. */ public class NodeDiscovery { - public ChronicleMap discoverMap(int udpBroadcastPort, final AddressAndPort ourAddressAndPort) throws IOException, InterruptedException { + private final AddressAndPort ourAddressAndPort; + private final UdpReplicationConfig udpConfig; - final AtomicInteger proposedIdentifier = new AtomicInteger(); + public NodeDiscovery(final AddressAndPort ourAddressAndPort) throws UnknownHostException { + this(8123, ourAddressAndPort, Inet4Address.getByName("255.255.255.255")); + } + + public NodeDiscovery(int udpBroadcastPort, final AddressAndPort ourAddressAndPort, InetAddress address) throws UnknownHostException { + this.ourAddressAndPort = ourAddressAndPort; + this.udpConfig = simple(address, udpBroadcastPort); + } + + public NodeDiscovery() throws SocketException, UnknownHostException { + this(new AddressAndPort(net.openhft.chronicle.map.ConcurrentExpiryMap.getDefaultAddress(), + (short) 8123)); + } + + public ChronicleMap discoverMap() throws IOException, InterruptedException { + + final AtomicInteger ourProposedIdentifier = new AtomicInteger(); final AtomicBoolean useAnotherIdentifier = new AtomicBoolean(); - final UdpReplicationConfig udpConfig = UdpReplicationConfig - .simple(Inet4Address.getByName("255.255.255.255"), udpBroadcastPort); final KnownNodes knownNodes = new KnownNodes(); final Set knownHostPorts = new ConcurrentSkipListSet(); @@ -64,14 +80,14 @@ public class NodeDiscovery { proposedIdentifiersWithHost.values()) { if (!proposedIdentifierWithHost.addressAndPort().equals(ourAddressAndPort)) { - int proposedIdentifer = proposedIdentifierWithHost.identifier(); - if (proposedIdentifer != -1) { - knownAndProposedIdentifiers.set(proposedIdentifer, true); + int proposedIdentifier = proposedIdentifierWithHost.identifier(); + if (proposedIdentifier != -1) { + knownAndProposedIdentifiers.set(proposedIdentifier, true); } knownHostPorts.add(proposedIdentifierWithHost.addressAndPort()); - if (proposedIdentifer == proposedIdentifier.get()) + if (proposedIdentifier == ourProposedIdentifier.get()) useAnotherIdentifier.set(true); @@ -99,7 +115,7 @@ public class NodeDiscovery { externalizable.sendBootStrap(ourHostPort); // once the count down latch is trigger we know we go something back from one of the nodes - if (countDownLatch.get().await(i*20, TimeUnit.MILLISECONDS)) + if (countDownLatch.get().await(i * 20, TimeUnit.MILLISECONDS)) break; } @@ -118,7 +134,7 @@ public class NodeDiscovery { for (; ; ) { useAnotherIdentifier.set(false); identifier = proposeRandomUnusedIdentifier(knownAndProposedIdentifiers, isFistTime); - proposedIdentifier.set(identifier); + ourProposedIdentifier.set(identifier); LOG.info("proposing to use identifier=" + identifier); isFistTime = false; @@ -134,7 +150,7 @@ public class NodeDiscovery { externalizable.sendBootStrap(proposedNodes); // once the count down latch is trigger we know we go something back from one of the nodes - if (countDownLatch.get().await(i*20, TimeUnit.MILLISECONDS)) { + if (countDownLatch.get().await(i * 20, TimeUnit.MILLISECONDS)) { if (useAnotherIdentifier.get()) { // given that another node host proposed the same identifier, we will choose a different one. LOG.info("Another node is using identifier=" + identifier + ", " + @@ -144,11 +160,10 @@ public class NodeDiscovery { break OUTER; } } else { - LOG.info("timed-out getting a response from the server so sending another boot-strap " + + LOG.debug("timed-out getting a response from the server so sending another boot-strap " + "message"); } - } LOG.info("looks like we are the only node in the grid, so going to use identifier=" + identifier); @@ -786,8 +801,6 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { proposedIdentifiersWithHost.put(bootstrap.addressAndPort, bootstrap); - // try { - // we've received a bootstrap message so will will now rebroadcast what we know, // after a random delay // Thread.sleep((int) (Math.random() * 9.0)); @@ -1070,6 +1083,16 @@ class ConcurrentExpiryMap inetAddresses = networkInterface.getInetAddresses(); + InetAddress inetAddress = inetAddresses.nextElement(); + + if (inetAddress == null) + throw new IllegalStateException(); + + return inetAddress.getAddress(); } } diff --git a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java index 21d03f521..322592b12 100644 --- a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java +++ b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java @@ -3,27 +3,17 @@ package net.openhft.chronicle.map; import org.junit.Ignore; import org.junit.Test; -import java.net.InetAddress; -import java.net.NetworkInterface; -import java.util.Enumeration; - public class NodeDiscoveryTest { @Test @Ignore public void testDiscoverMap() throws Exception { + + + final NodeDiscovery nodeDiscovery = new NodeDiscovery(); - - NetworkInterface networkInterface = ConcurrentExpiryMap.defaultNetworkInterface(); - Enumeration inetAddresses = networkInterface.getInetAddresses(); - - inetAddresses.nextElement(); - AddressAndPort ourAddressAndPort1 = new AddressAndPort(inetAddresses.nextElement().getAddress(), - (short) 1237); - - final ChronicleMap map = nodeDiscovery.discoverMap(8123, ourAddressAndPort1); - + final ChronicleMap map = nodeDiscovery.discoverMap(); Thread.sleep(1000000); }