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

Some renames in replication API, fixed potential issue with networkInterface configuration for udp multicast

Browse Source
This commit is contained in:
leventov
2014-11-08 05:26:10 +03:00
parent 0bcfa7ee43
commit c0d9099da9
29 changed files with 215 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/ChronicleHashBuilder.java
+10 -10
View File
@@ -18,8 +18,8 @@
package net.openhft.chronicle.hash;
import net.openhft.chronicle.hash.replication.SimpleReplication;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.chronicle.hash.serialization.*;
import net.openhft.chronicle.map.*;
@@ -237,7 +237,7 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
*
* @param timeProvider a new time provider for replication needs
* @return this builder back
* @see #replication(SimpleReplication)
* @see #replication(SingleChronicleHashReplication)
*/
B timeProvider(TimeProvider timeProvider);
@@ -376,14 +376,14 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* <p>By default, hash containers, created by this builder doesn't replicate their data.
*
* <p>This method call overrides all previous replication configurations of this builder, made
* either by this method or {@link #replication(byte, TcpConfig)} shortcut method.
* either by this method or {@link #replication(byte, TcpTransportAndNetworkConfig)} shortcut method.
*
* @param replication the replication config
* @return this builder back
* @see ChronicleHashInstanceConfig#replicated(SimpleReplication)
* @see #replication(byte, TcpConfig)
* @see ChronicleHashInstanceConfig#replicated(SingleChronicleHashReplication)
* @see #replication(byte, TcpTransportAndNetworkConfig)
*/
B replication(SimpleReplication replication);
B replication(SingleChronicleHashReplication replication);
/**
* Shortcut for {@code replication(SimpleReplication.builder()
@@ -392,10 +392,10 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* @param identifier the network-wide identifier of the containers, created by this builder
* @param tcpTransportAndNetwork configuration of tcp connection and network
* @return this builder back
* @see #replication(SimpleReplication)
* @see ChronicleHashInstanceConfig#replicated(byte, TcpConfig)
* @see #replication(SingleChronicleHashReplication)
* @see ChronicleHashInstanceConfig#replicated(byte, TcpTransportAndNetworkConfig)
*/
B replication(byte identifier, TcpConfig tcpTransportAndNetwork);
B replication(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork);
StatelessClientConfig<C> statelessClient(InetSocketAddress remoteAddress);
src/main/java/net/openhft/chronicle/hash/ChronicleHashInstanceConfig.java
+9 -9
View File
@@ -17,17 +17,17 @@
package net.openhft.chronicle.hash;
import net.openhft.chronicle.hash.replication.ReplicationChannel;
import net.openhft.chronicle.hash.replication.SimpleReplication;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import java.io.File;
import java.io.IOException;
public interface ChronicleHashInstanceConfig<C extends ChronicleHash> {
ChronicleHashInstanceConfig<C> replicated(byte identifier, TcpConfig tcpTransportAndNetwork);
ChronicleHashInstanceConfig<C> replicated(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork);
ChronicleHashInstanceConfig<C> replicated(SimpleReplication replication);
ChronicleHashInstanceConfig<C> replicated(SingleChronicleHashReplication replication);
/**
* Configures replication of the hash containers, created by this builder, via so called
@@ -35,19 +35,19 @@ public interface ChronicleHashInstanceConfig<C extends ChronicleHash> {
* <a href="https://github.com/OpenHFT/Chronicle-Map#channels-and-channelprovider">the
* section about Channels and ChannelProvider in ChronicleMap manual</a> for more information.
*
* <p>Another way to establish replication is {@link #replicated(SimpleReplication)} method
* or it's shortcut: {@link #replicated(byte, TcpConfig)}.
* <p>Another way to establish replication is {@link #replicated(SingleChronicleHashReplication)} method
* or it's shortcut: {@link #replicated(byte, TcpTransportAndNetworkConfig)}.
*
* <p>By default, hash containers, created by this builder doesn't replicate their data.
*
* <p>This method call overrides all previous replication configurations of this builder, made
* either by means of this method, {@link #replicated(SimpleReplication)}
* or {@link #replicated(byte, TcpConfig)} method calls.
* either by means of this method, {@link #replicated(SingleChronicleHashReplication)}
* or {@link #replicated(byte, TcpTransportAndNetworkConfig)} method calls.
*
* @param channel the channel responsible for gathering updates of hash containers,
* created by this builder, and replicating them over network
* @return this builder object back
* @see #replicated(SimpleReplication)
* @see #replicated(SingleChronicleHashReplication)
*/
ChronicleHashInstanceConfig<C> replicatedViaChannel(ReplicationChannel channel);
src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java
+8 -8
View File
@@ -20,8 +20,8 @@ import org.jetbrains.annotations.Nullable;
class AbstractReplication {
private final byte localIdentifier;
private final TcpConfig tcpConfig;
private final UdpConfig udpConfig;
private final TcpTransportAndNetworkConfig tcpConfig;
private final UdpTransportConfig udpConfig;
AbstractReplication(byte localIdentifier, Builder builder) {
this.localIdentifier = localIdentifier;
@@ -34,25 +34,25 @@ class AbstractReplication {
}
@Nullable
public TcpConfig tcpTransportAndNetwork() {
public TcpTransportAndNetworkConfig tcpTransportAndNetwork() {
return tcpConfig;
}
@Nullable
public UdpConfig udpTransport() {
public UdpTransportConfig udpTransport() {
return udpConfig;
}
static abstract class Builder<B extends Builder> {
private TcpConfig tcpConfig = null;
private UdpConfig udpConfig = null;
private TcpTransportAndNetworkConfig tcpConfig = null;
private UdpTransportConfig udpConfig = null;
public B tcpTransportAndNetwork(TcpConfig tcpConfig) {
public B tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) {
this.tcpConfig = tcpConfig;
return (B) this;
}
public B udpTransport(UdpConfig udpConfig) {
public B udpTransport(UdpTransportConfig udpConfig) {
this.udpConfig = udpConfig;
return (B) this;
}
src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
+2 -2
View File
@@ -63,12 +63,12 @@ public final class ReplicationHub extends AbstractReplication {
private Builder() {}
@Override
public Builder tcpTransportAndNetwork(TcpConfig tcpConfig) {
public Builder tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) {
return super.tcpTransportAndNetwork(tcpConfig);
}
@Override
public Builder udpTransport(UdpConfig udpConfig) {
public Builder udpTransport(UdpTransportConfig udpConfig) {
return super.udpTransport(udpConfig);
}
src/main/java/net/openhft/chronicle/hash/replication/SimpleReplication.java → src/main/java/net/openhft/chronicle/hash/replication/SingleChronicleHashReplication.java
+6 -7
View File
@@ -16,32 +16,31 @@
package net.openhft.chronicle.hash.replication;
//TODO better name
public final class SimpleReplication extends AbstractReplication {
public final class SingleChronicleHashReplication extends AbstractReplication {
public static Builder builder() {
return new Builder();
}
SimpleReplication(byte localIdentifier, Builder builder) {
SingleChronicleHashReplication(byte localIdentifier, Builder builder) {
super(localIdentifier, builder);
}
public static final class Builder extends AbstractReplication.Builder<Builder> {
private Builder() {}
public SimpleReplication create(byte identifier) {
public SingleChronicleHashReplication createWithId(byte identifier) {
check();
return new SimpleReplication(identifier, this);
return new SingleChronicleHashReplication(identifier, this);
}
@Override
public Builder tcpTransportAndNetwork(TcpConfig tcpConfig) {
public Builder tcpTransportAndNetwork(TcpTransportAndNetworkConfig tcpConfig) {
return super.tcpTransportAndNetwork(tcpConfig);
}
@Override
public Builder udpTransport(UdpConfig udpConfig) {
public Builder udpTransport(UdpTransportConfig udpConfig) {
return super.udpTransport(udpConfig);
}
}
src/main/java/net/openhft/chronicle/hash/replication/TcpConfig.java → src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java
+31 -25
View File
@@ -26,7 +26,7 @@ import static java.util.Collections.unmodifiableSet;
import static java.util.concurrent.TimeUnit.SECONDS;
public final class TcpConfig {
public final class TcpTransportAndNetworkConfig {
private static final int DEFAULT_PACKET_SIZE = 1024 * 8;
private static final long DEFAULT_HEART_BEAT_INTERVAL = 20;
@@ -41,10 +41,12 @@ public final class TcpConfig {
private final TimeUnit heartBeatIntervalUnit;
private final IdentifierListener identifierListener;
private TcpConfig(int serverPort, Set<InetSocketAddress> endpoints, int packetSize,
boolean autoReconnectedUponDroppedConnection,
ThrottlingConfig throttlingConfig, long heartBeatInterval,
TimeUnit heartBeatIntervalUnit, IdentifierListener identifierListener) {
private TcpTransportAndNetworkConfig(int serverPort, Set<InetSocketAddress> endpoints,
int packetSize,
boolean autoReconnectedUponDroppedConnection,
ThrottlingConfig throttlingConfig, long heartBeatInterval,
TimeUnit heartBeatIntervalUnit,
IdentifierListener identifierListener) {
this.serverPort = serverPort;
this.endpoints = endpoints;
this.packetSize = packetSize;
@@ -56,29 +58,30 @@ public final class TcpConfig {
}
public static TcpConfig forSendingNode(int serverPort, InetSocketAddress... endpoints) {
public static TcpTransportAndNetworkConfig forSendingNode(int serverPort,
InetSocketAddress... endpoints) {
return forSendingNode(serverPort, Arrays.asList(endpoints));
}
public static TcpConfig forSendingNode(int serverPort,
public static TcpTransportAndNetworkConfig forSendingNode(int serverPort,
Collection<InetSocketAddress> endpoints) {
if (endpoints.isEmpty())
throw new IllegalArgumentException("There should be some endpoints");
return forUnknownTopology(serverPort, endpoints);
}
public static TcpConfig forReceivingOnlyNode(int serverPort) {
public static TcpTransportAndNetworkConfig forReceivingOnlyNode(int serverPort) {
return forUnknownTopology(serverPort, Collections.<InetSocketAddress>emptyList());
}
public static TcpConfig forUnknownTopology(int serverPort,
public static TcpTransportAndNetworkConfig forUnknownTopology(int serverPort,
Collection<InetSocketAddress> endpoints) {
for (final InetSocketAddress endpoint : endpoints) {
if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName()))
throw new IllegalArgumentException("endpoint=" + endpoint
+ " can not point to the same port as the server");
}
return new TcpConfig(
return new TcpTransportAndNetworkConfig(
serverPort, unmodifiableSet(new HashSet<InetSocketAddress>(endpoints)),
DEFAULT_PACKET_SIZE,
true, // autoReconnectedUponDroppedConnection
@@ -92,8 +95,9 @@ public final class TcpConfig {
return identifierListener;
}
public TcpConfig nonUniqueIdentifierListener(final IdentifierListener identifierListener) {
return new TcpConfig(serverPort, endpoints, packetSize,
public TcpTransportAndNetworkConfig nonUniqueIdentifierListener(
final IdentifierListener identifierListener) {
return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize,
autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval,
heartBeatIntervalUnit, identifierListener);
}
@@ -102,8 +106,9 @@ public final class TcpConfig {
return autoReconnectedUponDroppedConnection;
}
public TcpConfig autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) {
return new TcpConfig(serverPort, endpoints, packetSize,
public TcpTransportAndNetworkConfig autoReconnectedUponDroppedConnection(
boolean autoReconnectedUponDroppedConnection) {
return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize,
autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval,
heartBeatIntervalUnit, identifierListener);
}
@@ -112,9 +117,9 @@ public final class TcpConfig {
return throttlingConfig;
}
public TcpConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
public TcpTransportAndNetworkConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "TCP");
return new TcpConfig(serverPort, endpoints, packetSize,
return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize,
autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval,
heartBeatIntervalUnit, identifierListener);
}
@@ -127,8 +132,8 @@ public final class TcpConfig {
return serverPort;
}
public TcpConfig serverPort(int serverPort) {
return new TcpConfig(serverPort, endpoints, packetSize,
public TcpTransportAndNetworkConfig serverPort(int serverPort) {
return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize,
autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval,
heartBeatIntervalUnit, identifierListener);
}
@@ -137,13 +142,13 @@ public final class TcpConfig {
return endpoints;
}
public TcpConfig endpoints(Set<InetSocketAddress> endpoints) {
public TcpTransportAndNetworkConfig endpoints(Set<InetSocketAddress> endpoints) {
for (final InetSocketAddress endpoint : endpoints) {
if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName()))
throw new IllegalArgumentException("endpoint=" + endpoint
+ " can not point to the same port as the server");
}
return new TcpConfig(serverPort, endpoints, packetSize,
return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize,
autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval,
heartBeatIntervalUnit, identifierListener);
}
@@ -152,16 +157,17 @@ public final class TcpConfig {
return packetSize;
}
public TcpConfig packetSize(int packetSize) {
public TcpTransportAndNetworkConfig packetSize(int packetSize) {
if (packetSize <= 0)
throw new IllegalArgumentException();
return new TcpConfig(serverPort, endpoints, packetSize,
return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize,
autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval,
heartBeatIntervalUnit, identifierListener);
}
public TcpConfig heartBeatInterval(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) {
return new TcpConfig(serverPort, endpoints, packetSize,
public TcpTransportAndNetworkConfig heartBeatInterval(long heartBeatInterval,
TimeUnit heartBeatIntervalUnit) {
return new TcpTransportAndNetworkConfig(serverPort, endpoints, packetSize,
autoReconnectedUponDroppedConnection, throttlingConfig, heartBeatInterval,
heartBeatIntervalUnit, identifierListener);
}
@@ -172,7 +178,7 @@ public final class TcpConfig {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
TcpConfig that = (TcpConfig) o;
TcpTransportAndNetworkConfig that = (TcpTransportAndNetworkConfig) o;
if (autoReconnectedUponDroppedConnection != that.autoReconnectedUponDroppedConnection) return false;
if (heartBeatInterval != that.heartBeatInterval) return false;
src/main/java/net/openhft/chronicle/hash/replication/UdpConfig.java → src/main/java/net/openhft/chronicle/hash/replication/UdpTransportConfig.java
+19 -19
View File
@@ -25,16 +25,16 @@ import java.net.InetAddress;
import java.net.NetworkInterface;
public final class UdpConfig {
public final class UdpTransportConfig {
private final InetAddress address;
private final int port;
private final NetworkInterface networkInterface;
private final @Nullable NetworkInterface networkInterface;
private final ThrottlingConfig throttlingConfig;
private UdpConfig(
private UdpTransportConfig(
InetAddress address,
int port,
NetworkInterface networkInterface,
@Nullable NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
if (address == null) {
throw new NullPointerException("Null address");
@@ -49,23 +49,16 @@ public final class UdpConfig {
this.throttlingConfig = throttlingConfig;
}
// TODO better name
public static UdpConfig simple(@NotNull InetAddress address, int port) {
public static UdpTransportConfig of(@NotNull InetAddress address, int port) {
if (address.isMulticastAddress())
throw new IllegalArgumentException();
return create(address, port, null, ThrottlingConfig.noThrottling());
}
public static UdpConfig multiCast(@NotNull InetAddress address, int port,
@NotNull NetworkInterface networkInterface) {
if (!address.isMulticastAddress() || networkInterface == null)
throw new IllegalArgumentException();
return create(address, port, networkInterface, ThrottlingConfig.noThrottling());
}
static UdpConfig create(InetAddress address, int port, NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
return new UdpConfig(address, port, networkInterface, throttlingConfig);
static UdpTransportConfig create(InetAddress address, int port,
@Nullable NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
return new UdpTransportConfig(address, port, networkInterface, throttlingConfig);
}
@NotNull
@@ -102,8 +95,8 @@ public final class UdpConfig {
if (o == this) {
return true;
}
if (o instanceof UdpConfig) {
UdpConfig that = (UdpConfig) o;
if (o instanceof UdpTransportConfig) {
UdpTransportConfig that = (UdpTransportConfig) o;
return (this.address.equals(that.address()))
&& (this.port == that.port())
&& (this.networkInterface.equals(that.networkInterface()))
@@ -126,7 +119,14 @@ public final class UdpConfig {
return h;
}
public UdpConfig throttlingConfig(@NotNull ThrottlingConfig throttlingConfig) {
public UdpTransportConfig networkInterfaceForMulticast(
@Nullable NetworkInterface networkInterface) {
if (!address().isMulticastAddress())
throw new IllegalArgumentException();
return create(address(), port(), networkInterface, throttlingConfig());
}
public UdpTransportConfig throttlingConfig(@NotNull ThrottlingConfig throttlingConfig) {
ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP");
return create(address(), port(), networkInterface(), throttlingConfig);
}
src/main/java/net/openhft/chronicle/map/AbstractChronicleMapBuilder.java
+29 -29
View File
@@ -139,7 +139,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
private DefaultValueProvider<K, V> defaultValueProvider = NullValueProvider.INSTANCE;
private PrepareValueBytes<K, V> prepareValueBytes = null;
private SimpleReplication simpleReplication = null;
private SingleChronicleHashReplication singleHashReplication = null;
AbstractChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
@@ -740,41 +740,41 @@ public abstract class AbstractChronicleMapBuilder<K, V,
}
@Override
public B replication(SimpleReplication replication) {
this.simpleReplication = replication;
public B replication(SingleChronicleHashReplication replication) {
this.singleHashReplication = replication;
return self();
}
@Override
public B replication(byte identifier, TcpConfig tcpTransportAndNetwork) {
return replication(SimpleReplication.builder()
.tcpTransportAndNetwork(tcpTransportAndNetwork).create(identifier));
public B replication(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) {
return replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpTransportAndNetwork).createWithId(identifier));
}
@Override
public ChronicleHashInstanceConfig<ChronicleMap<K, V>> instance() {
return new InstanceConfig<>(this.clone(), simpleReplication, null, null);
return new InstanceConfig<>(this.clone(), singleHashReplication, null, null);
}
@Override
public ChronicleMap<K, V> createPersistedTo(File file) throws IOException {
return createWithFile(file, simpleReplication, null);
return createWithFile(file, singleHashReplication, null);
}
@Override
public ChronicleMap<K, V> create() throws IOException {
return createWithoutFile(simpleReplication, null);
return createWithoutFile(singleHashReplication, null);
}
ChronicleMap<K, V> create(InstanceConfig<K, V> ib) throws IOException {
if (ib.file != null) {
return createWithFile(ib.file, ib.simpleReplication, ib.channel);
return createWithFile(ib.file, ib.singleHashReplication, ib.channel);
} else {
return createWithoutFile(ib.simpleReplication, ib.channel);
return createWithoutFile(ib.singleHashReplication, ib.channel);
}
}
ChronicleMap<K, V> createWithFile(File file, SimpleReplication simpleReplication,
ChronicleMap<K, V> createWithFile(File file, SingleChronicleHashReplication singleHashReplication,
ReplicationChannel channel) throws IOException {
for (int i = 0; i < 10; i++) {
if (file.exists() && file.length() > 0) {
@@ -785,7 +785,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
(VanillaChronicleMap<K, ?, ?, V, ?, ?>) ois.readObject();
map.headerSize = roundUpMapHeaderSize(fis.getChannel().position());
map.createMappedStoreAndSegments(file);
return establishReplication(map, simpleReplication, channel);
return establishReplication(map, singleHashReplication, channel);
} catch (ClassNotFoundException e) {
throw new IOException(e);
} finally {
@@ -805,7 +805,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
if (!file.exists())
throw new FileNotFoundException("Unable to create " + file);
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap(simpleReplication, channel);
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap(singleHashReplication, channel);
FileOutputStream fos = new FileOutputStream(file);
ObjectOutputStream oos = new ObjectOutputStream(fos);
@@ -818,7 +818,7 @@ public abstract class AbstractChronicleMapBuilder<K, V,
oos.close();
}
return establishReplication(map, simpleReplication, channel);
return establishReplication(map, singleHashReplication, channel);
}
/**
@@ -839,12 +839,12 @@ public abstract class AbstractChronicleMapBuilder<K, V,
}
private ChronicleMap<K, V> createWithoutFile(
SimpleReplication simpleReplication, ReplicationChannel channel) throws IOException {
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap(simpleReplication, channel);
SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException {
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap(singleHashReplication, channel);
BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE,
map.sizeInBytes(), true);
map.createMappedStoreAndSegments(bytesStore);
return establishReplication(map, simpleReplication, channel);
return establishReplication(map, singleHashReplication, channel);
}
ChronicleMap<K, V> createStatelessMap(StatelessMapConfig<K, V> statelessBuilder)
@@ -856,13 +856,13 @@ public abstract class AbstractChronicleMapBuilder<K, V,
}
private VanillaChronicleMap<K, ?, ?, V, ?, ?> newMap(
SimpleReplication simpleReplication, ReplicationChannel channel) throws IOException {
boolean replicated = simpleReplication != null || channel != null;
SingleChronicleHashReplication singleHashReplication, ReplicationChannel channel) throws IOException {
boolean replicated = singleHashReplication != null || channel != null;
preMapConstruction(replicated);
if (replicated) {
byte identifier;
if (simpleReplication != null) {
identifier = simpleReplication.identifier();
if (singleHashReplication != null) {
identifier = singleHashReplication.identifier();
} else {
identifier = channel.hub().identifier();
}
@@ -889,19 +889,19 @@ public abstract class AbstractChronicleMapBuilder<K, V,
}
private ChronicleMap<K, V> establishReplication(
VanillaChronicleMap<K, ?, ?, V, ?, ?> map, SimpleReplication simpleReplication,
VanillaChronicleMap<K, ?, ?, V, ?, ?> map, SingleChronicleHashReplication singleHashReplication,
ReplicationChannel channel) throws IOException {
if (map instanceof ReplicatedChronicleMap) {
if (simpleReplication != null && channel != null) {
if (singleHashReplication != null && channel != null) {
throw new AssertionError("Only one non-null replication should be passed");
}
ReplicatedChronicleMap result = (ReplicatedChronicleMap) map;
List<Replicator> replicators = new ArrayList<>(2);
if (simpleReplication != null) {
if (simpleReplication.tcpTransportAndNetwork() != null)
replicators.add(Replicators.tcp(simpleReplication.tcpTransportAndNetwork()));
if (simpleReplication.udpTransport() != null)
replicators.add(Replicators.udp(simpleReplication.udpTransport()));
if (singleHashReplication != null) {
if (singleHashReplication.tcpTransportAndNetwork() != null)
replicators.add(Replicators.tcp(singleHashReplication.tcpTransportAndNetwork()));
if (singleHashReplication.udpTransport() != null)
replicators.add(Replicators.udp(singleHashReplication.udpTransport()));
} else {
ReplicationHub hub = channel.hub();
int entrySize = entrySize(true);
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+4 -4
View File
@@ -20,8 +20,8 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.ReplicationHub;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.UdpConfig;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
import net.openhft.lang.collection.DirectBitSet;
import net.openhft.lang.collection.SingleThreadedDirectBitSet;
import net.openhft.lang.io.ByteBufferBytes;
@@ -58,7 +58,7 @@ final class ChannelProvider implements Closeable {
if (channelProvider != null)
return channelProvider;
channelProvider = new ChannelProvider(hub);
TcpConfig tcpConfig = hub.tcpTransportAndNetwork();
TcpTransportAndNetworkConfig tcpConfig = hub.tcpTransportAndNetwork();
if (tcpConfig != null) {
final TcpReplicator tcpReplicator = new TcpReplicator(
channelProvider.asReplica,
@@ -67,7 +67,7 @@ final class ChannelProvider implements Closeable {
channelProvider.add(tcpReplicator);
}
UdpConfig udpConfig = hub.udpTransport();
UdpTransportConfig udpConfig = hub.udpTransport();
if (udpConfig != null) {
final UdpReplicator udpReplicator =
new UdpReplicator(
src/main/java/net/openhft/chronicle/map/InstanceConfig.java
+10 -10
View File
@@ -18,8 +18,8 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.replication.ReplicationChannel;
import net.openhft.chronicle.hash.replication.SimpleReplication;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import java.io.File;
import java.io.IOException;
@@ -28,30 +28,30 @@ final class InstanceConfig<K, V>
implements ChronicleHashInstanceConfig<ChronicleMap<K, V>> {
final AbstractChronicleMapBuilder<K, V, ?> mapBuilder;
final SimpleReplication simpleReplication;
final SingleChronicleHashReplication singleHashReplication;
final ReplicationChannel channel;
final File file;
InstanceConfig(AbstractChronicleMapBuilder<K, V, ?> mapBuilder,
SimpleReplication simpleReplication,
SingleChronicleHashReplication singleHashReplication,
ReplicationChannel channel,
File file) {
this.mapBuilder = mapBuilder;
this.simpleReplication = simpleReplication;
this.singleHashReplication = singleHashReplication;
this.channel = channel;
this.file = file;
}
@Override
public ChronicleHashInstanceConfig<ChronicleMap<K, V>> replicated(
byte identifier, TcpConfig tcpTransportAndNetwork) {
return replicated(SimpleReplication.builder()
.tcpTransportAndNetwork(tcpTransportAndNetwork).create(identifier));
byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) {
return replicated(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpTransportAndNetwork).createWithId(identifier));
}
@Override
public ChronicleHashInstanceConfig<ChronicleMap<K, V>> replicated(
SimpleReplication replication) {
SingleChronicleHashReplication replication) {
return new InstanceConfig<>(mapBuilder, replication, null, file);
}
@@ -63,7 +63,7 @@ final class InstanceConfig<K, V>
@Override
public ChronicleHashInstanceConfig<ChronicleMap<K, V>> persistedTo(File file) {
return new InstanceConfig<>(mapBuilder, simpleReplication, channel, file);
return new InstanceConfig<>(mapBuilder, singleHashReplication, channel, file);
}
@Override
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+7 -7
View File
@@ -19,8 +19,8 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.IdentifierListener;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.UdpConfig;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
import net.openhft.lang.collection.ATSDirectBitSet;
import net.openhft.lang.collection.DirectBitSet;
import net.openhft.lang.io.ByteBufferBytes;
@@ -42,7 +42,7 @@ import java.util.concurrent.atomic.AtomicReference;
import static java.nio.ByteBuffer.allocateDirect;
import static java.util.concurrent.TimeUnit.SECONDS;
import static net.openhft.chronicle.hash.replication.UdpConfig.simple;
import static net.openhft.chronicle.hash.replication.UdpTransportConfig.of;
import static net.openhft.chronicle.map.ConcurrentExpiryMap.getDefaultAddress;
import static net.openhft.chronicle.map.DiscoveryNodeBytesMarshallable.ProposedNodes;
import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.BOOTSTRAP_BYTES;
@@ -56,7 +56,7 @@ import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
public class NodeDiscovery {
private final AddressAndPort ourAddressAndPort;
private final UdpConfig udpConfig;
private final UdpTransportConfig udpConfig;
private final DiscoveryNodeBytesMarshallable discoveryNodeBytesMarshallable;
private final AtomicReference<NodeDiscoveryEventListener> nodeDiscoveryEventListenerAtomicReference =
new AtomicReference<NodeDiscoveryEventListener>();
@@ -71,7 +71,7 @@ public class NodeDiscovery {
@NotNull InetAddress tcpAddress,
@NotNull InetAddress udpBroadcastAddress) throws IOException {
this.ourAddressAndPort = new AddressAndPort(tcpAddress.getAddress(), tcpPort);
this.udpConfig = simple(udpBroadcastAddress, udpBroadcastPort);
this.udpConfig = of(udpBroadcastAddress, udpBroadcastPort);
knownNodes = new KnownNodes();
discoveryNodeBytesMarshallable = new DiscoveryNodeBytesMarshallable
@@ -221,7 +221,7 @@ public class NodeDiscovery {
// add our identifier and host:port to the list of known identifiers
knownNodes.add(ourAddressAndPort, identifier);
final TcpConfig tcpConfig = TcpConfig
final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig
.forUnknownTopology(ourAddressAndPort.getPort(),
toInetSocketCollection(knownHostPorts))
.heartBeatInterval(1, SECONDS)
@@ -360,7 +360,7 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
* @throws java.io.IOException
*/
NodeDiscoveryBroadcaster(
final UdpConfig replicationConfig,
final UdpTransportConfig replicationConfig,
final int serializedEntrySize,
final BytesMarshallable externalizable)
throws IOException {
src/main/java/net/openhft/chronicle/map/Replicators.java
+4 -4
View File
@@ -16,8 +16,8 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.UdpConfig;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
import org.jetbrains.annotations.NotNull;
import java.io.Closeable;
@@ -36,7 +36,7 @@ final class Replicators {
private Replicators() {
}
static Replicator tcp(final TcpConfig replicationConfig) {
static Replicator tcp(final TcpTransportAndNetworkConfig replicationConfig) {
return new Replicator() {
@Override
@@ -60,7 +60,7 @@ final class Replicators {
}
static Replicator udp(
final UdpConfig replicationConfig) {
final UdpTransportConfig replicationConfig) {
return new Replicator() {
@Override
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+3 -3
View File
@@ -19,7 +19,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.IdentifierListener;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.ThrottlingConfig;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
@@ -74,7 +74,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
private final byte localIdentifier;
private final int maxEntrySizeBytes;
private final Replica.EntryExternalizable externalizable;
private final TcpConfig replicationConfig;
private final TcpTransportAndNetworkConfig replicationConfig;
private final StatelessServerConnector statelessServerConnector;
private long selectorTimeout;
@@ -89,7 +89,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
*/
public TcpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable externalizable,
@NotNull final TcpConfig replicationConfig,
@NotNull final TcpTransportAndNetworkConfig replicationConfig,
final int maxEntrySizeBytes,
@Nullable StatelessServerConnector statelessServerConnector) throws IOException {
src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java
+13 -4
View File
@@ -18,7 +18,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.UdpConfig;
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -69,7 +69,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
* @param localIdentifier just used for logging
* @throws IOException
*/
UdpChannelReplicator(@NotNull final UdpConfig replicationConfig,
UdpChannelReplicator(@NotNull final UdpTransportConfig replicationConfig,
final int maxPayloadSize,
final byte localIdentifier)
throws IOException {
@@ -178,10 +178,17 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
final InetSocketAddress hostAddress = new InetSocketAddress(port);
client.configureBlocking(false);
if (address.isMulticastAddress()) {
client.setOption(IP_MULTICAST_IF, networkInterface);
client.setOption(SO_REUSEADDR, true);
client.bind(hostAddress);
client.join(address, networkInterface);
if (networkInterface != null) {
// This is probably not needed, because client socket doesn't send datagrams,
// but since EVERYBODY on the internet configures this for any channels, and
// I don't see any harm this config could make, I leave it here
client.setOption(IP_MULTICAST_IF, networkInterface);
client.join(address, networkInterface);
} else {
client.join(address, NetworkInterface.getByInetAddress(hostAddress.getAddress()));
}
if (LOG.isDebugEnabled())
LOG.debug("Connecting via multicast, group=" + address);
} else {
@@ -264,6 +271,8 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
.setOption(StandardSocketOptions.IP_MULTICAST_LOOP, false)
.setOption(StandardSocketOptions.SO_BROADCAST, true)
.setOption(SO_REUSEADDR, true);
if (networkInterface != null)
server.setOption(StandardSocketOptions.IP_MULTICAST_IF, networkInterface);
// the registration has be be run on the same thread as the selector
addPendingRegistration(new Runnable() {
src/main/java/net/openhft/chronicle/map/UdpReplicator.java
+2 -2
View File
@@ -18,7 +18,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.UdpConfig;
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
import net.openhft.lang.io.ByteBufferBytes;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
@@ -43,7 +43,7 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
public UdpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable entryExternalizable,
@NotNull final UdpConfig replicationConfig,
@NotNull final UdpTransportConfig replicationConfig,
final int serializedEntrySize)
throws IOException {
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+4 -4
View File
@@ -19,8 +19,8 @@
package net.openhft.chronicle.set;
import net.openhft.chronicle.hash.*;
import net.openhft.chronicle.hash.replication.SimpleReplication;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.TimeProvider;
import net.openhft.chronicle.hash.serialization.BytesReader;
import net.openhft.chronicle.hash.serialization.BytesWriter;
@@ -266,13 +266,13 @@ public class ChronicleSetBuilder<E>
}
@Override
public ChronicleSetBuilder<E> replication(SimpleReplication replication) {
public ChronicleSetBuilder<E> replication(SingleChronicleHashReplication replication) {
chronicleMapBuilder.replication(replication);
return this;
}
@Override
public ChronicleSetBuilder<E> replication(byte identifier, TcpConfig tcpTransportAndNetwork) {
public ChronicleSetBuilder<E> replication(byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) {
chronicleMapBuilder.replication(identifier, tcpTransportAndNetwork);
return this;
}
src/main/java/net/openhft/chronicle/set/InstanceConfig.java
+4 -4
View File
@@ -18,8 +18,8 @@ package net.openhft.chronicle.set;
import net.openhft.chronicle.hash.ChronicleHashInstanceConfig;
import net.openhft.chronicle.hash.replication.ReplicationChannel;
import net.openhft.chronicle.hash.replication.SimpleReplication;
import net.openhft.chronicle.hash.replication.TcpConfig;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.map.ChronicleMap;
import java.io.File;
@@ -35,12 +35,12 @@ final class InstanceConfig<E> implements ChronicleHashInstanceConfig<ChronicleSe
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> replicated(
byte identifier, TcpConfig tcpTransportAndNetwork) {
byte identifier, TcpTransportAndNetworkConfig tcpTransportAndNetwork) {
return new InstanceConfig<>(mapConfig.replicated(identifier, tcpTransportAndNetwork));
}
@Override
public ChronicleHashInstanceConfig<ChronicleSet<E>> replicated(SimpleReplication replication) {
public ChronicleHashInstanceConfig<ChronicleSet<E>> replicated(SingleChronicleHashReplication replication) {
return new InstanceConfig<>(mapConfig.replicated(replication));
}