diff --git a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java index 924b95a5f..c34fe9906 100644 --- a/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java @@ -410,7 +410,8 @@ public interface ChronicleHashBuilderThis method is a shortcut for {@code instance().create()}. * * @return a new off-heap hash container - * @throws IOException if any IO error relates to off-heap memory allocation occurs + * @throws IOException if any IO error relates to off-heap memory allocation, + * or establishing replication connections, occurs * @see #createPersistedTo(File) * @see #instance() */ @@ -436,8 +437,8 @@ public interface ChronicleHashBuilder { + @Nullable + public RemoteNodeValidator remoteNodeValidator() { + return remoteNodeValidator; + } + + /** + * Builder of {@link AbstractReplication} configurations. + * + * @param the concrete {@link AbstractReplication} subclass: {@link + * SingleChronicleHashReplication} or {@link ReplicationHub} + * @param the concrete builder subclass: {@link SingleChronicleHashReplication.Builder} or + * {@link ReplicationHub.Builder} + */ + public static abstract class Builder> { private TcpTransportAndNetworkConfig tcpConfig = null; private UdpTransportConfig udpConfig = null; + private RemoteNodeValidator remoteNodeValidator = null; - public B tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) { + // package-private to forbid subclassing from outside of the package + Builder() {} + + @NotNull + public B tcpTransportAndNetwork(@Nullable TcpTransportAndNetworkConfig tcpConfig) { this.tcpConfig = tcpConfig; return (B) this; } - public B udpTransport(UdpTransportConfig udpConfig) { + @NotNull + public B udpTransport(@Nullable UdpTransportConfig udpConfig) { this.udpConfig = udpConfig; return (B) this; } + @NotNull + public B remoteNodeValidator(@Nullable RemoteNodeValidator remoteNodeValidator) { + this.remoteNodeValidator = remoteNodeValidator; + return (B) this; + } + + @NotNull + public abstract R createWithId(byte identifier); + void check() { if (udpConfig == null && tcpConfig == null) throw new IllegalStateException( diff --git a/src/main/java/net/openhft/chronicle/hash/replication/IdentifierListener.java b/src/main/java/net/openhft/chronicle/hash/replication/RemoteNodeValidator.java similarity index 89% rename from src/main/java/net/openhft/chronicle/hash/replication/IdentifierListener.java rename to src/main/java/net/openhft/chronicle/hash/replication/RemoteNodeValidator.java index 959f88d8e..a90fc1832 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/IdentifierListener.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/RemoteNodeValidator.java @@ -23,7 +23,7 @@ import java.net.SocketAddress; /** * @author Rob Austin. */ -public interface IdentifierListener { +public interface RemoteNodeValidator { /** * checks the identifier that is unique and we haven't seen it before, unless it comes from the same port @@ -33,6 +33,6 @@ public interface IdentifierListener { * @param remoteAddress remoteAddress * @return true if unique */ - boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress); + boolean validate(byte remoteIdentifier, SocketAddress remoteAddress); } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java index 72527c251..0ea18e674 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.hash.replication; + public final class ReplicationChannel { private ReplicationHub hub; private final short channelId; diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java index eb137ac1b..41bc3f135 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java @@ -16,10 +16,11 @@ package net.openhft.chronicle.hash.replication; -import java.io.IOException; +import org.jetbrains.annotations.NotNull; public final class ReplicationHub extends AbstractReplication { + @NotNull public static Builder builder() { return new Builder(); } @@ -55,34 +56,27 @@ public final class ReplicationHub extends AbstractReplication { return channel; } - public static final class Builder extends AbstractReplication.Builder { - + public static final class Builder extends AbstractReplication.Builder { private int maxEntrySize = 1024; private int maxNumberOfChronicles = 128; private Builder() {} - @Override - public Builder tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) { - return super.tcpTransportAndNetwork(tcpConfig); - } - - @Override - public Builder udpTransport(UdpTransportConfig udpConfig) { - return super.udpTransport(udpConfig); - } - + @NotNull public Builder maxEntrySize(int maxEntrySize) { this.maxEntrySize = maxEntrySize; return this; } + @NotNull public Builder maxNumberOfChannels(int maxNumberOfChannels) { this.maxNumberOfChronicles = maxNumberOfChannels; return this; } - public ReplicationHub createWithId(byte identifier) throws IOException { + @NotNull + @Override + public ReplicationHub createWithId(byte identifier) { check(); return new ReplicationHub(identifier, this); } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java b/src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java index 6dc8e058c..03640a894 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java @@ -16,8 +16,11 @@ package net.openhft.chronicle.hash.replication; +import org.jetbrains.annotations.NotNull; + public final class SingleChronicleHashReplication extends AbstractReplication { + @NotNull public static Builder builder() { return new Builder(); } @@ -26,22 +29,15 @@ public final class SingleChronicleHashReplication extends AbstractReplication { super(localIdentifier, builder); } - public static final class Builder extends AbstractReplication.Builder { + public static final class Builder + extends AbstractReplication.Builder { private Builder() {} + @NotNull + @Override public SingleChronicleHashReplication createWithId(byte identifier) { check(); return new SingleChronicleHashReplication(identifier, this); } - - @Override - public Builder tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) { - return super.tcpTransportAndNetwork(tcpConfig); - } - - @Override - public Builder udpTransport(UdpTransportConfig udpConfig) { - return super.udpTransport(udpConfig); - } } } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java index bfc277e75..878fc951c 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java @@ -39,14 +39,12 @@ public final class TcpTransportAndNetworkConfig { private final ThrottlingConfig throttlingConfig; private final long heartBeatInterval; private final TimeUnit heartBeatIntervalUnit; - private final IdentifierListener identifierListener; private TcpTransportAndNetworkConfig(int serverPort, Set endpoints, int packetSize, boolean autoReconnectedUponDroppedConnection, ThrottlingConfig throttlingConfig, long heartBeatInterval, - TimeUnit heartBeatIntervalUnit, - IdentifierListener identifierListener) { + TimeUnit heartBeatIntervalUnit) { this.serverPort = serverPort; this.endpoints = endpoints; this.packetSize = packetSize; @@ -54,10 +52,8 @@ public final class TcpTransportAndNetworkConfig { this.throttlingConfig = throttlingConfig; this.heartBeatInterval = heartBeatInterval; this.heartBeatIntervalUnit = heartBeatIntervalUnit; - this.identifierListener = identifierListener; } - public static TcpTransportAndNetworkConfig forSendingNode(int serverPort, InetSocketAddress... endpoints) { return forSendingNode(serverPort, Arrays.asList(endpoints)); @@ -87,19 +83,7 @@ public final class TcpTransportAndNetworkConfig { true, // autoReconnectedUponDroppedConnection ThrottlingConfig.noThrottling(), DEFAULT_HEART_BEAT_INTERVAL, - DEFAULT_HEART_BEAT_INTERVAL_UNIT, - null); // identifierListener - } - - public IdentifierListener nonUniqueIdentifierListener() { - return identifierListener; - } - - public TcpTransportAndNetworkConfig nonUniqueIdentifierListener( - final IdentifierListener identifierListener) { - return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, - autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, - heartBeatIntervalUnit, identifierListener); + DEFAULT_HEART_BEAT_INTERVAL_UNIT); } public boolean autoReconnectedUponDroppedConnection() { @@ -110,7 +94,7 @@ public final class TcpTransportAndNetworkConfig { boolean autoReconnectedUponDroppedConnection) { return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, - heartBeatIntervalUnit, identifierListener); + heartBeatIntervalUnit); } public ThrottlingConfig throttlingConfig() { @@ -121,7 +105,7 @@ public final class TcpTransportAndNetworkConfig { ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "TCP"); return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, - heartBeatIntervalUnit, identifierListener); + heartBeatIntervalUnit); } public long heartBeatInterval(TimeUnit unit) { @@ -135,7 +119,7 @@ public final class TcpTransportAndNetworkConfig { public TcpTransportAndNetworkConfig serverPort(int serverPort) { return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, - heartBeatIntervalUnit, identifierListener); + heartBeatIntervalUnit); } public Set endpoints() { @@ -150,7 +134,7 @@ public final class TcpTransportAndNetworkConfig { } return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, - heartBeatIntervalUnit, identifierListener); + heartBeatIntervalUnit); } public int packetSize() { @@ -162,14 +146,14 @@ public final class TcpTransportAndNetworkConfig { throw new IllegalArgumentException(); return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, - heartBeatIntervalUnit, identifierListener); + heartBeatIntervalUnit); } public TcpTransportAndNetworkConfig heartBeatInterval(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) { return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize, autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval, - heartBeatIntervalUnit, identifierListener); + heartBeatIntervalUnit); } diff --git a/src/main/java/net/openhft/chronicle/hash/replication/ThrottlingConfig.java b/src/main/java/net/openhft/chronicle/hash/replication/ThrottlingConfig.java index c9c410d45..953727262 100644 --- a/src/main/java/net/openhft/chronicle/hash/replication/ThrottlingConfig.java +++ b/src/main/java/net/openhft/chronicle/hash/replication/ThrottlingConfig.java @@ -23,7 +23,7 @@ import java.util.concurrent.TimeUnit; import static java.util.concurrent.TimeUnit.MILLISECONDS; -public class ThrottlingConfig { +public final class ThrottlingConfig { private static final long DEFAULT_BUCKET_INTERVAL = 100L; private static final TimeUnit DEFAULT_BUCKET_INTERVAL_UNIT = MILLISECONDS; diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java index f555e49b5..333d006c5 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java @@ -899,7 +899,7 @@ public abstract class AbstractChronicleMapBuilder replicators = new ArrayList<>(2); if (singleHashReplication != null) { if (singleHashReplication.tcpTransportAndNetwork() != null) - replicators.add(Replicators.tcp(singleHashReplication.tcpTransportAndNetwork())); + replicators.add(Replicators.tcp(singleHashReplication)); if (singleHashReplication.udpTransport() != null) replicators.add(Replicators.udp(singleHashReplication.udpTransport())); } else { diff --git a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java index cd039e325..9d37dfa06 100644 --- a/src/main/java/net/openhft/chronicle/map/ChannelProvider.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java @@ -63,7 +63,7 @@ final class ChannelProvider implements Closeable { final TcpReplicator tcpReplicator = new TcpReplicator( channelProvider.asReplica, channelProvider.asEntryExternalizable, - tcpConfig, hub.maxEntrySize(), null); + tcpConfig, hub.maxEntrySize(), null, hub.remoteNodeValidator()); channelProvider.add(tcpReplicator); } diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index bb181fd9a..eadb51821 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -18,7 +18,8 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.IdentifierListener; +import net.openhft.chronicle.hash.replication.RemoteNodeValidator; +import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.UdpTransportConfig; import net.openhft.lang.collection.ATSDirectBitSet; @@ -203,13 +204,13 @@ public class NodeDiscovery { // we should make a local copy as this may change - final IdentifierListener identifierListener = new IdentifierListener() { + final RemoteNodeValidator remoteNodeValidator = new RemoteNodeValidator() { private final ConcurrentMap identifiers = new ConcurrentHashMap(); @Override - public boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress) { + public boolean validate(byte remoteIdentifier, SocketAddress remoteAddress) { final SocketAddress lastKnownAddress = identifiers.putIfAbsent(remoteIdentifier, remoteAddress); if (remoteAddress.equals(lastKnownAddress)) return true; @@ -224,8 +225,7 @@ public class NodeDiscovery { final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .forUnknownTopology(ourAddressAndPort.getPort(), toInetSocketCollection(knownHostPorts)) - .heartBeatInterval(1, SECONDS) - .nonUniqueIdentifierListener(identifierListener); + .heartBeatInterval(1, SECONDS); LOG.info("Using Remote identifier=" + identifier); @@ -234,7 +234,11 @@ public class NodeDiscovery { return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) - .replication(identifier, tcpConfig).create(); + .replication(SingleChronicleHashReplication.builder() + .tcpTransportAndNetwork(tcpConfig) + .remoteNodeValidator(remoteNodeValidator) + .createWithId(identifier)) + .create(); } diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index 14c76b778..4664d931b 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -16,6 +16,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.hash.replication.AbstractReplication; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.UdpTransportConfig; import org.jetbrains.annotations.NotNull; @@ -36,7 +37,7 @@ final class Replicators { private Replicators() { } - static Replicator tcp(final TcpTransportAndNetworkConfig replicationConfig) { + static Replicator tcp(final AbstractReplication replication) { return new Replicator() { @Override @@ -53,8 +54,12 @@ final class Replicators { (keyValueSerializer, (VanillaChronicleMap) chronicleMap, builder.entrySize(true)); - return new TcpReplicator(replica, entryExternalizable, replicationConfig, - builder.entrySize(true), statelessServer); + TcpTransportAndNetworkConfig tcpConfig = replication.tcpTransportAndNetwork(); + assert tcpConfig != null; + return new TcpReplicator(replica, entryExternalizable, + tcpConfig, + builder.entrySize(true), statelessServer, + replication.remoteNodeValidator()); } }; } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index d57070057..c5b7c3e62 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -18,7 +18,7 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.hash.replication.IdentifierListener; +import net.openhft.chronicle.hash.replication.RemoteNodeValidator; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; import net.openhft.chronicle.hash.replication.ThrottlingConfig; import net.openhft.lang.io.ByteBufferBytes; @@ -76,6 +76,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable private final Replica.EntryExternalizable externalizable; private final TcpTransportAndNetworkConfig replicationConfig; private final StatelessServerConnector statelessServerConnector; + private final @Nullable RemoteNodeValidator remoteNodeValidator; private long selectorTimeout; @@ -91,7 +92,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable @NotNull final Replica.EntryExternalizable externalizable, @NotNull final TcpTransportAndNetworkConfig replicationConfig, final int maxEntrySizeBytes, - @Nullable StatelessServerConnector statelessServerConnector) throws IOException { + @Nullable StatelessServerConnector statelessServerConnector, + @Nullable RemoteNodeValidator remoteNodeValidator) throws IOException { super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(), maxEntrySizeBytes); @@ -110,6 +112,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable this.externalizable = externalizable; this.replicationConfig = replicationConfig; + this.remoteNodeValidator = remoteNodeValidator; + start(); } @@ -478,12 +482,11 @@ final 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 - final IdentifierListener identifierListener = - replicationConfig.nonUniqueIdentifierListener(); final SocketAddress remoteAddress = socketChannel.getRemoteAddress(); - if ((identifierListener != null && !identifierListener.isIdentifierUnique(remoteIdentifier, - remoteAddress)) || remoteIdentifier == localIdentifier) + if ((remoteNodeValidator != null && + !remoteNodeValidator.validate(remoteIdentifier, remoteAddress)) + || remoteIdentifier == localIdentifier) // 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,