Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-134 ( release draft cut of the code so that I can test it across my network )

Browse Source
This commit is contained in:
Rob Austin
2014-10-06 15:05:18 +01:00
parent 2b3ec75bd0
commit 6c0ec26911
4 changed files with 74 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+47 -18
View File
@@ -31,8 +31,7 @@ import static net.openhft.chronicle.map.Replicators.tcp;
*/
public class NodeDiscovery {
public ChronicleMap<Integer, CharSequence> discoverMap(int udpBroadcastPort, final int tcpPort) throws IOException, InterruptedException {
public ChronicleMap<Integer, CharSequence> 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> countDownLatch = new AtomicReference<CountDownLatch>(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<AddressAndPort, DiscoveryNodeBytesMarshallable.ProposedNodes> 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, ProposedNodes>(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<K extends BytesMarshallable, V extends BytesMarshallab
this.queue.poll();
}
}
/*public static void main(String... args) {
String ip;
try {
Enumeration<NetworkInterface> 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<InetAddress> 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);
}
}*/
}
src/main/java/net/openhft/chronicle/map/UdpReplicator.java
-2
View File
@@ -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;
src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java
+11 -3
View File
@@ -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);
src/test/java/net/openhft/chronicle/map/NodeDiscoveryTest.java
+16 -1
View File
@@ -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<Integer, CharSequence> map = nodeDiscovery.discoverMap(8123, 1237);
NetworkInterface en0 = NetworkInterface.getByName("en0");
Enumeration<InetAddress> inetAddresses = en0.getInetAddresses();
AddressAndPort ourAddressAndPort1 = new AddressAndPort(inetAddresses.nextElement().getAddress(),
(short) 1237);
final ChronicleMap<Integer, CharSequence> map = nodeDiscovery.discoverMap(8123, ourAddressAndPort1);
Thread.sleep(1000000);
}
}