diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index fc73d4a4b..c35fa92bc 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -31,8 +31,7 @@ import static net.openhft.chronicle.map.Replicators.tcp; */ public class NodeDiscovery { - - public ChronicleMap discoverMap(int udpBroadcastPort, final int tcpPort) throws IOException, InterruptedException { + public ChronicleMap discoverMap(int udpBroadcastPort, final AddressAndPort ourAddressAndPort) throws IOException, InterruptedException { final AtomicInteger proposedIdentifier = new AtomicInteger(); final AtomicBoolean useAnotherIdentifier = new AtomicBoolean(); @@ -49,11 +48,7 @@ public class NodeDiscovery { final AtomicReference countDownLatch = new AtomicReference(new CountDownLatch(1)); - final AddressAndPort ourAddressAndPort = new AddressAndPort(InetAddress.getLocalHost() - .getAddress(), - (short) tcpPort); - - final UDPEventListener udpEventListener = new UDPEventListener() { + final NodeDiscoveryEventListener nodeDiscoveryEventListener = new NodeDiscoveryEventListener() { @Override public void onRemoteNodeEvent(KnownNodes remoteNode, ConcurrentExpiryMap proposedIdentifiersWithHost) { @@ -80,7 +75,8 @@ public class NodeDiscovery { }; - final DiscoveryNodeBytesMarshallable externalizable = new DiscoveryNodeBytesMarshallable(knownNodes, udpEventListener); + final DiscoveryNodeBytesMarshallable externalizable = new DiscoveryNodeBytesMarshallable + (knownNodes, nodeDiscoveryEventListener, ourAddressAndPort); final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster = new NodeDiscoveryBroadcaster(udpConfig, 1024, externalizable); @@ -133,7 +129,7 @@ public class NodeDiscovery { if (countDownLatch.get().await(50, 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+", " + + LOG.info("Another node is using identifier=" + identifier + ", " + "going to have to select another one."); continue OUTER; } else { @@ -144,7 +140,7 @@ public class NodeDiscovery { } - LOG.info("looks like we are the only node in the grid, so going to use identifier="+identifier); + LOG.info("looks like we are the only node in the grid, so going to use identifier=" + identifier); break; } @@ -169,7 +165,7 @@ public class NodeDiscovery { knownNodes.add(ourAddressAndPort, identifier); final TcpReplicationConfig tcpConfig = TcpReplicationConfig - .of(tcpPort, toInetSocketArray(knownHostPorts)) + .of(ourAddressAndPort.getPort(), toInetSocketArray(knownHostPorts)) .heartBeatInterval(1, SECONDS).nonUniqueIdentifierListener(identifierListener); return ChronicleMapBuilder.of(Integer.class, @@ -634,7 +630,7 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { private final KnownNodes remoteNode; private final AtomicBoolean bootstrapRequired = new AtomicBoolean(); - private final UDPEventListener udpEventListener; + private final NodeDiscoveryEventListener nodeDiscoveryEventListener; private ProposedNodes ourProposedIdentifier; @@ -642,6 +638,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { ConcurrentExpiryMap(AddressAndPort.class, ProposedNodes.class); + private AddressAndPort ourAddressAndPort; + public ProposedNodes getOurProposedIdentifier() { return ourProposedIdentifier; } @@ -656,9 +654,12 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { Replica.ModificationNotifier modificationNotifier; - public DiscoveryNodeBytesMarshallable(final KnownNodes remoteNode, UDPEventListener udpEventListener) { + public DiscoveryNodeBytesMarshallable(@NotNull final KnownNodes remoteNode, + @NotNull final NodeDiscoveryEventListener nodeDiscoveryEventListener, + @NotNull final AddressAndPort ourAddressAndPort) { this.remoteNode = remoteNode; - this.udpEventListener = udpEventListener; + this.nodeDiscoveryEventListener = nodeDiscoveryEventListener; + this.ourAddressAndPort = ourAddressAndPort; } public KnownNodes getRemoteNodes() { @@ -740,7 +741,12 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { final ProposedNodes bootstrap = readBootstrapMessage(in); if (bootstrap != null) { - LOG.debug("received Bootstrap"); + + + if (bootstrap.addressAndPort().equals(this.ourAddressAndPort)) + return; + + LOG.info("Received Bootstrap from " + bootstrap); proposedIdentifiersWithHost.put(bootstrap.addressAndPort, bootstrap); @@ -764,8 +770,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { this.remoteNode.readMarshallable(in); this.proposedIdentifiersWithHost.readMarshallable(in); - if (udpEventListener != null) - udpEventListener.onRemoteNodeEvent(remoteNode, proposedIdentifiersWithHost); + if (nodeDiscoveryEventListener != null) + nodeDiscoveryEventListener.onRemoteNodeEvent(remoteNode, proposedIdentifiersWithHost); } @@ -862,7 +868,7 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { } -interface UDPEventListener { +interface NodeDiscoveryEventListener { /** * called when we have received a UDP message, this is called after the message has been parsed */ @@ -970,4 +976,27 @@ class ConcurrentExpiryMap interfaces = NetworkInterface.getNetworkInterfaces(); + while (interfaces.hasMoreElements()) { + NetworkInterface iface = interfaces.nextElement(); + // filters out 127.0.0.1 and inactive interfaces + if (iface.isLoopback() || !iface.isUp()) + continue; + + Enumeration addresses = iface.getInetAddresses(); + while (addresses.hasMoreElements()) { + InetAddress addr = addresses.nextElement(); + ip = addr.getHostAddress(); + System.out.println(iface.getDisplayName() + " " + ip); + } + } + } catch (SocketException e) { + throw new RuntimeException(e); + } + }*/ } diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 927c8edc8..e32fac525 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -38,7 +38,6 @@ class UdpReplicator extends UdpChannelReplicator implements Replica.Modification private static final Logger LOG = LoggerFactory.getLogger(UdpReplicator.class.getName()); - UdpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable entryExternalizable, @NotNull final UdpReplicationConfig replicationConfig, @@ -58,7 +57,6 @@ class UdpReplicator extends UdpChannelReplicator implements Replica.Modification start(); } - private static class UdpSocketChannelEntryReader implements EntryReader { private final Replica.EntryExternalizable externalizable; diff --git a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java index 9d35c921f..a809edc35 100644 --- a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java +++ b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java @@ -7,6 +7,7 @@ import org.testng.Assert; import java.io.IOException; import java.net.Inet4Address; +import java.net.InetAddress; import static java.nio.ByteBuffer.allocate; @@ -18,6 +19,10 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { @Test public void testsTheSerializationOfTheNodeDiscovery() throws IOException, InterruptedException { + final AddressAndPort ourAddressAndPort = new AddressAndPort(InetAddress.getLocalHost() + .getAddress(), + (short) 1024); + // write broadcast our address on the bootstrap method byte[] server1Address = Inet4Address.getLocalHost().getAddress(); AddressAndPort server1AddressAndPort = new AddressAndPort(server1Address, (short) 1234); @@ -27,7 +32,8 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { AddressAndPort server2AddressAndPort = new AddressAndPort(server1Address, (short) 1234); KnownNodes server1KnownNodes = new KnownNodes(); - DiscoveryNodeBytesMarshallable server1BytesExternalizable = new DiscoveryNodeBytesMarshallable(server1KnownNodes, null); + DiscoveryNodeBytesMarshallable server1BytesExternalizable = new DiscoveryNodeBytesMarshallable + (server1KnownNodes, null, ourAddressAndPort); // for the unit test we are using a ByteBufferBytes, but iltimately this data would have been send // and received via UDP @@ -44,7 +50,8 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { KnownNodes server2KnownNodes = new KnownNodes(); server2KnownNodes.add(server2AddressAndPort, (byte) SERVER2_IDENTIFER); - DiscoveryNodeBytesMarshallable server2BytesExternalizable = new DiscoveryNodeBytesMarshallable(server2KnownNodes, null); + DiscoveryNodeBytesMarshallable server2BytesExternalizable = new DiscoveryNodeBytesMarshallable + (server2KnownNodes, null, ourAddressAndPort); // add our identifier and host:port to the list of known identifiers @@ -105,7 +112,8 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { KnownNodes knownNodes = new KnownNodes(); - DiscoveryNodeBytesMarshallable bytesExternalizable = new DiscoveryNodeBytesMarshallable(knownNodes, null); + DiscoveryNodeBytesMarshallable bytesExternalizable = new DiscoveryNodeBytesMarshallable + (knownNodes, null, ourAddressAndPort); DiscoveryNodeBytesMarshallable.ProposedNodes proposedNodes = new DiscoveryNodeBytesMarshallable.ProposedNodes(server1AddressAndPort, (byte) PROPOSED_IDENTIFIER); diff --git a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java index 60a9b52fd..48a07b236 100644 --- a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java +++ b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java @@ -1,12 +1,27 @@ 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(); - final ChronicleMap map = nodeDiscovery.discoverMap(8123, 1237); + + NetworkInterface en0 = NetworkInterface.getByName("en0"); + Enumeration inetAddresses = en0.getInetAddresses(); + + AddressAndPort ourAddressAndPort1 = new AddressAndPort(inetAddresses.nextElement().getAddress(), + (short) 1237); + + final ChronicleMap map = nodeDiscovery.discoverMap(8123, ourAddressAndPort1); + + Thread.sleep(1000000); } } \ No newline at end of file