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

Moved name() from TcpTransportAndNetworkConfig to AbstractReplication

Browse Source
This commit is contained in:
leventov
2015-02-03 22:37:41 +07:00
parent 8ed1c8d2af
commit 5fe9e686d2
12 changed files with 126 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/replication/AbstractReplication.java
+24 -11
View File
@@ -36,6 +36,7 @@ public abstract class AbstractReplication implements Serializable {
@Nullable
transient RemoteNodeValidator remoteNodeValidator;
private final boolean bootstrapOnlyLocalEntries;
private final String name;
// package-private to forbid subclassing from outside of the package
AbstractReplication(byte localIdentifier, Builder builder) {
@@ -44,6 +45,7 @@ public abstract class AbstractReplication implements Serializable {
udpConfig = builder.udpConfig;
remoteNodeValidator = builder.remoteNodeValidator;
bootstrapOnlyLocalEntries = builder.bootstrapOnlyLocalEntries;
name = builder.name;
}
@Override
@@ -52,7 +54,8 @@ public abstract class AbstractReplication implements Serializable {
", tcpConfig=" + tcpConfig +
", udpConfig=" + udpConfig +
", remoteNodeValidator=" + remoteNodeValidator +
", bootstrapOnlyLocalEntries=" + bootstrapOnlyLocalEntries;
", bootstrapOnlyLocalEntries=" + bootstrapOnlyLocalEntries +
", name=" + name;
}
public byte identifier() {
@@ -77,6 +80,10 @@ public abstract class AbstractReplication implements Serializable {
public boolean bootstrapOnlyLocalEntries() {
return bootstrapOnlyLocalEntries;
}
public String name() {
return name;
}
/**
* Builder of {@link AbstractReplication} configurations. This class and it's subclasses are
@@ -93,6 +100,7 @@ public abstract class AbstractReplication implements Serializable {
private UdpTransportConfig udpConfig = null;
private RemoteNodeValidator remoteNodeValidator = null;
private boolean bootstrapOnlyLocalEntries = false;
private String name = "(unknown)";
// package-private to forbid subclassing from outside of the package
Builder() {
@@ -146,6 +154,17 @@ public abstract class AbstractReplication implements Serializable {
return (B) this;
}
/**
* Configures replication "name", which could appear in log messages and replication
* thread names. Default is {@code "(unknown)"}.
*
* @param name human-readable replication name
* @return this builder back
*/
public B name(String name) {
this.name = name;
return (B) this;
}
/**
* Creates a Replication instance with the given node (server) identifier.
@@ -153,10 +172,10 @@ public abstract class AbstractReplication implements Serializable {
* @param identifier the node (server) identifier of the returned replication
* @return a new Replication instance with the specified node (server) identifier
* @throws IllegalArgumentException if the given identifier is non-positive
* @throws IllegalStateException if neither {@link #tcpTransportAndNetwork(TcpTransportAndNetworkConfig)}
* nor {@link #udpTransport(UdpTransportConfig)} are
* configured to non-{@code null}. At least one of the
* transport-level configs should be specified.
* @throws IllegalStateException if neither {@link #tcpTransportAndNetwork(
* TcpTransportAndNetworkConfig)} nor {@link #udpTransport(UdpTransportConfig)} are
* configured to non-{@code null}. At least one of the transport-level configs
* should be specified.
*/
@NotNull
public abstract R createWithId(byte identifier);
@@ -166,11 +185,5 @@ public abstract class AbstractReplication implements Serializable {
throw new IllegalArgumentException("Identifier must be positive, " + identifier +
" given");
}
@Override
public String toString() {
return ", udpConfig=" + udpConfig +
", remoteNodeValidator=" + remoteNodeValidator;
}
}
}
src/main/java/net/openhft/chronicle/hash/replication/TcpTransportAndNetworkConfig.java
+1 -20
View File
@@ -42,7 +42,6 @@ public final class TcpTransportAndNetworkConfig implements Serializable {
private final ThrottlingConfig throttlingConfig;
private final long heartBeatInterval;
private final TimeUnit heartBeatIntervalUnit;
private String name = "(unknown)";
private TcpTransportAndNetworkConfig(int serverPort, Set<InetSocketAddress> endpoints,
int tcpBufferSize,
@@ -138,16 +137,6 @@ public final class TcpTransportAndNetworkConfig implements Serializable {
return tcpBufferSize;
}
/**
* @param name the name used here is reflected in the Tcp Replicator and is typically used when
* debugging
* @return self
*/
public TcpTransportAndNetworkConfig name(String name) {
this.name = name;
return this;
}
public TcpTransportAndNetworkConfig tcpBufferSize(int tcpBufferSize) {
if (tcpBufferSize <= 0)
throw new IllegalArgumentException();
@@ -200,8 +189,7 @@ public final class TcpTransportAndNetworkConfig implements Serializable {
@Override
public String toString() {
return "TcpReplicationConfig{" +
"name=" + name() +
", serverPort=" + serverPort +
"serverPort=" + serverPort +
", endpoints=" + endpoints +
", tcpBufferSize=" + tcpBufferSize +
", autoReconnectedUponDroppedConnection=" + autoReconnectedUponDroppedConnection +
@@ -210,11 +198,4 @@ public final class TcpTransportAndNetworkConfig implements Serializable {
", heartBeatIntervalUnit=" + heartBeatIntervalUnit +
'}';
}
/**
* @return the name of this configuration
*/
public String name() {
return this.name;
}
}
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
+1 -1
View File
@@ -66,7 +66,7 @@ final class ChannelProvider implements Closeable {
final TcpReplicator tcpReplicator = new TcpReplicator(
channelProvider.asReplica,
channelProvider.asEntryExternalizable,
tcpConfig, hub.remoteNodeValidator(), null);
tcpConfig, hub.remoteNodeValidator(), null, hub.name());
channelProvider.add(tcpReplicator);
}
src/main/java/net/openhft/chronicle/map/Replicators.java
+2 -1
View File
@@ -126,7 +126,8 @@ final class Replicators {
return new TcpReplicator(replica, entryExternalizable,
tcpConfig,
replication.remoteNodeValidator(), statelessClientParameters);
replication.remoteNodeValidator(), statelessClientParameters,
replication.name());
}
};
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+3 -2
View File
@@ -123,7 +123,8 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
@NotNull final Replica.EntryExternalizable externalizable,
@NotNull final TcpTransportAndNetworkConfig replicationConfig,
@Nullable final RemoteNodeValidator remoteNodeValidator,
@Nullable final StatelessClientParameters statelessClientParameters)
@Nullable final StatelessClientParameters statelessClientParameters,
String name)
throws IOException {
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig());
@@ -144,7 +145,7 @@ final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Clo
this.replicationConfig = replicationConfig;
this.remoteNodeValidator = remoteNodeValidator;
this.name = replicationConfig.name();
this.name = name;
start();
}
src/test/java/eg/BGChronicleTest.java
+25 -11
View File
@@ -1,5 +1,6 @@
package eg;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
@@ -38,18 +39,21 @@ public class BGChronicleTest {
public Closeable startChronicleMapServer(int port) throws InterruptedException, IOException {
TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(port)
.heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS)
.name("serverMap")
.autoReconnectedUponDroppedConnection(true);
File file = File.createTempFile("bg-test", ".deleteme");
file.deleteOnExit();
final ChronicleMap<byte[], byte[]> replica1 = ChronicleMapBuilder.of(byte[].class, byte[].class)
final ChronicleMap<byte[], byte[]> replica1 =
ChronicleMapBuilder.of(byte[].class, byte[].class)
.entries(KEYS)
.constantKeySizeBySample(new byte[KEY_SIZE])
.constantValueSizeBySample(new byte[VALUE_SIZE])
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) 1, serverConfig)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(serverConfig)
.name("serverMap")
.createWithId((byte) 1))
.createPersistedTo(file);
if (REPLICAS == 1)
@@ -59,17 +63,22 @@ public class BGChronicleTest {
file2.deleteOnExit();
final InetSocketAddress serverHostPort = new InetSocketAddress("localhost", port);
final TcpTransportAndNetworkConfig clientConfig = TcpTransportAndNetworkConfig.of(port + 1, serverHostPort)
.heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("clientMap")
final TcpTransportAndNetworkConfig clientConfig =
TcpTransportAndNetworkConfig.of(port + 1, serverHostPort)
.heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS)
.autoReconnectedUponDroppedConnection(true);
final ChronicleMap<byte[], byte[]> replica2 = ChronicleMapBuilder.of(byte[].class, byte[].class)
final ChronicleMap<byte[], byte[]> replica2 =
ChronicleMapBuilder.of(byte[].class, byte[].class)
.entries(KEYS)
.constantKeySizeBySample(new byte[KEY_SIZE])
.constantValueSizeBySample(new byte[VALUE_SIZE])
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) 2, clientConfig)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(clientConfig)
.name("clientMap")
.createWithId((byte) 2))
.createPersistedTo(file2);
return new Closeable() {
@@ -261,8 +270,9 @@ public class BGChronicleTest {
final InetSocketAddress serverHostPort = new InetSocketAddress(hostNameOrIpAddress, port);
final TcpTransportAndNetworkConfig clientConfig = TcpTransportAndNetworkConfig.of(port + 1, serverHostPort)
.heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS).name("clientMap")
final TcpTransportAndNetworkConfig clientConfig =
TcpTransportAndNetworkConfig.of(port + 1, serverHostPort)
.heartBeatInterval(CHRONICLE_HEARTBEAT_SECONDS, TimeUnit.SECONDS)
.autoReconnectedUponDroppedConnection(true);
System.out.println("clients connecting to " + serverHostPort);
@@ -276,13 +286,17 @@ public class BGChronicleTest {
futureList.add(es.submit(new Callable<Void>() {
@Override
public Void call() throws IOException {
try (ChronicleMap<byte[], byte[]> map = ChronicleMapBuilder.of(byte[].class, byte[].class)
try (ChronicleMap<byte[], byte[]> map =
ChronicleMapBuilder.of(byte[].class, byte[].class)
.entries(KEYS)
.constantKeySizeBySample(new byte[KEY_SIZE])
.constantValueSizeBySample(new byte[VALUE_SIZE])
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) (10 + clientId), clientConfig)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(clientConfig)
.name("clientMap")
.createWithId((byte) (10 + clientId)))
.create()) {
System.out.println((clientId + 1) + "/" + CLIENTS + " client started");
src/test/java/net/openhft/chronicle/map/CHMUseCasesTest.java
+13 -6
View File
@@ -2,6 +2,7 @@ package net.openhft.chronicle.map;
import com.google.common.primitives.Chars;
import net.openhft.chronicle.hash.function.Function;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.map.fromdocs.BondVOInterface;
import net.openhft.lang.io.ByteBufferBytes;
@@ -248,13 +249,15 @@ public class CHMUseCasesTest {
map2 = null;
{
final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig
.of(8086).name("server")
.of(8086)
.heartBeatInterval(1, TimeUnit.SECONDS)
.tcpBufferSize(1024 * 64);
map2 = builder
.replication((byte) 1, tcpConfig1)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfig1).name("server")
.createWithId((byte) 1))
.instance()
.name("server")
.create();
@@ -263,12 +266,14 @@ public class CHMUseCasesTest {
}
{
final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of
(8087, new InetSocketAddress("localhost", 8086)).name("map2")
(8087, new InetSocketAddress("localhost", 8086))
.heartBeatInterval(1, TimeUnit.SECONDS)
.tcpBufferSize(1024 * 64);
map1 = builder
.replication((byte) 2, tcpConfig2)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfig2).name("map2")
.createWithId((byte) 2))
.instance()
.name("map2")
.create();
@@ -284,13 +289,15 @@ public class CHMUseCasesTest {
case STATELESS: {
{
final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig
.of(8086).name("server")
.of(8086)
.heartBeatInterval(1, TimeUnit.SECONDS)
.tcpBufferSize(1024 * 64);
map2 = builder
.replication((byte) 1, tcpConfig1)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfig1).name("server")
.createWithId((byte) 1))
.instance()
.name("server")
.create();
src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java
+6 -3
View File
@@ -20,6 +20,7 @@ package net.openhft.chronicle.map;
import com.google.common.collect.HashBiMap;
import com.google.common.primitives.Ints;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.model.DataValueGenerator;
@@ -1689,8 +1690,7 @@ public class ChronicleMapTest {
@Test
public void testByteArrayKeySizeBySample() throws IOException {
TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(8877)
.name("serverMap");
TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(8877);
File mapFile = getPersistenceFile();
@@ -1698,7 +1698,10 @@ public class ChronicleMapTest {
for (int i = 0; i < 2; i++) {
try (ChronicleMap server = ChronicleMapBuilder.of(byte[].class, byte[][].class)
.replication((byte) 1, serverConfig)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(serverConfig)
.name("serverMap")
.createWithId((byte) 1))
.constantKeySizeBySample(new byte[14])
.createPersistedTo(mapFile)) {
src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
+5 -4
View File
@@ -20,6 +20,7 @@ package net.openhft.chronicle.map;
import junit.framework.Assert;
import net.openhft.chronicle.hash.function.Function;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
@@ -751,11 +752,11 @@ public class StatelessClientTest {
@Test
public void testsConstantKeySizeBySampleIsSentFromServer() throws IOException {
TcpTransportAndNetworkConfig serverConfig = TcpTransportAndNetworkConfig.of(8875)
.name("serverMap");
try (ChronicleMap server = ChronicleMapBuilder.of(byte[].class, CharSequence.class)
.replication((byte) 1, serverConfig)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(TcpTransportAndNetworkConfig.of(8875))
.name("serverMap")
.createWithId((byte) 1))
.constantKeySizeBySample(new byte[14])
.create()) {
src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java
+24 -9
View File
@@ -18,6 +18,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import org.junit.After;
import org.junit.Before;
@@ -90,22 +91,32 @@ public class TCPSocketReplicationBootStrapTest {
@Test
public void testBootstrapAndHeartbeat() throws IOException, InterruptedException {
TcpTransportAndNetworkConfig map1Config = TcpTransportAndNetworkConfig.of(8068, Arrays.asList(new InetSocketAddress("localhost", 8067)))
.heartBeatInterval(1L, TimeUnit.SECONDS).name("map1").autoReconnectedUponDroppedConnection
(true);
TcpTransportAndNetworkConfig map1Config = TcpTransportAndNetworkConfig
.of(8068, Arrays.asList(new InetSocketAddress("localhost", 8067)))
.heartBeatInterval(1L, TimeUnit.SECONDS)
.autoReconnectedUponDroppedConnection(true);
map1 = (ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?>) ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.replication((byte) 1, map1Config).create();
map1 = (ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?>)
ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(map1Config)
.name("map1")
.createWithId((byte) 1))
.create();
File persistenceFile = getPersistenceFile();
TcpTransportAndNetworkConfig map2Config = TcpTransportAndNetworkConfig.of(8067)
.heartBeatInterval(1L, TimeUnit.SECONDS).name("map2");
.heartBeatInterval(1L, TimeUnit.SECONDS);
final ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?> map2a =
(ReplicatedChronicleMap<Integer, ?, ?, CharSequence, ?, ?>)
ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.replication((byte) 2, map2Config).createPersistedTo(persistenceFile);
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(map2Config)
.name("map2")
.createWithId((byte) 2))
.createPersistedTo(persistenceFile);
map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2
@@ -130,9 +141,13 @@ public class TCPSocketReplicationBootStrapTest {
// now restart map2a and doConnect it to map1, map1 should bootstrap the missing entry
TcpTransportAndNetworkConfig tcpConfigNewMap2 = TcpTransportAndNetworkConfig.of(8067)
.heartBeatInterval(1L, TimeUnit.SECONDS).name("newMap2");
.heartBeatInterval(1L, TimeUnit.SECONDS);
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.replication((byte) 2, tcpConfigNewMap2).createPersistedTo(persistenceFile);
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfigNewMap2)
.name("newMap2")
.createWithId((byte) 2))
.createPersistedTo(persistenceFile);
// add data into it
waitTillEqual(5000);
src/test/java/net/openhft/chronicle/map/TcpReplicationLargeEntryTest.java
+11 -4
View File
@@ -1,5 +1,6 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.lang.MemoryUnit;
import org.junit.After;
@@ -40,7 +41,7 @@ public class TcpReplicationLargeEntryTest {
{
final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port,
endpoint).autoReconnectedUponDroppedConnection(true).name(" map1")
endpoint).autoReconnectedUponDroppedConnection(true)
.heartBeatInterval(1, TimeUnit.SECONDS)
.tcpBufferSize(1024 * 64);
@@ -49,21 +50,27 @@ public class TcpReplicationLargeEntryTest {
.constantValueSizeBySample(sampleValue)
.entries(2)
.actualSegments(1)
.replication((byte) 1, tcpConfig1)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfig1)
.name("map1")
.createWithId((byte) 1))
.instance()
.name("map1")
.create();
}
{
final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of
(s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2")
(s_port + 1).autoReconnectedUponDroppedConnection(true)
.heartBeatInterval(1, TimeUnit.SECONDS)
.tcpBufferSize(1024 * 64);
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.constantValueSizeBySample(sampleValue)
.entries(2)
.replication((byte) 2, tcpConfig2)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfig2)
.name("map2")
.createWithId((byte) 2))
.instance()
.name("map2")
.create();
src/test/java/net/openhft/chronicle/map/TcpReplicationSoakTest.java
+11 -4
View File
@@ -1,5 +1,6 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.SingleChronicleHashReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.model.Byteable;
@@ -38,7 +39,7 @@ public class TcpReplicationSoakTest {
{
final TcpTransportAndNetworkConfig tcpConfig1 = TcpTransportAndNetworkConfig.of(s_port,
endpoint).autoReconnectedUponDroppedConnection(true).name(" map1")
endpoint).autoReconnectedUponDroppedConnection(true)
.heartBeatInterval(1, TimeUnit.SECONDS)
.tcpBufferSize(1024 * 64);
@@ -46,20 +47,26 @@ public class TcpReplicationSoakTest {
map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(Builder.SIZE + Builder.SIZE)
.actualSegments(1)
.replication((byte) 1, tcpConfig1)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfig1)
.name("map1")
.createWithId((byte) 1))
.instance()
.name("map1")
.create();
}
{
final TcpTransportAndNetworkConfig tcpConfig2 = TcpTransportAndNetworkConfig.of
(s_port + 1).autoReconnectedUponDroppedConnection(true).name("map2")
(s_port + 1).autoReconnectedUponDroppedConnection(true)
.heartBeatInterval(1, TimeUnit.SECONDS)
.tcpBufferSize(1024 * 64);
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(Builder.SIZE + Builder.SIZE)
.replication((byte) 2, tcpConfig2)
.replication(SingleChronicleHashReplication.builder()
.tcpTransportAndNetwork(tcpConfig2)
.name("map2")
.createWithId((byte) 2))
.instance()
.name("map2")
.create();