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

Renamed IdentifierListener to RemoteNodeValidator and moved this configuration from TcpConfig to Replication builders

Browse Source
This commit is contained in:
leventov
2014-11-10 07:53:30 +03:00
parent 49b51d151d
commit 6962919732
13 changed files with 101 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
+4 -3
View File
@@ -410,7 +410,8 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* <p>This 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<K, C extends ChronicleHash,
* @param file the file with existing hash container or a desired location of a new off-heap
* persisted hash container
* @return a hash container mapped to the given file
* @throws IOException if any IO error, related to off-heap memory allocation or file mapping
* occurs
* @throws IOException if any IO error, related to off-heap memory allocation or file mapping,
* or establishing replication connections, occurs
* @see ChronicleHash#file()
* @see ChronicleHash#close()
* @see #create()
src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java
+41 -6
View File
@@ -16,17 +16,24 @@
package net.openhft.chronicle.hash.replication;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
class AbstractReplication {
/**
* Common configurations of {@link SingleChronicleHashReplication} and {@link ReplicationHub}.
*/
public abstract class AbstractReplication {
private final byte localIdentifier;
private final TcpTransportAndNetworkConfig tcpConfig;
private final UdpTransportConfig udpConfig;
private final @Nullable TcpTransportAndNetworkConfig tcpConfig;
private final @Nullable UdpTransportConfig udpConfig;
private final @Nullable RemoteNodeValidator remoteNodeValidator;
// package-private to forbid subclassing from outside of the package
AbstractReplication(byte localIdentifier, Builder builder) {
this.localIdentifier = localIdentifier;
tcpConfig = builder.tcpConfig;
udpConfig = builder.udpConfig;
remoteNodeValidator = builder.remoteNodeValidator;
}
public byte identifier() {
@@ -43,20 +50,48 @@ class AbstractReplication {
return udpConfig;
}
static abstract class Builder<B extends Builder> {
@Nullable
public RemoteNodeValidator remoteNodeValidator() {
return remoteNodeValidator;
}
/**
* Builder of {@link AbstractReplication} configurations.
*
* @param <R> the concrete {@link AbstractReplication} subclass: {@link
* SingleChronicleHashReplication} or {@link ReplicationHub}
* @param <B> the concrete builder subclass: {@link SingleChronicleHashReplication.Builder} or
* {@link ReplicationHub.Builder}
*/
public static abstract class Builder<R extends AbstractReplication, B extends Builder<R, B>> {
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(
src/main/java/net/openhft/chronicle/hash/replication/IdentifierListener.java → src/main/java/net/openhft/chronicle/hash/replication/RemoteNodeValidator.java
+2 -2
View File
@@ -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);
}
src/main/java/net/openhft/chronicle/hash/replication/ReplicationChannel.java
+1
View File
@@ -16,6 +16,7 @@
package net.openhft.chronicle.hash.replication;
public final class ReplicationChannel {
private ReplicationHub hub;
private final short channelId;
src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
+8 -14
View File
@@ -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<Builder> {
public static final class Builder extends AbstractReplication.Builder<ReplicationHub, 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);
}
src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java
+7 -11
View File
@@ -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<Builder> {
public static final class Builder
extends AbstractReplication.Builder<SingleChronicleHashReplication, 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);
}
}
}
src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java
+8 -24
View File
@@ -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<InetSocketAddress> 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<InetSocketAddress> 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);
}
src/main/java/net/openhft/chronicle/hash/replication/ThrottlingConfig.java
+1 -1
View File
@@ -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;
src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java
+1 -1
View File
@@ -899,7 +899,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
List<Replicator> 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 {
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+1 -1
View File
@@ -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);
}
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+10 -6
View File
@@ -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<Byte, SocketAddress> identifiers = new ConcurrentHashMap<Byte,
SocketAddress>();
@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();
}
src/main/java/net/openhft/chronicle/map/Replicators.java
+8 -3
View File
@@ -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());
}
};
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+9 -6
View File
@@ -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,