diff --git a/pom.xml b/pom.xml index 487fb1653..29f224587 100755 --- a/pom.xml +++ b/pom.xml @@ -19,7 +19,6 @@ xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> - net.openhft java-parent-pom @@ -31,7 +30,6 @@ -Xdoclint:none - 4.0.0 chronicle-map 2.0.0-SNAPSHOT @@ -78,20 +76,19 @@ lang - - junit - junit - test - - org.slf4j slf4j-api - + + junit + junit + test + + org.slf4j slf4j-simple @@ -129,7 +126,6 @@ - org.apache.maven.plugins maven-compiler-plugin @@ -143,7 +139,6 @@ - org.apache.maven.plugins maven-scm-publish-plugin diff --git a/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java b/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java index abb75446d..37bc7f66c 100644 --- a/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java +++ b/src/main/java/net/openhft/chronicle/map/DiscoveryCluster.java @@ -4,58 +4,241 @@ import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.collection.DirectBitSet; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; +import org.jetbrains.annotations.NotNull; import java.io.IOException; -import java.net.Inet4Address; +import java.net.*; import java.nio.ByteBuffer; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ConcurrentSkipListSet; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static java.nio.ByteBuffer.wrap; +import static java.util.concurrent.TimeUnit.SECONDS; +import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG; +import static net.openhft.chronicle.map.Replicators.tcp; /** * @author Rob Austin. */ -public class DiscoveryCluster implements UDPEventListener { +public class DiscoveryCluster { - public static final int PORT = 1235; - DirectBitSet bitSet = newBitSet(128); + public ChronicleMap discoverMap(int udpBroadcastPort, final int tcpPort) throws IOException, InterruptedException { - public void discover() throws IOException, InterruptedException { + final AtomicInteger proposedIdentifier = new AtomicInteger(); + final AtomicBoolean useAnotherIdentifier = new AtomicBoolean(); final UdpReplicationConfig udpConfig = UdpReplicationConfig - .simple(Inet4Address.getByName("255.255.255.255"), PORT); + .simple(Inet4Address.getByName("255.255.255.255"), udpBroadcastPort); final Bytes identifierBitSetBits = new ByteBufferBytes(ByteBuffer.allocate(128 / 8)); final RemoteNodes remoteNodes = new RemoteNodes(identifierBitSetBits); + final DirectBitSet knownIdentifiers = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate + (128 / 8))); + + final AddressAndPort ourAddressAndPort = new AddressAndPort(InetAddress.getLocalHost() + .getAddress(), + (short) tcpPort); + + final Set knownHostPorts = new ConcurrentSkipListSet(); + + final UDPEventListener udpEventListener = new UDPEventListener() { + + @Override + public void onRemoteNodeEvent(@NotNull final Set usedHostPorts, + @NotNull final ATSDirectBitSet usedIdentifiers, + @NotNull final Set proposedHostPortIdentifiers) { + + knownHostPorts.addAll(usedHostPorts); + + orBitSets(usedIdentifiers, knownIdentifiers); + + for (BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost : proposedHostPortIdentifiers) { + if (!proposedIdentifierWithHost.addressAndPort().equals(ourAddressAndPort)) { + + int remoteIdentifier = proposedIdentifierWithHost.identifier(); + knownIdentifiers.set(remoteIdentifier, true); + knownHostPorts.add(proposedIdentifierWithHost.addressAndPort()); + + if (remoteIdentifier == proposedIdentifier.get()) + useAnotherIdentifier.set(true); + } + } + } + }; + + final BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener); + + final BytesExternalizableImpl.ProposedIdentifierWithHost ourHostPort = new + BytesExternalizableImpl.ProposedIdentifierWithHost(ourAddressAndPort, (byte) -1); + + // to start with we will send a bootstrap that just contains our hostname without and identifier + externalizable.setOurProposedIdentifier(ourHostPort); + externalizable.sendBootStrap(); + Thread.sleep(10); + + // we should not get back some identifiers + // the identifiers will come back to the callback on the nio thread, the update arrives at the + // onRemoteNodeEvent + + externalizable.sendBootStrap(); + Thread.sleep(10); + byte identifier; + + + boolean isFistTime = true; + + for (; ; ) { + + identifier = proposeRandomUnusedIdentifier(knownIdentifiers, isFistTime); + proposedIdentifier.set(identifier); + + isFistTime = false; + + final BytesExternalizableImpl.ProposedIdentifierWithHost proposedIdentifierWithHost = new + BytesExternalizableImpl.ProposedIdentifierWithHost(ourAddressAndPort, identifier); + + externalizable.setOurProposedIdentifier(proposedIdentifierWithHost); + externalizable.sendBootStrap(); + + Thread.sleep(10); + + for (int j = 0; j < 3; j++) { + + externalizable.sendBootStrap(); + Thread.sleep(10); + + if (useAnotherIdentifier.get()) { + // given that another node host proposed the same identifier, we will choose a different one. + continue; + } + } + + break; + } - BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, this); final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster = new NodeDiscoveryBroadcaster(udpConfig, 1024, externalizable); externalizable.setModificationNotifier(nodeDiscoveryBroadcaster); - externalizable.sendBootStrap(); - Thread.sleep(100); - externalizable.sendBootStrap(); - // we have to wait a while, we have not way of knowing how long it will take for a UDP message to - // arrive, but the more UDP messages we receive the better, is it will make our selection of a - // unique identifier more reliable. - Thread.sleep(100); + // we should make a local copy as this may change + + final IdentifierListener identifierListener = new IdentifierListener() { + + final ConcurrentMap identifiers = new ConcurrentHashMap(); + + @Override + public boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress) { + final SocketAddress socketAddress = identifiers.putIfAbsent(remoteIdentifier, remoteAddress); + return socketAddress == null; + } + }; + + + final TcpReplicationConfig tcpConfig = TcpReplicationConfig + .of(tcpPort, toInetSocketArray(knownHostPorts)) + .heartBeatInterval(1, SECONDS).nonUniqueIdentifierListener(identifierListener); + + return ChronicleMapBuilder.of(Integer.class, + CharSequence.class) + .entries(20000L) + .addReplicator(tcp(identifier, tcpConfig)).create(); } - @Override - public void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId) { + private InetSocketAddress[] toInetSocketArray(Set source) throws + UnknownHostException { - if (remoteProposedId == -1) - bitSet.set(remoteProposedId); + // make a safe copy + final HashSet addressAndPorts = new HashSet(source); - // propose and identifier + if (addressAndPorts.isEmpty()) + return new InetSocketAddress[0]; + + final InetSocketAddress[] addresses = new InetSocketAddress[addressAndPorts.size()]; + + int i = 0; + + for (final AddressAndPort addressAndPort : addressAndPorts) { + addresses[i++] = new InetSocketAddress(InetAddress.getByAddress(addressAndPort.address()) + .getHostAddress(), addressAndPort.port()); + } + return addresses; } + /** + * bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset + * and returns the destination + * + * @param source + * @param destination + * @return + */ + private DirectBitSet orBitSets(@NotNull final DirectBitSet source, + @NotNull final DirectBitSet destination) { + + // merges the two bit-sets together + for (int i = (int) source.nextSetBit(0); i > 0; + i = (int) source.nextSetBit(i + 1)) { + try { + destination.set(i, true); + } catch (IndexOutOfBoundsException e) { + LOG.error("", e); + } + } + + return destination; + } + + + private byte proposeRandomUnusedIdentifier(final DirectBitSet knownIdentifiers, boolean isFirstTime) throws UnknownHostException { + byte possible; + + + // the first time, rather than choosing a random number, we will choose the last value of the IP + // address as our random number, ( or at least something that is based upon it) + if (isFirstTime) { + byte[] address = InetAddress.getLocalHost().getAddress(); + int lastAddress = address[address.length - 1]; + if (lastAddress > 127) + lastAddress = lastAddress - 127; + if (lastAddress > 127) + lastAddress = lastAddress - 127; + + possible = (byte) lastAddress; + } else + possible = (byte) (Math.random() * 128); + + int count = 0; + for (; ; ) { + + if (knownIdentifiers.setIfClear(possible)) { + return possible; + } + + count++; + + if (count == 128) { + throw new IllegalStateException("The grid is full, its not possible for any more nodes to " + + "going the grid."); + } + + if (possible == 128) + possible = 0; + else + possible++; + } + } /** * creates a bit set based on a number of bits diff --git a/src/main/java/net/openhft/chronicle/map/IdentifierListener.java b/src/main/java/net/openhft/chronicle/map/IdentifierListener.java new file mode 100644 index 000000000..2d324af8e --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/IdentifierListener.java @@ -0,0 +1,20 @@ +package net.openhft.chronicle.map; + +import java.net.SocketAddress; + +/** + * @author Rob Austin. + */ +public interface IdentifierListener { + + /** + * checks the identifier that is unique and we haven't see it before, unless it comes from the same port + * and host. + * + * @param remoteIdentifier + * @param remoteAddress + * @return + */ + boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress); + +} diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java index 0b99af1d0..44b75a892 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryBroadcaster.java @@ -9,14 +9,19 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; -import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.DatagramChannel; -import java.util.Comparator; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Queue; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.atomic.AtomicBoolean; +import static net.openhft.chronicle.map.BytesExternalizableImpl.ProposedIdentifierWithHost; import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES; import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG; @@ -42,7 +47,6 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { BOOTSTRAP_BYTES.append(BOOTSTRAP); } - /** * @param replicationConfig * @param externalizable @@ -56,15 +60,10 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { super(replicationConfig, serializedEntrySize, UNUSED); - - final UdpSocketChannelEntryWriter writer = new UdpSocketChannelEntryWriter(1024, externalizable, - this); - final UdpSocketChannelEntryReader reader = new UdpSocketChannelEntryReader(1024, externalizable - ); - + final UdpSocketChannelEntryWriter writer = new UdpSocketChannelEntryWriter(1024, externalizable, this); + final UdpSocketChannelEntryReader reader = new UdpSocketChannelEntryReader(1024, externalizable); setReader(reader); - setWriter(writer); start(); @@ -72,12 +71,10 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { private static class UdpSocketChannelEntryReader implements EntryReader { - private final ByteBuffer in; private final ByteBufferBytes out; private final BytesExternalizable externalizable; - /** * @param serializedEntrySize the maximum size of an entry include the meta data * @param externalizable supports reading and writing serialize entries @@ -135,11 +132,9 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { public static class UdpSocketChannelEntryWriter implements EntryWriter { - private final ByteBuffer out; private final ByteBufferBytes in; - @NotNull private final BytesExternalizable externalizable; private UdpChannelReplicator udpReplicator; @@ -154,8 +149,6 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { // we make the buffer twice as large just to give ourselves headroom out = ByteBuffer.allocateDirect(serializedEntrySize * 2); in = new ByteBufferBytes(out); - - } @@ -194,19 +187,15 @@ public class NodeDiscoveryBroadcaster extends UdpChannelReplicator { long size = in.position() - start; - // we'll write the size inverted at the start in.writeShort(0, ~((int) size)); in.writeUnsignedShort(SIZE_OF_SHORT, (int) size); - out.limit((int) in.position()); udpReplicator.disableWrites(); return socketChannel.write(out); - - } } @@ -230,48 +219,31 @@ class RemoteNodes { private final Bytes activeIdentifiersBitSetBytes; - private ConcurrentSkipListSet inetSocketAddresses; + private final ConcurrentSkipListSet addressAndPorts; private final ATSDirectBitSet atsDirectBitSet; - private byte ourProposedID = -1; - private final ATSDirectBitSet allProposedID; + /** * @param activeIdentifiersBitSetBytes byte sof a bitset containing the known identifiers */ RemoteNodes(final Bytes activeIdentifiersBitSetBytes) { - this.inetSocketAddresses = new ConcurrentSkipListSet(new Comparator() { - - @Override - public int compare(InetSocketAddress o1, InetSocketAddress o2) { - - int result = Integer.compare(o1.getPort(), o2.getPort()); - if (result != 0) - return result; - - return o1.getHostName().compareTo(o2.getHostName()); - - } - }); + this.addressAndPorts = new ConcurrentSkipListSet(); this.activeIdentifiersBitSetBytes = activeIdentifiersBitSetBytes; this.atsDirectBitSet = new ATSDirectBitSet(this.activeIdentifiersBitSetBytes); - - // this will only hold bits in it while we in the phase of searching for an id - this.allProposedID = new ATSDirectBitSet(new ByteBufferBytes(ByteBuffer.allocate(128 / 8))); } - - public Set inetSocketAddresses() { - return inetSocketAddresses; + public Set addressAndPorts() { + return addressAndPorts; } public Bytes activeIdentifierBytes() { return activeIdentifiersBitSetBytes; } - public void add(InetSocketAddress inetSocketAddress) { - inetSocketAddresses.add(inetSocketAddress); + public void add(AddressAndPort inetSocketAddress) { + addressAndPorts.add(inetSocketAddress); } public DirectBitSet activeIdentifierBitSet() { @@ -279,15 +251,88 @@ class RemoteNodes { } - public byte ourProposedID() { - return ourProposedID; +} + + +class AddressAndPort implements Comparable { + private byte[] address; + private short port; + + public AddressAndPort(byte[] address, short port) { + if (address.length != 4) + throw new IllegalStateException("address.length should equal '4' but is length=" + + address.length); + this.address = address; + this.port = port; } - public DirectBitSet allProposedID() { - return allProposedID; + public byte[] getAddress() { + return address; } + + public short getPort() { + return port; + } + + public byte[] address() { + + return address; + } + + public short port() { + return port; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + AddressAndPort that = (AddressAndPort) o; + + if (port != that.port) return false; + if (!Arrays.equals(address, that.address)) return false; + + return true; + } + + @Override + public int hashCode() { + int result = Arrays.hashCode(address); + result = 31 * result + (int) port; + return result; + } + + + @Override + public int compareTo(AddressAndPort o) { + int i = 0; + for (byte b : address) { + int compare = Byte.compare(b, o.address[i++]); + if (compare != 0) + return compare; + + } + return Short.compare(port, o.port); + } + + + @Override + public String toString() { + return "AddressAndPort{" + + "address=" + numericToTextFormat(address) + + ", port=" + port + + '}'; + } + + static String numericToTextFormat(byte[] src) { + return (src[0] & 0xff) + "." + (src[1] & 0xff) + "." + (src[2] & 0xff) + "." + (src[3] & 0xff); + } + + } + class BytesExternalizableImpl implements BytesExternalizable { private final RemoteNodes allNodes; @@ -295,6 +340,16 @@ class BytesExternalizableImpl implements BytesExternalizable { private final AtomicBoolean bootstrapRequired = new AtomicBoolean(); private final UDPEventListener udpEventListener; + private ProposedIdentifierWithHost ourProposedIdentifier; + + public ProposedIdentifierWithHost getOurProposedIdentifier() { + return ourProposedIdentifier; + } + + public void setOurProposedIdentifier(ProposedIdentifierWithHost ourProposedIdentifier) { + this.ourProposedIdentifier = ourProposedIdentifier; + } + public void setModificationNotifier(Replica.ModificationNotifier modificationNotifier) { this.modificationNotifier = modificationNotifier; @@ -319,24 +374,40 @@ class BytesExternalizableImpl implements BytesExternalizable { if (bootstrapRequired.getAndSet(false)) { + + final ProposedIdentifierWithHost ourProposedIdentifier = getOurProposedIdentifier(); + if (ourProposedIdentifier == null || ourProposedIdentifier.addressAndPort == null) + return; + BOOTSTRAP_BYTES.clear(); destination.write(BOOTSTRAP_BYTES); + + + for (byte b : ourProposedIdentifier.addressAndPort.address()) { + destination.writeByte(b); + } + + destination.writeShort(ourProposedIdentifier.addressAndPort.port()); + destination.writeByte(ourProposedIdentifier.identifier()); return; } - final Set inetSocketAddresses = allNodes.inetSocketAddresses(); + final Set addressAndPorts = allNodes.addressAndPorts(); // write the number of hosts and ports - short count = (short) inetSocketAddresses.size(); + short count = (short) addressAndPorts.size(); destination.writeUnsignedShort(count); // write all the host and ports - for (InetSocketAddress inetSocketAddress : inetSocketAddresses) { - destination.writeUTF(inetSocketAddress.getHostName()); - destination.writeInt(inetSocketAddress.getPort()); + for (AddressAndPort inetSocketAddress : addressAndPorts) { + for (byte b : inetSocketAddress.address()) { + destination.writeByte(b); + } + + destination.writeShort(inetSocketAddress.port()); } - destination.write(allNodes.ourProposedID()); + // destination.write(allNodes.ourProposedID()); Bytes bytes = allNodes.activeIdentifierBytes(); //bytes.position(0); @@ -346,36 +417,123 @@ class BytesExternalizableImpl implements BytesExternalizable { // we will store the size of the bitset destination.writeUnsignedShort((int) bytes.remaining()); - // then write it out - destination.write(bytes); + // we are no going to broadcast all the nodes that have been bootstaping in the last 200 + expireOldBootstrapStats(System.currentTimeMillis() - 200); + + // this is where we will store the count of the bootstrap data which is written below + destination.skip(2); + long countLocation = destination.position(); + int countBootstrapDataRepublished = 0; + + // rebroadcast all the host:port:identifier of the nodes that have sent a bootstrap in the last 200 + // milliseconds + for (ProposedIdentifierWithHost proposedIdentifierWithHost : bootstrapStatsMap.values()) { + + for (byte b : proposedIdentifierWithHost.addressAndPort.address()) { + destination.writeByte(b); + } + + destination.writeShort(proposedIdentifierWithHost.addressAndPort.port()); + destination.writeByte(proposedIdentifierWithHost.identifier()); + countBootstrapDataRepublished++; + } + + destination.writeInt(countLocation, countBootstrapDataRepublished); + } + final ConcurrentMap bootstrapStatsMap = new + ConcurrentHashMap(); + + final Queue proposedIdentifierWithHostExpiryQueue = new ConcurrentLinkedQueue(); + + + private void expireOldBootstrapStats(long timeOlderThan) { + for (; ; ) { + + final ProposedIdentifierWithHost proposedIdentifierWithHost = proposedIdentifierWithHostExpiryQueue.peek(); + + if (proposedIdentifierWithHost == null) + break; + + if (proposedIdentifierWithHost.timestamp < timeOlderThan) { + // only remote it if it has not changed + bootstrapStatsMap.remove(proposedIdentifierWithHost.addressAndPort, proposedIdentifierWithHost); + } + + proposedIdentifierWithHostExpiryQueue.poll(); + } + } + + @Override public void readExternalBytes(@NotNull Bytes source) { - if (isBootstrap(source)) { + final ProposedIdentifierWithHost bootstrap = readBootstrapMessage(source); + if (bootstrap != null) { LOG.debug("received Bootstrap"); - // we've received a bootstrap message so will will now rebroadcast what we know - onChange(); + bootstrapStatsMap.put(bootstrap.addressAndPort, bootstrap); + proposedIdentifierWithHostExpiryQueue.add(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)); + + onChange(); + + } catch (InterruptedException e) { + LOG.error("", e); + } return; } int count = source.readUnsignedShort(); + + final Set allNodeInTheGrid = new HashSet(); + for (int i = 0; i < count; i++) { + byte[] address = new byte[4]; + address[0] = source.readByte(); + address[1] = source.readByte(); + address[2] = source.readByte(); + address[3] = source.readByte(); - final String host = source.readUTF(); - final int port = source.readInt(); + final short port = source.readShort(); - LOG.debug("received InetSocketAddress(" + host + "," + port + ")"); - allNodes.add(new InetSocketAddress(host, port)); + LOG.debug("received InetSocketAddress(" + address + "," + port + ")"); + allNodeInTheGrid.add(new AddressAndPort(address, port)); } - byte remoteProposedId = source.readByte(); + int countRebroadcast = source.readShort(); + final Set proposedIdentifierWithHosts = new + HashSet(); + + // rebroadcast all the host:port:identifier of the nodes that have sent a bootstrap in the last 200 + // milliseconds + for (int i = 0; i < countRebroadcast; i++) { + byte[] address = new byte[4]; + address[0] = source.readByte(); + address[1] = source.readByte(); + address[2] = source.readByte(); + address[3] = source.readByte(); + + final short port = source.readShort(); + final byte identifier = source.readByte(); + + final ProposedIdentifierWithHost proposed = new ProposedIdentifierWithHost( + address, port, identifier); + + proposedIdentifierWithHosts.add(proposed); + } // the number of bytes in the bitset int sizeInBytes = source.readUnsignedShort(); @@ -385,14 +543,14 @@ class BytesExternalizableImpl implements BytesExternalizable { return; } + //todo we should not have to do this ! - maybe a bug in BitSet of ByteBuffer - Bytes sourceBytes = toNewBuffer(source); + Bytes usedIdentifiers = toNewBuffer(source); - // merges the source into the destination and returns the destination - orBitSets(new ATSDirectBitSet(sourceBytes), allNodes.activeIdentifierBitSet()); + udpEventListener.onRemoteNodeEvent(allNodeInTheGrid, + new ATSDirectBitSet(usedIdentifiers), + proposedIdentifierWithHosts); - - udpEventListener.onRemoteNodeEvent(allNodes, remoteProposedId); } @@ -426,11 +584,62 @@ class BytesExternalizableImpl implements BytesExternalizable { } + public static class ProposedIdentifierWithHost { + + private final int identifier; + private final long timestamp; + private final AddressAndPort addressAndPort; + + public ProposedIdentifierWithHost(@NotNull final AddressAndPort addressAndPort, + byte identifier) { + this.addressAndPort = addressAndPort; + this.identifier = identifier; + this.timestamp = System.currentTimeMillis(); + } + + public ProposedIdentifierWithHost(byte[] address, short port, byte identifier) { + this(new AddressAndPort(address, port), identifier); + } + + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + ProposedIdentifierWithHost that = (ProposedIdentifierWithHost) o; + + if (identifier != that.identifier) return false; + if (timestamp != that.timestamp) return false; + if (addressAndPort != null ? !addressAndPort.equals(that.addressAndPort) : that.addressAndPort != null) + return false; + + return true; + } + + @Override + public int hashCode() { + int result = identifier; + result = 31 * result + (int) (timestamp ^ (timestamp >>> 32)); + result = 31 * result + (addressAndPort != null ? addressAndPort.hashCode() : 0); + return result; + } + + public int identifier() { + return identifier; + + } + + public AddressAndPort addressAndPort() { + return addressAndPort; + } + } + /** * @param source * @return returns true if the UDP message contains the text 'BOOTSTRAP' */ - private boolean isBootstrap(Bytes source) { + private ProposedIdentifierWithHost readBootstrapMessage(Bytes source) { final long start = source.position(); @@ -439,58 +648,40 @@ class BytesExternalizableImpl implements BytesExternalizable { long size = BOOTSTRAP_BYTES.limit(); if (size < source.remaining()) - return false; - - if (source.remaining() < size) - return false; + return null; for (int i = 0; i < size; i++) { - final byte actualByte = source.readByte(start + i); + final byte byteRead = source.readByte(start + i); final byte expectedByte = BOOTSTRAP_BYTES.readByte(i); - if (!(expectedByte == actualByte)) - return false; + if (!(expectedByte == byteRead)) + return null; } - return true; + byte[] address = new byte[4]; + address[0] = source.readByte(); + address[1] = source.readByte(); + address[2] = source.readByte(); + address[3] = source.readByte(); + + final short port = source.readShort(); + final byte identifier = source.readByte(); + + return new ProposedIdentifierWithHost(address, port, identifier); + } finally { source.position(start); } } - /** - * bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset - * and returns the destination - * - * @param source - * @param destination - * @return - */ - private DirectBitSet orBitSets(@NotNull final DirectBitSet source, - @NotNull final DirectBitSet destination) { - - - // merges the two bit-sets together - for (int i = (int) source.nextSetBit(0); i > 0; - i = (int) source.nextSetBit(i + 1)) { - try { - destination.set(i, true); - } catch (IndexOutOfBoundsException e) { - LOG.error("", e); - } - } - - return destination; - } - public void sendBootStrap() { bootstrapRequired.set(true); } - public void add(InetSocketAddress interfaceAddress) { - allNodes.add(interfaceAddress); + public void add(AddressAndPort addressAndPort) { + allNodes.add(addressAndPort); } @@ -504,10 +695,9 @@ interface UDPEventListener { /** * called when we have received a UDP message, this is called after the message has been parsed - * - * @param allNodes - * @param remoteProposedId a remote node has proposed using this as an id */ - void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId); + void onRemoteNodeEvent(Set usedHostPorts, + ATSDirectBitSet usedIdentifers, + Set proposedHostPortIdentifiers); } diff --git a/src/main/java/net/openhft/chronicle/map/NonUniqueIdentifierListener.java b/src/main/java/net/openhft/chronicle/map/NonUniqueIdentifierListener.java deleted file mode 100644 index 250a664ee..000000000 --- a/src/main/java/net/openhft/chronicle/map/NonUniqueIdentifierListener.java +++ /dev/null @@ -1,9 +0,0 @@ -package net.openhft.chronicle.map; - -/** - * @author Rob Austin. - */ -public interface NonUniqueIdentifierListener { - - void onNonUniqueIdentifier(byte remoteIdentifier); -} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java index 60238f102..b213f4455 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java @@ -16,6 +16,8 @@ package net.openhft.chronicle.map; +import org.jetbrains.annotations.NotNull; + import java.net.InetSocketAddress; import java.util.Arrays; import java.util.Collection; @@ -41,34 +43,38 @@ public class TcpReplicationConfig { private ThrottlingConfig throttlingConfig = ThrottlingConfig.noThrottling(); private long heartBeatInterval = DEFAULT_HEART_BEAT_INTERVAL; private TimeUnit heartBeatIntervalUnit = DEFAULT_HEART_BEAT_INTERVAL_UNIT; - private NonUniqueIdentifierListener nonUniqueIdentifierListener; + public IdentifierListener identifierListener; TcpReplicationConfig() { } - public NonUniqueIdentifierListener nonUniqueIdentifierListener() { - return nonUniqueIdentifierListener; + public IdentifierListener identifierListener() { + return identifierListener; } - public void nonUniqueIdentifierListener(NonUniqueIdentifierListener nonUniqueIdentifierListener) { - this.nonUniqueIdentifierListener = nonUniqueIdentifierListener; + public TcpReplicationConfig nonUniqueIdentifierListener(@NotNull final IdentifierListener + identifierListener) { + this.identifierListener = identifierListener; + return this; } public boolean autoReconnectedUponDroppedConnection() { return autoReconnectedUponDroppedConnection; } - public void autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) { + public TcpReplicationConfig autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) { this.autoReconnectedUponDroppedConnection = autoReconnectedUponDroppedConnection; + return this; } public ThrottlingConfig throttlingConfig() { return throttlingConfig; } - public void throttlingConfig(ThrottlingConfig throttlingConfig) { + public TcpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) { this.throttlingConfig = throttlingConfig; + return this; } @@ -84,15 +90,16 @@ public class TcpReplicationConfig { return serverPort; } - public void serverPort(int serverPort) { + public TcpReplicationConfig serverPort(int serverPort) { this.serverPort = serverPort; + return this; } public Set endpoints() { return endpoints; } - public void endpoints(Set endpoints) { + public TcpReplicationConfig endpoints(Set endpoints) { for (final InetSocketAddress endpoint : endpoints) { if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName())) @@ -101,6 +108,7 @@ public class TcpReplicationConfig { } this.endpoints = endpoints; + return this; } public int packetSize() { @@ -133,15 +141,17 @@ public class TcpReplicationConfig { } - public void packetSize(int packetSize) { + public TcpReplicationConfig packetSize(int packetSize) { if (packetSize <= 0) throw new IllegalArgumentException(); this.packetSize = packetSize; + return this; } - public void throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) { + public TcpReplicationConfig throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) { this.heartBeatInterval = heartBeatInterval; this.heartBeatIntervalUnit = heartBeatIntervalUnit; + return this; } @@ -194,4 +204,6 @@ public class TcpReplicationConfig { this.heartBeatIntervalUnit = unit; return this; } + + } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index ea372f7ce..0038800e5 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -23,10 +23,7 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; import java.io.IOException; -import java.net.ConnectException; -import java.net.InetSocketAddress; -import java.net.ServerSocket; -import java.net.SocketException; +import java.net.*; import java.nio.ByteBuffer; import java.nio.channels.*; import java.util.BitSet; @@ -86,7 +83,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { selectorTimeout = Math.min(heartBeatIntervalMillis, throttleBucketInterval); - this.replica = replica; this.localIdentifier = replica.identifier(); this.maxEntrySizeBytes = maxEntrySizeBytes; @@ -424,19 +420,27 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { // this can occur sometimes when if 2 or more remote node attempt to use node discovery at the // same time - if (remoteIdentifier == localIdentifier) { - final NonUniqueIdentifierListener listener = replicationConfig.nonUniqueIdentifierListener(); + final IdentifierListener identifierListener = replicationConfig.identifierListener; + final SocketAddress remoteAddress = socketChannel.getRemoteAddress(); + + boolean identifierIsUnique; + + if (identifierListener != null) { + identifierIsUnique = identifierListener.isIdentifierUnique(remoteIdentifier, remoteAddress); + } else { + identifierIsUnique = remoteIdentifier != localIdentifier; + } + + if (identifierIsUnique) { - if (listener != null) - listener.onNonUniqueIdentifier(remoteIdentifier); // throwing this exception will cause us to disconnect, both the client and server // will be able to detect the the remote and local identifiers are the same, // as the identifier is send early on in the hand shaking via the connect(() and accept() // methods - throw new IllegalStateException("dropping connection, as the remote and local-identifier " + - "are the same, identifier=" + localIdentifier); + throw new IllegalStateException("dropping connection, " + + "as the remote-identifier is already being used, identifier=" + remoteIdentifier); } attached.remoteModificationIterator = replica.acquireModificationIterator(remoteIdentifier, diff --git a/src/test/java/net/openhft/chronicle/map/ConcurrentMap8Test.java b/src/test/java/net/openhft/chronicle/map/ConcurrentMap8Test.java index cbc5b4879..0bf73407a 100644 --- a/src/test/java/net/openhft/chronicle/map/ConcurrentMap8Test.java +++ b/src/test/java/net/openhft/chronicle/map/ConcurrentMap8Test.java @@ -26,16 +26,8 @@ package net.openhft.chronicle.map; import junit.framework.TestSuite;*/ import net.openhft.chronicle.map.jrs166.JSR166TestCase; -import org.junit.Test; -import java.io.File; -import java.io.IOException; -import java.util.*; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.function.BiFunction; - -import static org.junit.Assert.*; public class ConcurrentMap8Test extends JSR166TestCase { @@ -43,7 +35,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { private static final double EPSILON = 1E-5; static ConcurrentMap longMap; - private static ConcurrentMap newLongMap() { + // ******** commented out for the moment as it only works in java8 ******************************** + + /* private static ConcurrentMap newLongMap() { //First create (or access if already created) the shared map final ChronicleMapBuilder builder = ChronicleMapBuilder.of(Long.class, Long.class); @@ -87,9 +81,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { } - /** + *//** * Returns a new map from Integers 1-5 to Strings "A"-"E". - */ + *//* private static ConcurrentMap map5() { ConcurrentMap map = newIntegerStringMap(); assertTrue(map.isEmpty()); @@ -105,9 +99,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { return map; } - /** + *//** * Returns a new map from Integers 1-5 to Strings "A"-"E". - */ + *//* private static ConcurrentHashMap map5ConcurrentHashMap() { ConcurrentHashMap map = new ConcurrentHashMap(5); assertTrue(map.isEmpty()); @@ -153,9 +147,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { return longMap; } - /** + *//** * getOrDefault returns value if present, else default - */ + *//* @Test public void testGetOrDefault() { ConcurrentMap map = map5(); @@ -163,9 +157,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(map.getOrDefault(six, "Z"), "Z"); } - /** + *//** * computeIfAbsent adds when the given key is not present - */ + *//* @Test public void testComputeIfAbsent() { ConcurrentMap map = map5(); @@ -173,18 +167,18 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertTrue(map.containsKey(six)); } - /** + *//** * computeIfAbsent does not replace if the key is already present - */ + *//* @Test public void testComputeIfAbsent2() { ConcurrentMap map = map5(); assertEquals("A", map.computeIfAbsent(one, (x) -> "Z")); } - /** + *//** * computeIfAbsent does not add if function returns null - */ + *//* @Test public void testComputeIfAbsent3() { ConcurrentMap map = map5(); @@ -192,9 +186,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(map.containsKey(six)); } - /** + *//** * computeIfPresent does not replace if the key is already present - */ + *//* @Test public void testComputeIfPresent() { ConcurrentMap map = map5(); @@ -202,18 +196,18 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(map.containsKey(six)); } - /** + *//** * computeIfPresent adds when the given key is not present - */ + *//* @Test public void testComputeIfPresent2() { ConcurrentMap map = map5(); assertEquals("Z", map.computeIfPresent(one, (x, y) -> "Z")); } - /** + *//** * compute does not replace if the function returns null - */ + *//* @Test public void testCompute() { ConcurrentMap map = map5(); @@ -221,27 +215,27 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(map.containsKey(six)); } - /** + *//** * compute adds when the given key is not present - */ + *//* @Test public void testCompute2() { ConcurrentMap map = map5(); assertEquals("Z", map.compute(six, (x, y) -> "Z")); } - /** + *//** * compute replaces when the given key is present - */ + *//* @Test public void testCompute3() { ConcurrentMap map = map5(); assertEquals("Z", map.compute(one, (x, y) -> "Z")); } - /** + *//** * compute removes when the given key is present and function returns null - */ + *//* @Test public void testCompute4() { ConcurrentMap map = map5(); @@ -252,27 +246,27 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(map.containsKey(one)); } - /** + *//** * merge adds when the given key is not present - */ + *//* @Test public void testMerge1() { ConcurrentMap map = map5(); assertEquals("Y", map.merge(six, "Y", (x, y) -> "Z")); } - /** + *//** * merge replaces when the given key is present - */ + *//* @Test public void testMerge2() { ConcurrentMap map = map5(); assertEquals("Z", map.merge(one, "Y", (x, y) -> "Z")); } - /** + *//** * merge removes when the given key is present and function returns null - */ + *//* @Test public void testMerge3() { ConcurrentMap map = map5(); @@ -280,9 +274,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(map.containsKey(one)); } - /* + *//* * replaceAll replaces all matching values. - */ + *//* @Test public void testReplaceAll() { ConcurrentMap map = map5(); @@ -299,18 +293,18 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals("Z", map.get(five)); } - /** + *//** * Default-constructed set is empty - */ + *//* @Test public void testNewKeySet() { Set a = ConcurrentHashMap.newKeySet(); assertTrue(a.isEmpty()); } - /** + *//** * keySet.addAll adds each element from the given collection - */ + *//* @Test public void testAddAll() { Set full = populatedSet(3); @@ -322,9 +316,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(6, full.size()); } - /** + *//** * keySet.addAll adds each element from the given collection that did not already exist in the set - */ + *//* @Test public void testAddAll2() { Set full = populatedSet(3); @@ -336,9 +330,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(5, full.size()); } - /** + *//** * keySet.add will not add the element if it already exists in the set - */ + *//* @Test public void testAdd2() { Set full = populatedSet(3); @@ -346,9 +340,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(3, full.size()); } - /** + *//** * keySet.add adds the element when it does not exist in the set - */ + *//* @Test public void testAdd3() { Set full = populatedSet(3); @@ -356,9 +350,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertTrue(full.contains(three)); } - /** + *//** * keySet.add throws UnsupportedOperationException if no default mapped value - */ + *//* @Test public void testAdd4() { Set full = map5().keySet(); @@ -369,9 +363,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { } } - /** + *//** * keySet.add throws NullPointerException if the specified key is null - */ + *//* @Test public void testAdd5() { Set full = populatedSet(3); @@ -382,9 +376,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { } } - /** + *//** * keyset.clear removes all elements from the set - */ + *//* @Test public void testClear() { Set full = populatedSet(3); @@ -392,9 +386,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(0, full.size()); } - /** + *//** * keyset.contains returns true for added elements - */ + *//* @Test public void testContains() { Set full = populatedSet(3); @@ -402,9 +396,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(full.contains(five)); } - /** + *//** * KeySets with equal elements are equal - */ + *//* @Test public void testEquals() { Set a = populatedSet(3); @@ -421,9 +415,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(a.hashCode(), b.hashCode()); } - /** + *//** * KeySet.containsAll returns true for collections with subset of elements - */ + *//* @Test public void testContainsAll() { Set full = populatedSet(3); @@ -435,9 +429,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(full.containsAll(v)); } - /** + *//** * KeySet.isEmpty is true when empty, else false - */ + *//* @Test public void testIsEmpty() { Set empty = ConcurrentHashMap.newKeySet(); @@ -446,9 +440,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(full.isEmpty()); } - /** + *//** * KeySet.iterator() returns an iterator containing the elements of the set - */ + *//* @Test public void testIterator() { Collection empty = ConcurrentHashMap.newKeySet(); @@ -479,9 +473,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { } } - /** + *//** * KeySet.iterator.remove removes current element - */ + *//* @Test public void testIteratorRemove() { Set q = populatedSet(3); @@ -495,9 +489,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertFalse(it.hasNext()); } - /** + *//** * KeySet.toString holds toString of elements - */ + *//* @Test public void testToString() { assertEquals("[]", ConcurrentHashMap.newKeySet().toString()); @@ -507,9 +501,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertTrue(s.contains(String.valueOf(i))); } - /** + *//** * KeySet.removeAll removes all elements from the given collection - */ + *//* @Test public void testRemoveAll() { Set full = populatedSet(3); @@ -520,9 +514,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(1, full.size()); } - /** + *//** * KeySet.remove removes an element - */ + *//* @Test public void testRemove() { Set full = populatedSet(3); @@ -531,9 +525,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(2, full.size()); } - /** + *//** * keySet.size returns the number of elements - */ + *//* @Test public void testSize() { Set empty = ConcurrentHashMap.newKeySet(); @@ -542,9 +536,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertEquals(0, empty.size()); } - /** + *//** * KeySet.toArray() returns an Object array containing all elements from the set - */ + *//* @Test public void testToArray() { Object[] a = ConcurrentHashMap.newKeySet().toArray(); @@ -562,9 +556,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertSame(Object[].class, full.toArray().getClass()); } - /** + *//** * toArray(Integer array) returns an Integer array containing all elements from the set - */ + *//* @Test public void testToArray2() { Collection empty = ConcurrentHashMap.newKeySet(); @@ -599,9 +593,9 @@ public class ConcurrentMap8Test extends JSR166TestCase { assertTrue(Arrays.asList(elements).containsAll(Arrays.asList(full.toArray(a)))); } - /** + *//** * A deserialized serialized set is equal - */ + *//* @Test public void testSerialization() throws Exception { int size = 20; @@ -623,5 +617,5 @@ public class ConcurrentMap8Test extends JSR166TestCase { } } - +*/ } \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java index 4005e33e0..f472b0602 100644 --- a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java +++ b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryBroadcasterTest.java @@ -1,6 +1,10 @@ package net.openhft.chronicle.map; import junit.framework.TestCase; +import org.junit.Ignore; +import org.junit.Test; + +import java.io.IOException; public class NodeDiscoveryBroadcasterTest extends TestCase { @@ -8,93 +12,25 @@ public class NodeDiscoveryBroadcasterTest extends TestCase { // test to stop it erroring with no test } - - /* @Test - @Ignore - public void test() throws IOException, InterruptedException { - - final UdpReplicationConfig udpConfig = UdpReplicationConfig - .simple(Inet4Address.getByName("255.255.255.255"), 1235); - - final Bytes identifierBitSetBits = new ByteBufferBytes(ByteBuffer.allocate(128 / 8)); - - final RemoteNodes remoteNodes = new RemoteNodes(identifierBitSetBits); - - UDPEventListener udpEventListener = new UDPEventListener() { - - @Override - public void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId) { - - } - }; - - BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener); + private ChronicleMap map1; + private ChronicleMap map2; + private volatile byte proposedIdentifier = 1; + private volatile boolean useAnotherIdentifer = false; - final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster - = new NodeDiscoveryBroadcaster(udpConfig, 1024, externalizable); - externalizable.setModificationNotifier(nodeDiscoveryBroadcaster); - - - Thread.sleep(1000); - - externalizable.sendBootStrap(); - - externalizable.add(new InetSocketAddress("myhost", 8888)); - externalizable.add(new InetSocketAddress("myOtherHost", 8888)); - externalizable.add((byte) 2); - externalizable.onChange(); - Thread.sleep(10000); - externalizable.add((byte) 2); - externalizable.onChange(); - Thread.sleep(10000); - externalizable.add((byte) 3); - externalizable.onChange(); - Thread.sleep(10000); - externalizable.add((byte) 4); - externalizable.onChange(); - Thread.sleep(10000); - externalizable.add((byte) 1); - externalizable.onChange(); - Thread.sleep(10000); - externalizable.onChange(); - - Thread.sleep(1000); - } @Test @Ignore - public void testJustListen() throws IOException, InterruptedException { + public void test2() throws IOException, InterruptedException { - final UdpReplicationConfig udpConfig = UdpReplicationConfig - .simple(Inet4Address.getByName("255.255.255.255"), 1235); + DiscoveryCluster discoveryCluster = new DiscoveryCluster(); - final Bytes identifierBitSetBits = new ByteBufferBytes(ByteBuffer.allocate(128 / 8)); - - final RemoteNodes remoteNodes = new RemoteNodes(identifierBitSetBits); - - UDPEventListener udpEventListener = new UDPEventListener() { - - @Override - public void onRemoteNodeEvent(RemoteNodes allNodes, byte remoteProposedId) { - - } - }; - - BytesExternalizableImpl externalizable = new BytesExternalizableImpl(remoteNodes, udpEventListener); + ChronicleMap map = discoveryCluster.discoverMap(1025, 8086); + } - final NodeDiscoveryBroadcaster nodeDiscoveryBroadcaster - = new NodeDiscoveryBroadcaster(udpConfig, 1024, externalizable); - externalizable.setModificationNotifier(nodeDiscoveryBroadcaster); - - - Thread.sleep(10000000); - - - }*/ }