From 4d889222e3333e0525c8cac10eb6879aea896535 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 25 Sep 2014 16:42:22 +0100 Subject: [PATCH] HCOLL-134 and HCOLL-145 ( release draft cut of the code so that I can test it across my network ) --- pom.xml | 4 - .../map/AbstractChannelReplicator.java | 7 +- .../map/NodeDiscoveryHostPortBroadcaster.java | 345 ++++++++++++++++++ .../chronicle/map/ReplicatingCluster.java | 5 +- .../chronicle/map/TcpReplicationConfig.java | 6 +- .../map/TcpReplicationConfigBean.java | 116 ++++++ .../chronicle/map/ThrottlingConfig.java | 18 +- .../chronicle/map/ThrottlingConfigBean.java | 87 +++++ .../chronicle/map/UdpChannelReplicator.java | 2 + .../chronicle/map/UdpReplicationConfig.java | 8 +- .../map/UdpReplicationConfigBean.java | 89 +++++ .../NodeDiscoveryHostPortBroadcasterTest.java | 31 ++ .../map/UDPSocketReplicationTest.java | 5 +- 13 files changed, 693 insertions(+), 30 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/NodeDiscoveryHostPortBroadcaster.java create mode 100644 src/main/java/net/openhft/chronicle/map/TcpReplicationConfigBean.java create mode 100644 src/main/java/net/openhft/chronicle/map/ThrottlingConfigBean.java create mode 100644 src/main/java/net/openhft/chronicle/map/UdpReplicationConfigBean.java create mode 100644 src/test/java/net/openhft/chronicle/map/NodeDiscoveryHostPortBroadcasterTest.java diff --git a/pom.xml b/pom.xml index 68acc6ecd..487fb1653 100755 --- a/pom.xml +++ b/pom.xml @@ -89,10 +89,6 @@ slf4j-api - - com.google.auto.value - auto-value - diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index dd7b30e2b..a481397fe 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -136,9 +136,6 @@ abstract class AbstractChannelReplicator implements Closeable { throttler.add(channel); } - interface ChannelReplicatorBuilder { - } - /** * throttles 'writes' to ensure the network is not swamped, this is achieved by periodically * de-registering the write selector during periods of high volume. @@ -211,9 +208,9 @@ abstract class AbstractChannelReplicator implements Closeable { if (bytesWritten > maxBytesInInterval) { for (SelectableChannel channel : channels) { final SelectionKey selectionKey = channel.keyFor(selector); - if (selectionKey != null) + if (selectionKey != null) { selectionKey.interestOps(selectionKey.interestOps() & ~OP_WRITE); - + } if (LOG.isDebugEnabled()) LOG.debug("Throttling UDP writes"); } diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscoveryHostPortBroadcaster.java b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryHostPortBroadcaster.java new file mode 100644 index 000000000..580d998c4 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscoveryHostPortBroadcaster.java @@ -0,0 +1,345 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.collection.ATSDirectBitSet; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; +import org.jetbrains.annotations.NotNull; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.DatagramChannel; +import java.util.HashSet; +import java.util.Set; + +/** + * @author Rob Austin. + */ +public class NodeDiscoveryHostPortBroadcaster extends UdpChannelReplicator { + + private static final byte UNUSED = (byte) -1; + + static final ByteBufferBytes BOOTSTRAP_BYTES; + + + static { + final String BOOTSTRAP = "BOOTSTRAP"; + BOOTSTRAP_BYTES = new ByteBufferBytes(ByteBuffer.allocate(2 + BOOTSTRAP.length())); + BOOTSTRAP_BYTES.write((short) BOOTSTRAP.length()); + BOOTSTRAP_BYTES.append(BOOTSTRAP); + + } + + /** + * @param replicationConfig + * @throws java.io.IOException + */ + NodeDiscoveryHostPortBroadcaster( + @NotNull final UdpReplicationConfig replicationConfig, + final int serializedEntrySize, + RemoteNodes remoteNodes) + throws IOException { + + super(replicationConfig, serializedEntrySize, UNUSED); + + + BytesExternalizable externalizable = new BytesExternalizableImpl(remoteNodes, this); + + setReader(new UdpSocketChannelEntryReader(serializedEntrySize, externalizable)); + + setWriter(new UdpSocketChannelEntryWriter(serializedEntrySize, externalizable, this)); + + start(); + } + + 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 + */ + UdpSocketChannelEntryReader(final int serializedEntrySize, + @NotNull final BytesExternalizable externalizable) { + // we make the buffer twice as large just to give ourselves headroom + in = ByteBuffer.allocateDirect(serializedEntrySize * 2); + + out = new ByteBufferBytes(in); + out.limit(0); + in.clear(); + this.externalizable = externalizable; + } + + /** + * reads entries from the socket till it is empty + * + * @param socketChannel the socketChannel that we will read from + * @throws IOException + * @throws InterruptedException + */ + public void readAll(@NotNull final DatagramChannel socketChannel) throws IOException, + InterruptedException { + + out.clear(); + in.clear(); + + socketChannel.receive(in); + + final int bytesRead = in.position(); + + if (bytesRead < SIZE_OF_SHORT + SIZE_OF_SHORT) + return; + + out.limit(in.position()); + + final short invertedSize = out.readShort(); + final int size = out.readUnsignedShort(); + + // check the the first 4 bytes are the inverted len followed by the len + // we do this to check that this is a valid start of entry, otherwise we throw it away + if (((short) ~size) != invertedSize) + return; + + if (out.remaining() != size) + return; + + externalizable.readExternalBytes(out); + } + + } + + private static class UdpSocketChannelEntryWriter implements EntryWriter { + + + private final ByteBuffer out; + private final ByteBufferBytes in; + + private final BytesExternalizable externalizable; + private UdpChannelReplicator udpReplicator; + + UdpSocketChannelEntryWriter(final int serializedEntrySize, + @NotNull final BytesExternalizable externalizable, + @NotNull final UdpChannelReplicator udpReplicator) { + this.udpReplicator = udpReplicator; + + // we make the buffer twice as large just to give ourselves headroom + out = ByteBuffer.allocateDirect(serializedEntrySize * 2); + in = new ByteBufferBytes(out); + + // write bootstrap to the 'in' buffer + { + // skip 2 bytes where we write the size + in.skip(SIZE_OF_SHORT); + + in.write(BOOTSTRAP_BYTES); + + // now write the size at the start + in.writeShort(0, (int) in.position() - SIZE_OF_SHORT); + + } + + this.externalizable = externalizable; + } + + + /** + * writes all the entries that have changed, to the tcp socket + * + * @param socketChannel + * @param modificationIterator + * @throws InterruptedException + * @throws java.io.IOException + */ + + /** + * update that are throttled are rejected. + * + * @param socketChannel the socketChannel that we will write to + * @throws InterruptedException + * @throws IOException + */ + public int writeAll(@NotNull final DatagramChannel socketChannel) + throws InterruptedException, IOException { + + out.clear(); + in.clear(); + in.skip(SIZE_OF_SHORT); + + // we'll write the size inverted at the start + in.writeShort(0, ~(in.readUnsignedShort(SIZE_OF_SHORT))); + out.limit((int) in.position()); + + int start = out.position(); + + // writes the contents of + externalizable.writeExternalBytes(in); + udpReplicator.disableWrites(); + // number of bytes written + return out.position() - start; + + + } + } + + +} + +/** + * supports reading and writing serialize entries + */ +interface BytesExternalizable { + + /** + * The map implements this method to save its contents. + * + * @param destination a buffer the entry will be written to, the segment may reject this operation and add + * zeroBytes, if the identifier in the entry did not match the maps local + */ + void writeExternalBytes(@NotNull Bytes destination); + + /** + * The map implements this method to restore its contents. This method must read the values in the same + * sequence and with the same types as were written by {@code writeExternalEntry()}. This method is + * typically called when we receive a remote replication event, this event could originate from either a + * remote {@code put(K key, V value)} or {@code remove(Object key)} + * + * @param source bytes to read an entry from + */ + void readExternalBytes(@NotNull Bytes source); + + +} + +class RemoteNodes { + + + private final Bytes activeIdentifiersBitSetBytes; + private Set inetSocketAddresses; + private final ATSDirectBitSet atsDirectBitSet; + + + /** + * @param inetSocketAddresses the host and ports which we should connect to for TCP replication + * @param activeIdentifiersBitSetBytes byte sof a bitset containing the known identifiers + */ + RemoteNodes(final Bytes activeIdentifiersBitSetBytes) { + this.inetSocketAddresses = new HashSet(); + this.activeIdentifiersBitSetBytes = activeIdentifiersBitSetBytes; + this.atsDirectBitSet = new ATSDirectBitSet(this.activeIdentifiersBitSetBytes); + } + + + public Set inetSocketAddresses() { + return inetSocketAddresses; + } + + public Bytes activeIdentifierBytes() { + return activeIdentifiersBitSetBytes; + } + + public void add(InetSocketAddress inetSocketAddress) { + inetSocketAddresses.add(inetSocketAddress); + } + + public ATSDirectBitSet activeIdentifierBitSet() { + return atsDirectBitSet; + } +} + +class BytesExternalizableImpl implements BytesExternalizable { + + private final RemoteNodes allNodes; + private final Replica.ModificationNotifier modificationNotifier; + //private final InetSocketAddress localInetSocketAddress; + + public BytesExternalizableImpl(final RemoteNodes allNodes, + final Replica.ModificationNotifier modificationNotifier) { + this.allNodes = allNodes; + this.modificationNotifier = modificationNotifier; + // todo this should only be added once we have connected - we will add our host and port to allNodes + //this.localInetSocketAddress = new InetSocketAddress(localHost, localPort); + //this.allNodes.add(localInetSocketAddress); + + } + + /** + * this is used to tell nodes that are connecting to us which host and ports are in our grid, along with + * all the identifiers. + * + * @param destination a buffer the entry will be written to, the segment may reject this operation and add + * zeroBytes, if the identifier in the entry did not match the maps local + */ + @Override + public void writeExternalBytes(@NotNull Bytes destination) { + + final Set inetSocketAddresses = allNodes.inetSocketAddresses(); + + // write the number of hosts and ports + destination.write((short) inetSocketAddresses.size()); + + // write all the host and ports + for (InetSocketAddress inetSocketAddress : inetSocketAddresses) { + destination.writeUTF(inetSocketAddress.getHostName()); + destination.writeInt(inetSocketAddress.getPort()); + } + + destination.write(allNodes.activeIdentifierBytes()); + } + + @Override + public void readExternalBytes(@NotNull Bytes source) { + + if (source.equals(NodeDiscoveryHostPortBroadcaster.BOOTSTRAP_BYTES)) { + // used to wake up the OP_WRITE + modificationNotifier.onChange(); + return; + } + + // the number of host and ports + int count = source.readShort(); + + for (int i = 0; i < count; i++) { + + final String host = source.readUTF(); + final int port = source.readInt(); + + allNodes.add(new InetSocketAddress(host, port)); + } + + int identifierBitSetBytes = source.readShort(); + source.limit(source.position() + identifierBitSetBytes); + + final ATSDirectBitSet sourceBitSet = new ATSDirectBitSet(source); + + final ATSDirectBitSet resultBitSet = allNodes.activeIdentifierBitSet(); + + // merges the source into the destination and returns the destination + orBitSets(sourceBitSet, resultBitSet); + + + } + + /** + * merges the source into the destination and returns the destination + * + * @param source + * @param destination + * @return + */ + private ATSDirectBitSet orBitSets(ATSDirectBitSet source, ATSDirectBitSet destination) { + // merge the two bit-sets together via 'or' + + long numberOfLongs = source.size() / 8; + for (int longIndex = 1; longIndex < numberOfLongs; longIndex++) { + destination.or(longIndex, source.getLong(longIndex)); + } + + return destination; + } + + +} \ No newline at end of file diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java b/src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java index 87d31d036..7b205d8a7 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java @@ -223,7 +223,7 @@ public final class ReplicatingCluster implements Closeable { final int chronicleChannel = bytes.readUnsignedShort(); final long lastModificationTime = bytes.readLong(); - // this could be null if once node has a chronicle channel before the other + // this could be null if one node has a chronicle channel before the other if (chronicleChannels[chronicleChannel] != null) { chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier, NOP) .dirtyEntries(lastModificationTime); @@ -315,6 +315,9 @@ public final class ReplicatingCluster implements Closeable { final PayloadProvider iterator = new PayloadProvider() { + + // todo it maybe possibel to replace this with a single atomic refernce, + // todo this way no aditional objects will be created final Queue payloads = new LinkedTransferQueue(); @Override diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java index cf07c522d..cedeb321b 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java @@ -16,8 +16,6 @@ package net.openhft.chronicle.map; -import com.google.auto.value.AutoValue; - import java.net.InetSocketAddress; import java.util.Collection; import java.util.HashSet; @@ -28,7 +26,7 @@ import static java.util.Arrays.asList; import static java.util.Collections.unmodifiableSet; import static java.util.concurrent.TimeUnit.SECONDS; -@AutoValue + public abstract class TcpReplicationConfig { private static final int DEFAULT_PACKET_SIZE = 1024 * 8; @@ -65,7 +63,7 @@ public abstract class TcpReplicationConfig { static TcpReplicationConfig create(int serverPort, Set endpoints, int packetSize, ThrottlingConfig throttlingConfig, long heartBeatInterval, TimeUnit heartBeatIntervalUnit) { - return new AutoValue_TcpReplicationConfig(serverPort, endpoints, packetSize, + return new TcpReplicationConfigBean(serverPort, endpoints, packetSize, throttlingConfig, heartBeatInterval, heartBeatIntervalUnit); } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfigBean.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfigBean.java new file mode 100644 index 000000000..dd8c676f0 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfigBean.java @@ -0,0 +1,116 @@ +package net.openhft.chronicle.map; + +import java.net.InetSocketAddress; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@javax.annotation.Generated("com.google.auto.value.processor.AutoValueProcessor") +final class TcpReplicationConfigBean extends TcpReplicationConfig { + private final int serverPort; + private final Set endpoints; + private final int packetSize; + private final ThrottlingConfig throttlingConfig; + private final long heartBeatInterval; + private final TimeUnit heartBeatIntervalUnit; + + TcpReplicationConfigBean( + int serverPort, + Set endpoints, + int packetSize, + ThrottlingConfig throttlingConfig, + long heartBeatInterval, + TimeUnit heartBeatIntervalUnit) { + this.serverPort = serverPort; + if (endpoints == null) { + throw new NullPointerException("Null endpoints"); + } + this.endpoints = endpoints; + this.packetSize = packetSize; + if (throttlingConfig == null) { + throw new NullPointerException("Null throttlingConfig"); + } + this.throttlingConfig = throttlingConfig; + this.heartBeatInterval = heartBeatInterval; + if (heartBeatIntervalUnit == null) { + throw new NullPointerException("Null heartBeatIntervalUnit"); + } + this.heartBeatIntervalUnit = heartBeatIntervalUnit; + } + + @Override + public int serverPort() { + return serverPort; + } + + @Override + public Set endpoints() { + return endpoints; + } + + @Override + public int packetSize() { + return packetSize; + } + + @Override + public ThrottlingConfig throttlingConfig() { + return throttlingConfig; + } + + @Override + long heartBeatInterval() { + return heartBeatInterval; + } + + @Override + TimeUnit heartBeatIntervalUnit() { + return heartBeatIntervalUnit; + } + + @Override + public String toString() { + return "TcpReplicationConfig{" + + "serverPort=" + serverPort + + ", endpoints=" + endpoints + + ", packetSize=" + packetSize + + ", throttlingConfig=" + throttlingConfig + + ", heartBeatInterval=" + heartBeatInterval + + ", heartBeatIntervalUnit=" + heartBeatIntervalUnit + + "}"; + } + + @Override + public boolean equals(Object o) { + if (o == this) { + return true; + } + if (o instanceof TcpReplicationConfig) { + TcpReplicationConfig that = (TcpReplicationConfig) o; + return (this.serverPort == that.serverPort()) + && (this.endpoints.equals(that.endpoints())) + && (this.packetSize == that.packetSize()) + && (this.throttlingConfig.equals(that.throttlingConfig())) + && (this.heartBeatInterval == that.heartBeatInterval()) + && (this.heartBeatIntervalUnit.equals(that.heartBeatIntervalUnit())); + } + return false; + } + + @Override + public int hashCode() { + int h = 1; + h *= 1000003; + h ^= serverPort; + h *= 1000003; + h ^= endpoints.hashCode(); + h *= 1000003; + h ^= packetSize; + h *= 1000003; + h ^= throttlingConfig.hashCode(); + h *= 1000003; + h ^= (heartBeatInterval >>> 32) ^ heartBeatInterval; + h *= 1000003; + h ^= heartBeatIntervalUnit.hashCode(); + return h; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/ThrottlingConfig.java b/src/main/java/net/openhft/chronicle/map/ThrottlingConfig.java index 46b7e0a77..50a1a9383 100644 --- a/src/main/java/net/openhft/chronicle/map/ThrottlingConfig.java +++ b/src/main/java/net/openhft/chronicle/map/ThrottlingConfig.java @@ -16,13 +16,12 @@ package net.openhft.chronicle.map; -import com.google.auto.value.AutoValue; import java.util.concurrent.TimeUnit; import static java.util.concurrent.TimeUnit.MILLISECONDS; -@AutoValue + public abstract class ThrottlingConfig { private static final long DEFAULT_BUCKET_INTERVAL = 100L; @@ -48,8 +47,8 @@ public abstract class ThrottlingConfig { } /** - * Returns a config which throttle the specified max bits per the given time unit - * with default bucketing interval. + * Returns a config which throttle the specified max bits per the given time unit with default bucketing + * interval. * * @param maxBits the preferred maximum bits * @param perUnit the time unit per which maximum bits specified @@ -73,7 +72,7 @@ public abstract class ThrottlingConfig { static ThrottlingConfig create( long throttle, TimeUnit throttlePerUnit, long throttleBucketInterval, TimeUnit throttleBucketIntervalUnit) { - return new AutoValue_ThrottlingConfig(throttle, throttlePerUnit, + return new ThrottlingConfigBean(throttle, throttlePerUnit, throttleBucketInterval, throttleBucketIntervalUnit); } @@ -82,8 +81,8 @@ public abstract class ThrottlingConfig { abstract TimeUnit throttlePerUnit(); /** - * Returns maximum bits per the given time unit, i. e. the throttling. {@code 0} (zero) - * designates there is no throttling. + * Returns maximum bits per the given time unit, i. e. the throttling. {@code 0} (zero) designates there + * is no throttling. * * @param perUnit maximum bits is returned per this time unit * @return maximum bits per the given time unit, or {@code 0} if there is no throttling @@ -97,9 +96,8 @@ public abstract class ThrottlingConfig { abstract TimeUnit bucketIntervalUnit(); /** - * Returns the throttle bucketing interval in the given time units. - *

- *

Default throttle bucketing interval is 100 millis. + * Returns the throttle bucketing interval in the given time units.

Default throttle bucketing + * interval is 100 millis. * * @param unit the time unit of the interval * @return the bucketing interval for throttling in the given time units diff --git a/src/main/java/net/openhft/chronicle/map/ThrottlingConfigBean.java b/src/main/java/net/openhft/chronicle/map/ThrottlingConfigBean.java new file mode 100644 index 000000000..8d67b336e --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/ThrottlingConfigBean.java @@ -0,0 +1,87 @@ +package net.openhft.chronicle.map; + +import java.util.concurrent.TimeUnit; + +@javax.annotation.Generated("com.google.auto.value.processor.AutoValueProcessor") +final class ThrottlingConfigBean extends ThrottlingConfig { + private final long throttle; + private final TimeUnit throttlePerUnit; + private final long bucketInterval; + private final TimeUnit bucketIntervalUnit; + + ThrottlingConfigBean( + long throttle, + TimeUnit throttlePerUnit, + long bucketInterval, + TimeUnit bucketIntervalUnit) { + this.throttle = throttle; + if (throttlePerUnit == null) { + throw new NullPointerException("Null throttlePerUnit"); + } + this.throttlePerUnit = throttlePerUnit; + this.bucketInterval = bucketInterval; + if (bucketIntervalUnit == null) { + throw new NullPointerException("Null bucketIntervalUnit"); + } + this.bucketIntervalUnit = bucketIntervalUnit; + } + + @Override + long throttle() { + return throttle; + } + + @Override + TimeUnit throttlePerUnit() { + return throttlePerUnit; + } + + @Override + long bucketInterval() { + return bucketInterval; + } + + @Override + TimeUnit bucketIntervalUnit() { + return bucketIntervalUnit; + } + + @Override + public String toString() { + return "ThrottlingConfig{" + + "throttle=" + throttle + + ", throttlePerUnit=" + throttlePerUnit + + ", bucketInterval=" + bucketInterval + + ", bucketIntervalUnit=" + bucketIntervalUnit + + "}"; + } + + @Override + public boolean equals(Object o) { + if (o == this) { + return true; + } + if (o instanceof ThrottlingConfig) { + ThrottlingConfig that = (ThrottlingConfig) o; + return (this.throttle == that.throttle()) + && (this.throttlePerUnit.equals(that.throttlePerUnit())) + && (this.bucketInterval == that.bucketInterval()) + && (this.bucketIntervalUnit.equals(that.bucketIntervalUnit())); + } + return false; + } + + @Override + public int hashCode() { + int h = 1; + h *= 1000003; + h ^= (throttle >>> 32) ^ throttle; + h *= 1000003; + h ^= throttlePerUnit.hashCode(); + h *= 1000003; + h ^= (bucketInterval >>> 32) ^ bucketInterval; + h *= 1000003; + h ^= bucketIntervalUnit.hashCode(); + return h; + } +} diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java index 5d6b39548..2ab0ba098 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -209,6 +209,8 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. final SelectionKey selectionKey = writeChannel.keyFor(this.selector); if (selectionKey != null) selectionKey.interestOps(selectionKey.interestOps() | OP_WRITE); + // we have just enabled it, so don't have to do this again. + shouldEnableOpWrite = false; } catch (Exception e) { LOG.error("", e); } diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java index 13725a13d..0e02d9e9d 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java @@ -16,14 +16,13 @@ package net.openhft.chronicle.map; -import com.google.auto.value.AutoValue; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import java.net.InetAddress; import java.net.NetworkInterface; -@AutoValue + public abstract class UdpReplicationConfig { /** @@ -36,6 +35,9 @@ public abstract class UdpReplicationConfig { public static UdpReplicationConfig simple(@NotNull InetAddress address, int port) { if (address.isMulticastAddress()) throw new IllegalArgumentException(); + + + return create(address, port, null, ThrottlingConfig.noThrottling()); } @@ -49,7 +51,7 @@ public abstract class UdpReplicationConfig { static UdpReplicationConfig create(InetAddress address, int port, NetworkInterface networkInterface, ThrottlingConfig throttlingConfig) { - return new AutoValue_UdpReplicationConfig(address, port, networkInterface, + return new UdpReplicationConfigBean(address, port, networkInterface, throttlingConfig); } diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicationConfigBean.java b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfigBean.java new file mode 100644 index 000000000..441aa4eac --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfigBean.java @@ -0,0 +1,89 @@ +package net.openhft.chronicle.map; + +import java.net.InetAddress; +import java.net.NetworkInterface; + + +final class UdpReplicationConfigBean extends UdpReplicationConfig { + private final InetAddress address; + private final int port; + private final NetworkInterface networkInterface; + private final ThrottlingConfig throttlingConfig; + + UdpReplicationConfigBean( + InetAddress address, + int port, + NetworkInterface networkInterface, + ThrottlingConfig throttlingConfig) { + if (address == null) { + throw new NullPointerException("Null address"); + } + this.address = address; + this.port = port; + + this.networkInterface = networkInterface; + if (throttlingConfig == null) { + throw new NullPointerException("Null throttlingConfig"); + } + this.throttlingConfig = throttlingConfig; + } + + @Override + public InetAddress address() { + return address; + } + + @Override + public int port() { + return port; + } + + @Override + public NetworkInterface networkInterface() { + return networkInterface; + } + + @Override + public ThrottlingConfig throttlingConfig() { + return throttlingConfig; + } + + @Override + public String toString() { + return "UdpReplicationConfig{" + + "address=" + address + + ", port=" + port + + ", networkInterface=" + networkInterface + + ", throttlingConfig=" + throttlingConfig + + "}"; + } + + @Override + public boolean equals(Object o) { + if (o == this) { + return true; + } + if (o instanceof UdpReplicationConfig) { + UdpReplicationConfig that = (UdpReplicationConfig) o; + return (this.address.equals(that.address())) + && (this.port == that.port()) + && (this.networkInterface.equals(that.networkInterface())) + && (this.throttlingConfig.equals(that.throttlingConfig())); + } + return false; + } + + @Override + public int hashCode() { + int h = 1; + h *= 1000003; + h ^= address.hashCode(); + h *= 1000003; + h ^= port; + h *= 1000003; + h ^= networkInterface.hashCode(); + h *= 1000003; + h ^= throttlingConfig.hashCode(); + return h; + } +} diff --git a/src/test/java/net/openhft/chronicle/map/NodeDiscoveryHostPortBroadcasterTest.java b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryHostPortBroadcasterTest.java new file mode 100644 index 000000000..2797273f8 --- /dev/null +++ b/src/test/java/net/openhft/chronicle/map/NodeDiscoveryHostPortBroadcasterTest.java @@ -0,0 +1,31 @@ +package net.openhft.chronicle.map; + +import junit.framework.TestCase; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; +import org.junit.Test; + +import java.io.IOException; +import java.net.Inet4Address; +import java.nio.ByteBuffer; + +public class NodeDiscoveryHostPortBroadcasterTest extends TestCase { + + @Test + 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); + + final NodeDiscoveryHostPortBroadcaster nodeDiscoveryHostPortBroadcaster + = new NodeDiscoveryHostPortBroadcaster(udpConfig, 1024, remoteNodes); + + Thread.sleep(5000); + } + + +} \ No newline at end of file diff --git a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java index 71690a2b5..3d14cb7fd 100644 --- a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java @@ -18,7 +18,6 @@ package net.openhft.chronicle.map; import org.junit.After; import org.junit.Before; -import org.junit.Ignore; import org.junit.Test; import java.io.Closeable; @@ -43,8 +42,8 @@ public class UDPSocketReplicationTest { UdpReplicationConfig udpConfig = UdpReplicationConfig .simple(Inet4Address.getByName("255.255.255.255"), udpPort); - return ChronicleMapBuilder.of(Integer.class, CharSequence.class) + return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) .addReplicator(udp((byte) identifier, udpConfig)) .create(Builder.getPersistenceFile()); @@ -75,7 +74,7 @@ public class UDPSocketReplicationTest { @Test - @Ignore + // @Ignore public void testBufferOverflow() throws IOException, InterruptedException { for (int i = 0; i < 1024; i++) {