mirror of
https://github.com/OpenHFT/Chronicle-Map.git
synced 2021-09-13 01:26:12 +00:00
HCOLL-163 kept the use of tcp() and udp() internal
This commit is contained in:
@@ -189,7 +189,7 @@ public interface ChronicleHashBuilder<K, C, B extends ChronicleHashBuilder<K, C,
|
||||
|
||||
B metaDataBytes(int metaDataBytes);
|
||||
|
||||
B replicators(byte identifier, Replicator... replicators);
|
||||
B replicators(byte identifier, ReplicatorConfig... replicators);
|
||||
|
||||
B timeProvider(TimeProvider timeProvider);
|
||||
|
||||
|
||||
@@ -26,8 +26,8 @@ public final class ChannelProviderBuilder {
|
||||
byte identifier;
|
||||
int maxEntrySize = 1024;
|
||||
int maxNumberOfChronicles = 128;
|
||||
private UdpReplicationConfig udpReplicationConfig = null;
|
||||
private TcpReplicationConfig tcpReplicationConfig = null;
|
||||
private UdpReplicationReplicatorConfig udpReplicationConfig = null;
|
||||
private TcpReplicationReplicatorConfig tcpReplicationConfig = null;
|
||||
|
||||
|
||||
|
||||
@@ -40,16 +40,16 @@ public final class ChannelProviderBuilder {
|
||||
return this;
|
||||
}
|
||||
|
||||
public ChannelProviderBuilder replicators(byte identifier, Config... configs) {
|
||||
public ChannelProviderBuilder replicators(byte identifier, ReplicatorConfig... replicatorConfigs) {
|
||||
this.identifier = identifier;
|
||||
|
||||
|
||||
for (Config config : configs) {
|
||||
for (ReplicatorConfig replicatorConfig : replicatorConfigs) {
|
||||
|
||||
if (config instanceof TcpReplicationConfig) {
|
||||
this.tcpReplicationConfig = (TcpReplicationConfig) config;
|
||||
} else if (config instanceof UdpReplicationConfig) {
|
||||
this.udpReplicationConfig = (UdpReplicationConfig) config;
|
||||
if (replicatorConfig instanceof TcpReplicationReplicatorConfig) {
|
||||
this.tcpReplicationConfig = (TcpReplicationReplicatorConfig) replicatorConfig;
|
||||
} else if (replicatorConfig instanceof UdpReplicationReplicatorConfig) {
|
||||
this.udpReplicationConfig = (UdpReplicationReplicatorConfig) replicatorConfig;
|
||||
} else
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@@ -551,11 +551,21 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
|
||||
return toString().hashCode();
|
||||
}
|
||||
|
||||
public ChronicleMapBuilder<K, V> replicators(byte identifier, Replicator... replicators) {
|
||||
public ChronicleMapBuilder<K, V> replicators(byte identifier, ReplicatorConfig... replicatorConfigs) {
|
||||
|
||||
this.identifier = identifier;
|
||||
this.replicators.clear();
|
||||
for (Replicator replicator : replicators) {
|
||||
for (ReplicatorConfig replicatorConfig : replicatorConfigs) {
|
||||
|
||||
Replicator replicator;
|
||||
if (replicatorConfig instanceof TcpReplicationReplicatorConfig) {
|
||||
replicator = Replicators.tcp((TcpReplicationReplicatorConfig) replicatorConfig);
|
||||
} else if (replicatorConfig instanceof UdpReplicationReplicatorConfig) {
|
||||
replicator = Replicators.udp((UdpReplicationReplicatorConfig) replicatorConfig);
|
||||
} else
|
||||
throw new UnsupportedOperationException();
|
||||
|
||||
|
||||
this.replicators.put(replicator.getClass(), replicator);
|
||||
}
|
||||
|
||||
|
||||
@@ -25,8 +25,7 @@ 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;
|
||||
import static net.openhft.chronicle.map.NodeDiscoveryBroadcaster.LOG;
|
||||
import static net.openhft.chronicle.map.Replicators.tcp;
|
||||
import static net.openhft.chronicle.map.UdpReplicationConfig.simple;
|
||||
import static net.openhft.chronicle.map.UdpReplicationReplicatorConfig.simple;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
@@ -36,7 +35,7 @@ import static net.openhft.chronicle.map.UdpReplicationConfig.simple;
|
||||
public class NodeDiscovery {
|
||||
|
||||
private final AddressAndPort ourAddressAndPort;
|
||||
private final UdpReplicationConfig udpConfig;
|
||||
private final UdpReplicationReplicatorConfig udpConfig;
|
||||
private final DiscoveryNodeBytesMarshallable discoveryNodeBytesMarshallable;
|
||||
private final AtomicReference<NodeDiscoveryEventListener> nodeDiscoveryEventListenerAtomicReference =
|
||||
new AtomicReference<NodeDiscoveryEventListener>();
|
||||
@@ -201,7 +200,7 @@ public class NodeDiscovery {
|
||||
// add our identifier and host:port to the list of known identifiers
|
||||
knownNodes.add(ourAddressAndPort, identifier);
|
||||
|
||||
final TcpReplicationConfig tcpConfig = TcpReplicationConfig
|
||||
final TcpReplicationReplicatorConfig tcpConfig = TcpReplicationReplicatorConfig
|
||||
.of(ourAddressAndPort.getPort(), toInetSocketArray(knownHostPorts))
|
||||
.heartBeatInterval(1, SECONDS).nonUniqueIdentifierListener(identifierListener);
|
||||
|
||||
@@ -212,7 +211,7 @@ public class NodeDiscovery {
|
||||
return ChronicleMapBuilder.of(Integer.class,
|
||||
CharSequence.class)
|
||||
.entries(20000L)
|
||||
.replicators(identifier, tcp(tcpConfig)).create();
|
||||
.replicators(identifier, tcpConfig).create();
|
||||
|
||||
}
|
||||
|
||||
@@ -340,7 +339,7 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
|
||||
* @throws java.io.IOException
|
||||
*/
|
||||
NodeDiscoveryBroadcaster(
|
||||
@NotNull final UdpReplicationConfig replicationConfig,
|
||||
@NotNull final UdpReplicationReplicatorConfig replicationConfig,
|
||||
final int serializedEntrySize,
|
||||
final BytesMarshallable externalizable)
|
||||
throws IOException {
|
||||
|
||||
+3
-1
@@ -3,5 +3,7 @@ package net.openhft.chronicle.map;
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
*/
|
||||
public interface Config {
|
||||
public interface ReplicatorConfig<B> {
|
||||
|
||||
|
||||
}
|
||||
@@ -25,7 +25,7 @@ public final class Replicators {
|
||||
private Replicators() {
|
||||
}
|
||||
|
||||
public static Replicator tcp(final TcpReplicationConfig replicationConfig) {
|
||||
public static Replicator tcp(final TcpReplicationReplicatorConfig replicationConfig) {
|
||||
return new Replicator() {
|
||||
|
||||
@Override
|
||||
@@ -39,7 +39,7 @@ public final class Replicators {
|
||||
}
|
||||
|
||||
public static Replicator udp(
|
||||
final UdpReplicationConfig replicationConfig) {
|
||||
final UdpReplicationReplicatorConfig replicationConfig) {
|
||||
return new Replicator() {
|
||||
|
||||
@Override
|
||||
|
||||
@@ -80,7 +80,7 @@ final class SerializationBuilder<E> implements Cloneable {
|
||||
if (role == Role.VALUE) {
|
||||
factory = concreteClass(eClass) && marshallerUseFactory(eClass) ?
|
||||
new NewInstanceObjectFactory<E>(eClass) :
|
||||
NullObjectFactory.of();
|
||||
(ObjectFactory<E>) NullObjectFactory.of();
|
||||
}
|
||||
|
||||
if (concreteClass(eClass) && Byteable.class.isAssignableFrom(eClass)) {
|
||||
|
||||
+14
-14
@@ -29,7 +29,7 @@ import static java.util.Collections.unmodifiableSet;
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
|
||||
|
||||
public class TcpReplicationConfig implements Config {
|
||||
public class TcpReplicationReplicatorConfig implements ReplicatorConfig {
|
||||
|
||||
private static final int DEFAULT_PACKET_SIZE = 1024 * 8;
|
||||
private static final long DEFAULT_HEART_BEAT_INTERVAL = 20;
|
||||
@@ -45,7 +45,7 @@ public class TcpReplicationConfig implements Config {
|
||||
private TimeUnit heartBeatIntervalUnit = DEFAULT_HEART_BEAT_INTERVAL_UNIT;
|
||||
public IdentifierListener identifierListener;
|
||||
|
||||
TcpReplicationConfig() {
|
||||
TcpReplicationReplicatorConfig() {
|
||||
|
||||
}
|
||||
|
||||
@@ -53,7 +53,7 @@ public class TcpReplicationConfig implements Config {
|
||||
return identifierListener;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig nonUniqueIdentifierListener(@NotNull final IdentifierListener
|
||||
public TcpReplicationReplicatorConfig nonUniqueIdentifierListener(@NotNull final IdentifierListener
|
||||
identifierListener) {
|
||||
this.identifierListener = identifierListener;
|
||||
return this;
|
||||
@@ -63,7 +63,7 @@ public class TcpReplicationConfig implements Config {
|
||||
return autoReconnectedUponDroppedConnection;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) {
|
||||
public TcpReplicationReplicatorConfig autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) {
|
||||
this.autoReconnectedUponDroppedConnection = autoReconnectedUponDroppedConnection;
|
||||
return this;
|
||||
}
|
||||
@@ -72,7 +72,7 @@ public class TcpReplicationConfig implements Config {
|
||||
return throttlingConfig;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
|
||||
public TcpReplicationReplicatorConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
|
||||
this.throttlingConfig = throttlingConfig;
|
||||
return this;
|
||||
}
|
||||
@@ -90,7 +90,7 @@ public class TcpReplicationConfig implements Config {
|
||||
return serverPort;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig serverPort(int serverPort) {
|
||||
public TcpReplicationReplicatorConfig serverPort(int serverPort) {
|
||||
this.serverPort = serverPort;
|
||||
return this;
|
||||
}
|
||||
@@ -99,7 +99,7 @@ public class TcpReplicationConfig implements Config {
|
||||
return endpoints;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig endpoints(Set<InetSocketAddress> endpoints) {
|
||||
public TcpReplicationReplicatorConfig endpoints(Set<InetSocketAddress> endpoints) {
|
||||
|
||||
for (final InetSocketAddress endpoint : endpoints) {
|
||||
if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName()))
|
||||
@@ -115,11 +115,11 @@ public class TcpReplicationConfig implements Config {
|
||||
return packetSize;
|
||||
}
|
||||
|
||||
public static TcpReplicationConfig of(int serverPort, InetSocketAddress... endpoints) {
|
||||
public static TcpReplicationReplicatorConfig of(int serverPort, InetSocketAddress... endpoints) {
|
||||
return of(serverPort, Arrays.asList(endpoints));
|
||||
}
|
||||
|
||||
public static TcpReplicationConfig of(int serverPort, Collection<InetSocketAddress> endpoints) {
|
||||
public static TcpReplicationReplicatorConfig of(int serverPort, Collection<InetSocketAddress> endpoints) {
|
||||
// at least in tests, we configure "receive-only" replication without endpoints.
|
||||
// TODO decide what to do with this case
|
||||
// if (endpoints.isEmpty())
|
||||
@@ -130,7 +130,7 @@ public class TcpReplicationConfig implements Config {
|
||||
+ " can not point to the same port as the server");
|
||||
}
|
||||
|
||||
TcpReplicationConfig tcpReplicationConfig = new TcpReplicationConfig();
|
||||
TcpReplicationReplicatorConfig tcpReplicationConfig = new TcpReplicationReplicatorConfig();
|
||||
|
||||
tcpReplicationConfig.serverPort(serverPort);
|
||||
tcpReplicationConfig.endpoints = unmodifiableSet(new HashSet<InetSocketAddress>(endpoints));
|
||||
@@ -141,14 +141,14 @@ public class TcpReplicationConfig implements Config {
|
||||
}
|
||||
|
||||
|
||||
public TcpReplicationConfig packetSize(int packetSize) {
|
||||
public TcpReplicationReplicatorConfig packetSize(int packetSize) {
|
||||
if (packetSize <= 0)
|
||||
throw new IllegalArgumentException();
|
||||
this.packetSize = packetSize;
|
||||
return this;
|
||||
}
|
||||
|
||||
public TcpReplicationConfig throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) {
|
||||
public TcpReplicationReplicatorConfig throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) {
|
||||
this.heartBeatInterval = heartBeatInterval;
|
||||
this.heartBeatIntervalUnit = heartBeatIntervalUnit;
|
||||
return this;
|
||||
@@ -160,7 +160,7 @@ public class TcpReplicationConfig implements Config {
|
||||
if (this == o) return true;
|
||||
if (o == null || getClass() != o.getClass()) return false;
|
||||
|
||||
TcpReplicationConfig that = (TcpReplicationConfig) o;
|
||||
TcpReplicationReplicatorConfig that = (TcpReplicationReplicatorConfig) o;
|
||||
|
||||
if (autoReconnectedUponDroppedConnection != that.autoReconnectedUponDroppedConnection) return false;
|
||||
if (heartBeatInterval != that.heartBeatInterval) return false;
|
||||
@@ -199,7 +199,7 @@ public class TcpReplicationConfig implements Config {
|
||||
'}';
|
||||
}
|
||||
|
||||
public TcpReplicationConfig heartBeatInterval(long time, TimeUnit unit) {
|
||||
public TcpReplicationReplicatorConfig heartBeatInterval(long time, TimeUnit unit) {
|
||||
this.heartBeatInterval = time;
|
||||
this.heartBeatIntervalUnit = unit;
|
||||
return this;
|
||||
@@ -55,7 +55,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
private final byte localIdentifier;
|
||||
private final int maxEntrySizeBytes;
|
||||
private final Replica.EntryExternalizable externalizable;
|
||||
private final TcpReplicationConfig replicationConfig;
|
||||
private final TcpReplicationReplicatorConfig replicationConfig;
|
||||
|
||||
private long selectorTimeout;
|
||||
|
||||
@@ -68,7 +68,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
|
||||
*/
|
||||
TcpReplicator(@NotNull final Replica replica,
|
||||
@NotNull final Replica.EntryExternalizable externalizable,
|
||||
@NotNull final TcpReplicationConfig replicationConfig,
|
||||
@NotNull final TcpReplicationReplicatorConfig replicationConfig,
|
||||
final int maxEntrySizeBytes) throws IOException {
|
||||
|
||||
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(),
|
||||
|
||||
@@ -66,7 +66,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
|
||||
* @param localIdentifier just used for logging
|
||||
* @throws IOException
|
||||
*/
|
||||
UdpChannelReplicator(@NotNull final UdpReplicationConfig replicationConfig,
|
||||
UdpChannelReplicator(@NotNull final UdpReplicationReplicatorConfig replicationConfig,
|
||||
final int maxPayloadSize, final byte localIdentifier)
|
||||
throws IOException {
|
||||
|
||||
|
||||
+7
-7
@@ -23,16 +23,16 @@ import java.net.InetAddress;
|
||||
import java.net.NetworkInterface;
|
||||
|
||||
|
||||
public abstract class UdpReplicationConfig implements Config {
|
||||
public abstract class UdpReplicationReplicatorConfig implements ReplicatorConfig {
|
||||
|
||||
/**
|
||||
* Package-private constructor forbids subclassing from outside of the package
|
||||
*/
|
||||
UdpReplicationConfig() {
|
||||
UdpReplicationReplicatorConfig() {
|
||||
// nothing to do
|
||||
}
|
||||
|
||||
public static UdpReplicationConfig simple(@NotNull InetAddress address, int port) {
|
||||
public static UdpReplicationReplicatorConfig simple(@NotNull InetAddress address, int port) {
|
||||
if (address.isMulticastAddress())
|
||||
throw new IllegalArgumentException();
|
||||
|
||||
@@ -40,17 +40,17 @@ public abstract class UdpReplicationConfig implements Config {
|
||||
return create(address, port, null, ThrottlingConfig.noThrottling());
|
||||
}
|
||||
|
||||
public static UdpReplicationConfig multiCast(@NotNull InetAddress address, int port,
|
||||
public static UdpReplicationReplicatorConfig 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 UdpReplicationConfig create(InetAddress address, int port,
|
||||
static UdpReplicationReplicatorConfig create(InetAddress address, int port,
|
||||
NetworkInterface networkInterface,
|
||||
ThrottlingConfig throttlingConfig) {
|
||||
return new UdpReplicationConfigBean(address, port, networkInterface,
|
||||
return new UdpReplicationReplicatorConfigBean(address, port, networkInterface,
|
||||
throttlingConfig);
|
||||
}
|
||||
|
||||
@@ -63,7 +63,7 @@ public abstract class UdpReplicationConfig implements Config {
|
||||
|
||||
public abstract ThrottlingConfig throttlingConfig();
|
||||
|
||||
public UdpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
|
||||
public UdpReplicationReplicatorConfig throttlingConfig(ThrottlingConfig throttlingConfig) {
|
||||
ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP");
|
||||
return create(address(), port(), networkInterface(), throttlingConfig);
|
||||
}
|
||||
+4
-4
@@ -4,13 +4,13 @@ import java.net.InetAddress;
|
||||
import java.net.NetworkInterface;
|
||||
|
||||
|
||||
final class UdpReplicationConfigBean extends UdpReplicationConfig {
|
||||
final class UdpReplicationReplicatorConfigBean extends UdpReplicationReplicatorConfig {
|
||||
private final InetAddress address;
|
||||
private final int port;
|
||||
private final NetworkInterface networkInterface;
|
||||
private final ThrottlingConfig throttlingConfig;
|
||||
|
||||
UdpReplicationConfigBean(
|
||||
UdpReplicationReplicatorConfigBean(
|
||||
InetAddress address,
|
||||
int port,
|
||||
NetworkInterface networkInterface,
|
||||
@@ -63,8 +63,8 @@ final class UdpReplicationConfigBean extends UdpReplicationConfig {
|
||||
if (o == this) {
|
||||
return true;
|
||||
}
|
||||
if (o instanceof UdpReplicationConfig) {
|
||||
UdpReplicationConfig that = (UdpReplicationConfig) o;
|
||||
if (o instanceof UdpReplicationReplicatorConfig) {
|
||||
UdpReplicationReplicatorConfig that = (UdpReplicationReplicatorConfig) o;
|
||||
return (this.address.equals(that.address()))
|
||||
&& (this.port == that.port())
|
||||
&& (this.networkInterface.equals(that.networkInterface()))
|
||||
@@ -40,7 +40,7 @@ class UdpReplicator extends UdpChannelReplicator implements Replica.Modification
|
||||
|
||||
UdpReplicator(@NotNull final Replica replica,
|
||||
@NotNull final Replica.EntryExternalizable entryExternalizable,
|
||||
@NotNull final UdpReplicationConfig replicationConfig,
|
||||
@NotNull final UdpReplicationReplicatorConfig replicationConfig,
|
||||
final int serializedEntrySize)
|
||||
throws IOException {
|
||||
|
||||
|
||||
@@ -87,9 +87,9 @@ public class ChronicleSetBuilder<E>
|
||||
* .keySize(10)
|
||||
* .create();}</pre>
|
||||
*
|
||||
* <p>(Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded
|
||||
* (and each character takes 2 bytes on-heap), because default off-heap {@link String} encoding
|
||||
* is UTF-8 in {@code ChronicleSet}.)
|
||||
* <p>(Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded (and each
|
||||
* character takes 2 bytes on-heap), because default off-heap {@link String} encoding is UTF-8 in {@code
|
||||
* ChronicleSet}.)
|
||||
*
|
||||
* @see #constantKeySizeBySample(Object)
|
||||
* @see #entrySize(int)
|
||||
@@ -119,13 +119,13 @@ public class ChronicleSetBuilder<E>
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
*
|
||||
* <p>In fully default case you can expect entry size to be about 120-130 bytes. But it is
|
||||
* strongly recommended always to configure {@linkplain #keySize(int) key size},
|
||||
* if they couldn't be derived statically.
|
||||
* <p>In fully default case you can expect entry size to be about 120-130 bytes. But it is strongly
|
||||
* recommended always to configure {@linkplain #keySize(int) key size}, if they couldn't be derived
|
||||
* statically.
|
||||
*
|
||||
* <p>If entry size is not configured explicitly by calling this method, it is computed based on
|
||||
* {@linkplain #metaDataBytes(int) meta data bytes}, plus {@linkplain #keySize(int) key size},
|
||||
* plus a few bytes required by implementations.
|
||||
* {@linkplain #metaDataBytes(int) meta data bytes}, plus {@linkplain #keySize(int) key size}, plus a few
|
||||
* bytes required by implementations.
|
||||
*/
|
||||
@Override
|
||||
public ChronicleSetBuilder<E> entrySize(int entrySize) {
|
||||
@@ -164,11 +164,12 @@ public class ChronicleSetBuilder<E>
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChronicleSetBuilder<E> replicators(byte identifier, Replicator... replicators) {
|
||||
public ChronicleSetBuilder<E> replicators(byte identifier, ReplicatorConfig... replicators) {
|
||||
chronicleMapBuilder.replicators(identifier, replicators);
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return " ChronicleSetBuilder{" +
|
||||
@@ -189,12 +190,6 @@ public class ChronicleSetBuilder<E>
|
||||
return toString().hashCode();
|
||||
}
|
||||
|
||||
public ChronicleSetBuilder<E> replicators(byte identifier, Replicator replicator) {
|
||||
chronicleMapBuilder.replicators(identifier, replicator);
|
||||
return this;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public ChronicleSetBuilder<E> timeProvider(TimeProvider timeProvider) {
|
||||
chronicleMapBuilder.timeProvider(timeProvider);
|
||||
|
||||
@@ -23,7 +23,6 @@ import java.io.RandomAccessFile;
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
import static net.openhft.chronicle.map.Replicators.tcp;
|
||||
|
||||
/**
|
||||
* @author Rob Austin.
|
||||
@@ -69,11 +68,11 @@ public class Builder {
|
||||
final byte identifier,
|
||||
final int serverPort,
|
||||
final InetSocketAddress... endpoints) throws IOException {
|
||||
TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints)
|
||||
TcpReplicationReplicatorConfig tcpConfig = TcpReplicationReplicatorConfig.of(serverPort, endpoints)
|
||||
.heartBeatInterval(1L, SECONDS);
|
||||
return ChronicleMapBuilder.of(kClass, vClass)
|
||||
.entries(20000L)
|
||||
.replicators(identifier, tcp( tcpConfig));
|
||||
.replicators(identifier, tcpConfig);
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -51,7 +51,7 @@ public class ChannelReplicationTest {
|
||||
|
||||
public void setup() throws IOException {
|
||||
{
|
||||
TcpReplicationConfig tcpConfig = TcpReplicationConfig
|
||||
TcpReplicationReplicatorConfig tcpConfig = TcpReplicationReplicatorConfig
|
||||
.of(8086, new InetSocketAddress("localhost", 8087))
|
||||
.heartBeatInterval(1, SECONDS);
|
||||
|
||||
@@ -66,8 +66,8 @@ public class ChannelReplicationTest {
|
||||
}
|
||||
|
||||
{
|
||||
TcpReplicationConfig tcpConfig =
|
||||
TcpReplicationConfig.of(8087).heartBeatInterval(1, SECONDS);
|
||||
TcpReplicationReplicatorConfig tcpConfig =
|
||||
TcpReplicationReplicatorConfig.of(8087).heartBeatInterval(1, SECONDS);
|
||||
|
||||
channelProviderB = new ChannelProviderBuilder().replicators((byte) 2, tcpConfig)
|
||||
.create();
|
||||
|
||||
@@ -59,7 +59,7 @@ public class PostChannelCreationTest {
|
||||
public void setup() throws IOException {
|
||||
|
||||
{
|
||||
final TcpReplicationConfig tcpConfig = TcpReplicationConfig
|
||||
final TcpReplicationReplicatorConfig tcpConfig = TcpReplicationReplicatorConfig
|
||||
.of(8086, new InetSocketAddress("localhost", 8087))
|
||||
.heartBeatInterval(1, SECONDS);
|
||||
|
||||
@@ -80,7 +80,7 @@ public class PostChannelCreationTest {
|
||||
|
||||
|
||||
{
|
||||
final TcpReplicationConfig tcpConfig = TcpReplicationConfig
|
||||
final TcpReplicationReplicatorConfig tcpConfig = TcpReplicationReplicatorConfig
|
||||
.of(8087, new InetSocketAddress("localhost", 8086))
|
||||
.heartBeatInterval(1, SECONDS);
|
||||
|
||||
|
||||
@@ -56,12 +56,14 @@ public class SingleMapClusterTest {
|
||||
public void setup() throws IOException {
|
||||
|
||||
{
|
||||
final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig
|
||||
final TcpReplicationReplicatorConfig tcpReplicationConfig = TcpReplicationReplicatorConfig
|
||||
.of(8086, new InetSocketAddress("localhost", 8087))
|
||||
.heartBeatInterval(1, SECONDS);
|
||||
|
||||
int maxEntrySize = 1024;
|
||||
byte identifier= 1;
|
||||
channelProviderA = new ChannelProviderBuilder()
|
||||
.replicators((byte) 1, tcpReplicationConfig).create();
|
||||
.maxEntrySize(maxEntrySize)
|
||||
.replicators(identifier, tcpReplicationConfig).create();
|
||||
// this is how you add maps after the custer is created
|
||||
map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
|
||||
.entries(1000)
|
||||
@@ -71,7 +73,7 @@ public class SingleMapClusterTest {
|
||||
|
||||
|
||||
{
|
||||
final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig
|
||||
final TcpReplicationReplicatorConfig tcpReplicationConfig = TcpReplicationReplicatorConfig
|
||||
.of(8087, new InetSocketAddress("localhost", 8086))
|
||||
.heartBeatInterval(1, SECONDS);
|
||||
|
||||
|
||||
+2
-2
@@ -44,12 +44,12 @@ public class TCPSocketReplicationTest3wayPutReturnsNull {
|
||||
final byte identifier,
|
||||
final int serverPort,
|
||||
final InetSocketAddress... endpoints) throws IOException {
|
||||
TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints);
|
||||
TcpReplicationReplicatorConfig tcpConfig = TcpReplicationReplicatorConfig.of(serverPort, endpoints);
|
||||
return (T) ChronicleMapBuilder.of(Integer.class, CharSequence.class)
|
||||
|
||||
.entries(1000)
|
||||
.putReturnsNull(true)
|
||||
.replicators(identifier, Replicators.tcp(tcpConfig))
|
||||
.replicators(identifier, tcpConfig)
|
||||
.create(Builder.getPersistenceFile());
|
||||
}
|
||||
|
||||
|
||||
@@ -27,7 +27,7 @@ import java.net.InetSocketAddress;
|
||||
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
import static net.openhft.chronicle.map.Builder.getPersistenceFile;
|
||||
import static net.openhft.chronicle.map.Replicators.tcp;
|
||||
import static net.openhft.chronicle.map.TcpReplicationReplicatorConfig.of;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
@@ -43,21 +43,20 @@ public class TwoMapOnDifferentServers {
|
||||
@Before
|
||||
public void setup() throws IOException {
|
||||
|
||||
final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig
|
||||
.of(8076, new InetSocketAddress("localhost", 8077))
|
||||
.heartBeatInterval(1, SECONDS);
|
||||
final TcpReplicationReplicatorConfig tcpConfig =
|
||||
of(8076, new InetSocketAddress("localhost", 8077))
|
||||
.heartBeatInterval(1, SECONDS);
|
||||
|
||||
|
||||
File persistenceFile = getPersistenceFile();
|
||||
map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
|
||||
.entries(20000)
|
||||
.replicators((byte) 1, tcp(tcpReplicationConfig))
|
||||
.replicators((byte) 1, tcpConfig)
|
||||
.create(persistenceFile);
|
||||
|
||||
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
|
||||
.entries(20000)
|
||||
.replicators((byte) 2, tcp(
|
||||
TcpReplicationConfig.of(8077).heartBeatInterval(1, SECONDS)))
|
||||
.replicators((byte) 2, of(8077).heartBeatInterval(1, SECONDS))
|
||||
.create(persistenceFile);
|
||||
}
|
||||
|
||||
|
||||
@@ -23,8 +23,6 @@ import java.io.Closeable;
|
||||
import java.io.IOException;
|
||||
import java.net.Inet4Address;
|
||||
|
||||
import static net.openhft.chronicle.map.Replicators.udp;
|
||||
|
||||
/**
|
||||
* Test ReplicatedChronicleMap where the Replicated is over a TCP Socket
|
||||
*
|
||||
@@ -39,12 +37,12 @@ public class UDPSocketReplicationTest {
|
||||
final int identifier,
|
||||
final int udpPort) throws IOException {
|
||||
|
||||
UdpReplicationConfig udpConfig = UdpReplicationConfig
|
||||
UdpReplicationReplicatorConfig udpConfig = UdpReplicationReplicatorConfig
|
||||
.simple(Inet4Address.getByName("255.255.255.255"), udpPort);
|
||||
|
||||
return ChronicleMapBuilder.of(Integer.class, CharSequence.class)
|
||||
.entries(1000)
|
||||
.replicators((byte) identifier, udp(udpConfig))
|
||||
.replicators((byte) identifier, udpConfig)
|
||||
.create(Builder.getPersistenceFile());
|
||||
}
|
||||
|
||||
|
||||
@@ -27,8 +27,7 @@ import java.net.InetSocketAddress;
|
||||
|
||||
import static java.util.concurrent.TimeUnit.SECONDS;
|
||||
import static net.openhft.chronicle.map.Builder.getPersistenceFile;
|
||||
import static net.openhft.chronicle.map.Replicators.tcp;
|
||||
import static net.openhft.chronicle.map.TcpReplicationConfig.of;
|
||||
import static net.openhft.chronicle.map.TcpReplicationReplicatorConfig.of;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
@@ -51,8 +50,8 @@ public class SetForTCPSocketReplicationTest3way {
|
||||
final int serverPort,
|
||||
final InetSocketAddress... endpoints) throws IOException {
|
||||
return (T) ChronicleSetBuilder.of(Integer.class)
|
||||
.replicators(identifier, tcp(of(serverPort, endpoints).
|
||||
heartBeatInterval(1L, SECONDS))).create(getPersistenceFile());
|
||||
.replicators(identifier, of(serverPort, endpoints).
|
||||
heartBeatInterval(1L, SECONDS)).create(getPersistenceFile());
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user