diff --git a/README.md b/README.md index bcbbbb674..f0b4706ff 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,7 @@ Click here to get the [Latest Version Number](http://search.maven.org/#search%7C * [Identifier](https://github.com/OpenHFT/Chronicle-Map#identifier) * [Port](https://github.com/OpenHFT/Chronicle-Map#port) * [Heart Beat Interval](https://github.com/OpenHFT/Chronicle-Map#heart-beat-interval) -* [Clustering](https://github.com/OpenHFT/Chronicle-Map#cluster) +* [Channels and the Channel Provider](https://github.com/OpenHFT/Chronicle-Map#channels-and-channelprovider) #### Miscellaneous @@ -63,7 +63,6 @@ Click here to get the [Latest Version Number](http://search.maven.org/#search%7C * [Replicating data between process on different servers with UDP] (https://github.com/OpenHFT/Chronicle-Map/blob/master/README.md#example--replicating-data-between-process-on-different-servers-using-udp) * [Creating a Chronicle Set and adding data to it](https://github.com/OpenHFT/Chronicle-Map/blob/master/README.md#example--creating-a-chronicle-set-and-adding-data-to-it) - #### Performance Topics * [Chronicle Map with Large Data ](https://github.com/OpenHFT/Chronicle-Map#chronicle-map-with-large-data) @@ -71,12 +70,10 @@ Click here to get the [Latest Version Number](http://search.maven.org/#search%7C * [Better to use small keys](https://github.com/OpenHFT/Chronicle-Map#better-to-use-small-keys) * [ConcurrentHashMap v ChronicleMap](https://github.com/OpenHFT/Chronicle-Map#concurrenthashmap-v-chroniclemap) - ### Overview Chronicle Map implements the `java.util.concurrent.ConcurrentMap`, however unlike the standard java map, ChronicleMap is able to share your entries accross processes: - ![](http://openhft.net/wp-content/uploads/2014/07/Chronicle-Map-diagram_04.jpg) ## When to use @@ -512,7 +509,7 @@ A heartbeat will only be send if no data is transmitted, if the maps are constan no heartbeat message is sent. If a map does not receive either data of a heartbeat the connection is dropped and re-established. -# Cluster +# Channels and ChannelProvider Chronicle Map TCP Replication lets you distribute a single Chronicle Map, to a number of servers across your network. Replication is point to point and the data transfer is bidirectional, so in the @@ -520,37 +517,42 @@ example of just two servers, they only have to be connected via a single tcp soc the data is transferred both ways. Which is great, however what if you wanted to replicate more than just one chronicle map, what if you were going to replicate two chronicle maps across your network, unfortunately with just TCP replication you would have to have two tcp socket connections, which is -not ideal. This is why we created Chronicle Clustering. Clustering lets you replicate numerous +not ideal. This is why we created Chronicle Channels. Channels let you replicate numerous Chronicle Maps via a single point to point socket connection. -Clustering is similar to TCP replication, where each map has to be given a unique identifier, but -when using Chronicle Clustering its the cluster that is given the unique identifier not the map. +Chronicle Channels are similar to TCP replication, where each map has to be given a unique identifier, but +when using Chronicle Channels its the channels that are given the unique identifier not the map. ``` java -ReplicatingClusterBuilder clusterBuilder = new ReplicatingClusterBuilder((byte) 2, 1024); +byte identifier = 2; +int maxEntrySize = 1024; +ChannelProvider replicator = new ChannelProviderBuilder(identifier, maxEntrySize) + .create(); ``` -In this example above the cluster is given the identifier of 2 +In this example above the channel is given the identifier of 2 In addition to specifying the identifier we also have to set the maximum entry size, this sets -the size of the memory buffers within the cluster. This has to be set manually, with clusters you -are able to attach additional maps to a cluster once its up and running, so the maximum size of each +the size of the memory buffers within the ChannelProvider. This has to be set manually, with channels you +are able to attach additional maps to a ChannelProvider once its up and running, so the maximum size of each entry in the map can not be known in advance and we don’t currently support automatic resizing of buffers. -Once you have created the cluster you should attach your tcpConfig +Once you have created the ChannelProvider you should attach your tcp configuration ``` java -TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8087).heartBeatInterval(1, SECONDS); -clusterBuilder.tcpReplication(tcpConfig); -ReplicatingCluster cluster = clusterBuilder.create(); +byte identifier = 2; +int maxEntrySize =1024; +ChannelProvider replicator = new ChannelProviderBuilder(identifier, 1024) + .tcpReplication(tcpConfig) + .create();; ``` -Attaching cluster replication to the map: +Attaching ChannelProvider replication to the map: ``` java ChronicleMap map = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(cluster.channelReplicator((short) 1)) + .channel(replicator.createChannel((short) 1)) .create(file); ``` @@ -567,11 +569,11 @@ If you inadvertently got the chronicle channels around the wrong way, then chron to replicate the wrong maps data. The chronicle channels don't have to be in order but they must be unique for each map you have. -Once you have created the cluster you may wish to hold onto the reference so that you can call close -once you have finished, this will close everything in the cluster +Once you have created the ChannelProvider you may wish to hold onto the reference so that you can call close +once you have finished, this will close everything in the ChannelProvider ``` java -cluster.close(); +replicator.close(); ``` #### Known Issues diff --git a/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java index 53345e6b9..20cd595b5 100644 --- a/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java @@ -189,9 +189,7 @@ public interface ChronicleHashBuilder replicators = new CopyOnWriteArraySet(); - ReplicatingCluster(ReplicatingClusterBuilder builder) { + ChannelProvider(ChannelProviderBuilder builder) { localIdentifier = builder.identifier; maxEntrySize = builder.maxEntrySize; chronicleChannels = new Replica[builder.maxNumberOfChronicles]; @@ -206,12 +206,12 @@ public final class ReplicatingCluster implements Closeable { /** * Returns a replicator, dedicated to the specified channel. Channel is basically just a number, that * should correspond on different servers for instances of the same replicated map. Only one replicator - * per channel could be obtained from a single {@code ReplicatingCluster}. The returned replicator could + * per channel could be obtained from a single {@code ReplicatingChannel}. The returned replicator could * be applied to a map at most once. * * @return a replicator, dedicated to the specified channel */ - public Replicator channelReplicator(short channel) { + public ChronicleChannel createChannel(short channel) { return new ChronicleChannel(channel); } @@ -393,14 +393,14 @@ public final class ReplicatingCluster implements Closeable { } } - private class ChronicleChannel extends Replicator implements Closeable { + public class ChronicleChannel extends Replicator implements Closeable { + private final short chronicleChannel; private ChronicleChannel(short chronicleChannel) { this.chronicleChannel = chronicleChannel; } - @Override public byte identifier() { return localIdentifier; } @@ -409,7 +409,7 @@ public final class ReplicatingCluster implements Closeable { protected Closeable applyTo(ChronicleMapBuilder builder, Replica map, EntryExternalizable entryExternalizable) { if (builder.entrySize() > maxEntrySize) { - throw new IllegalArgumentException("During ReplicatingClusterBuilder setup, " + + throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " + "maxEntrySize=" + maxEntrySize + " was specified, but map with " + "entrySize=" + builder.entrySize() + " is attempted to apply" + "to the replicator"); diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatingClusterBuilder.java b/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java similarity index 67% rename from src/main/java/net/openhft/chronicle/map/ReplicatingClusterBuilder.java rename to src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java index c5a85fc82..ca1a00f00 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatingClusterBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java @@ -21,40 +21,48 @@ import java.io.IOException; /** * @author Rob Austin. */ -public final class ReplicatingClusterBuilder { +public final class ChannelProviderBuilder { - final byte identifier; - final int maxEntrySize; + byte identifier; + int maxEntrySize = 1024; int maxNumberOfChronicles = 128; private UdpReplicationConfig udpReplicationConfig = null; private TcpReplicationConfig tcpReplicationConfig = null; - ReplicatingClusterBuilder(byte identifier, final int maxEntrySize) { - this.identifier = identifier; + + + public int maxEntrySize() { + return maxEntrySize; + } + + public ChannelProviderBuilder maxEntrySize(int maxEntrySize) { this.maxEntrySize = maxEntrySize; - if (identifier <= 0) { - throw new IllegalArgumentException("Identifier must be positive, identifier=" + - identifier); + return this; + } + + public ChannelProviderBuilder replicators(byte identifier, Config... configs) { + this.identifier = identifier; + + + for (Config config : configs) { + + if (config instanceof TcpReplicationConfig) { + this.tcpReplicationConfig = (TcpReplicationConfig) config; + } else if (config instanceof UdpReplicationConfig) { + this.udpReplicationConfig = (UdpReplicationConfig) config; + } else + throw new UnsupportedOperationException(); } - } - - public ReplicatingClusterBuilder udpReplication(UdpReplicationConfig replicationConfig) { - this.udpReplicationConfig = replicationConfig; return this; } - public ReplicatingClusterBuilder tcpReplication(TcpReplicationConfig replicationConfig) { - this.tcpReplicationConfig = replicationConfig; - return this; - } - - public ReplicatingClusterBuilder maxNumberOfChronicles(int maxNumberOfChronicles) { + public ChannelProviderBuilder maxNumberOfChronicles(int maxNumberOfChronicles) { this.maxNumberOfChronicles = maxNumberOfChronicles; return this; } - public ReplicatingCluster create() throws IOException { - final ReplicatingCluster replicatingCluster = new ReplicatingCluster(this); + public ChannelProvider create() throws IOException { + final ChannelProvider replicatingCluster = new ChannelProvider(this); if (tcpReplicationConfig != null) { final TcpReplicator tcpReplicator = new TcpReplicator( replicatingCluster.asReplica, diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index fd3cfdda4..5160423da 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -27,9 +27,12 @@ import net.openhft.chronicle.map.serialization.MetaProvider; import net.openhft.chronicle.map.threadlocal.Provider; import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import net.openhft.lang.Maths; -import net.openhft.lang.io.*; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.BytesStore; +import net.openhft.lang.io.DirectStore; import net.openhft.lang.io.serialization.*; -import net.openhft.lang.io.serialization.impl.*; +import net.openhft.lang.io.serialization.impl.VanillaBytesMarshallerFactory; import net.openhft.lang.model.Byteable; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; @@ -44,8 +47,8 @@ import java.util.concurrent.TimeUnit; import static net.openhft.chronicle.map.Objects.builderEquals; /** - * {@code ChronicleMapBuilder} manages the whole set of {@link ChronicleMap} configurations, could - * be used as a classic builder and/or factory. This means that in addition to the standard builder + * {@code ChronicleMapBuilder} manages the whole set of {@link ChronicleMap} configurations, could be used as + * a classic builder and/or factory. This means that in addition to the standard builder * usage pattern: 
{@code
  * ChronicleMap map = ChronicleMapBuilder
  *     .of(Key.class, Value.class)
@@ -63,25 +66,22 @@ import static net.openhft.chronicle.map.Objects.builderEquals;
  *
  * 
Use static {@link #of(Class, Class)} method to obtain a {@code ChronicleMapBuilder} instance.
  *
- * 
{@code ChronicleMapBuilder} is mutable. Configuration methods mutate the builder and
- * return the builder itself back to support chaining pattern, rather than the builder copies
- * with the corresponding configuration changed. To make an independent configuration,
- * {@linkplain #clone} the builder.
+ * 
{@code ChronicleMapBuilder} is mutable. Configuration methods mutate the builder and return the
+ * builder itself back to support chaining pattern, rather than the builder copies with the corresponding
+ * configuration changed. To make an independent configuration, {@linkplain #clone} the builder.
  *
- * 
Later in this documentation, "ChronicleMap" means
- * "ChronicleMaps, created by {@code ChronicleMapBuilder}", unless specified different, because
- * theoretically someone might provide {@code ChronicleMap} implementations with completely
- * different properties.
+ * 
Later in this documentation, "ChronicleMap" means "ChronicleMaps, created by {@code
+ * ChronicleMapBuilder}", unless specified different, because theoretically someone might provide {@code
+ * ChronicleMap} implementations with completely different properties.
  *
- * 
{@code ChronicleMap} ("ChronicleMaps, created by {@code ChronicleMapBuilder}") currently
- * doesn't support resizing. That is why you should always configure {@linkplain
- * #entries(long) number of entries} you are going to insert into the created map at most.
- * See {@link #entries(long)} method documentation for more information on this.
+ * 
{@code ChronicleMap} ("ChronicleMaps, created by {@code ChronicleMapBuilder}") currently doesn't support
+ * resizing. That is why you should always configure {@linkplain #entries(long) number of entries} you
+ * are going to insert into the created map at most. See {@link #entries(long)} method documentation
+ * for more information on this.
  *
- * 
{@code ChronicleMap} allocates memory by equally sized chunks. This size is called
- * {@linkplain #entrySize(int) entry size}, you are strongly recommended to configure it to achieve
- * least memory consumption and best speed. See {@link #entrySize(int)} method documentation for
- * more information on this.
+ * 
{@code ChronicleMap} allocates memory by equally sized chunks. This size is called {@linkplain
+ * #entrySize(int) entry size}, you are strongly recommended to configure it to achieve least memory
+ * consumption and best speed. See {@link #entrySize(int)} method documentation for more information on this.
  *
  * @param  key type of the maps, produced by this builder
  * @param  value type of the maps, produced by this builder
@@ -94,11 +94,12 @@ public class ChronicleMapBuilder implements Cloneable,
 
     public static final short UDP_REPLICATION_MODIFICATION_ITERATOR_ID = 128;
     private static final Logger LOG = LoggerFactory.getLogger(ChronicleMapBuilder.class.getName());
+
     SerializationBuilder keyBuilder;
     SerializationBuilder valueBuilder;
-    Replicator firstReplicator;
-    Map, Replicator> replicators = new HashMap, Replicator>();
-    boolean forceReplicatedImpl = false;
+
+    private Map, Replicator> replicators = new HashMap, Replicator>();
+
     // used when configuring the number of segments.
     private int minSegments = -1;
     private int actualSegments = -1;
@@ -126,6 +127,7 @@ public class ChronicleMapBuilder implements Cloneable,
             MapEventListeners.nop();
     private V defaultValue = null;
     private DefaultValueProvider defaultValueProvider = null;
+    private byte identifier = -1;
 
     ChronicleMapBuilder(Class keyClass, Class valueClass) {
         keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY);
@@ -156,6 +158,7 @@ public class ChronicleMapBuilder implements Cloneable,
         }
     }
 
+
     @Override
     public ChronicleMapBuilder minSegments(int minSegments) {
         this.minSegments = minSegments;
@@ -185,9 +188,9 @@ public class ChronicleMapBuilder implements Cloneable,
      *     .keySize(10)
      *     // shouldn't specify valueSize(), because it is statically known
      *     .create();}
- * (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 ChronicleMap}.) + * (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 + * ChronicleMap}.) * * @see #constantKeySizeBySample(Object) * @see #valueSize(int) @@ -224,17 +227,16 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Configures the optimal number of bytes, taken by serialized form of values, put into maps, - * created by this builder. If value size is always the same, call - * {@link #constantValueSizeBySample(Object)} method instead of this one. + * Configures the optimal number of bytes, taken by serialized form of values, put into maps, created by + * this builder. If value size is always the same, call {@link #constantValueSizeBySample(Object)} method + * instead of this one. * - *

If value is a boxed primitive type or {@link Byteable} subclass, i. e. if value size - * is known statically, it is automatically accounted and shouldn't be specified by user. + *

If value is a boxed primitive type or {@link Byteable} subclass, i. e. if value size is known + * statically, it is automatically accounted and shouldn't be specified by user. * - *

If value size varies moderately, specify the size higher than average, but lower than - * the maximum possible, to minimize average memory overuse. If value size varies in a wide - * range, it's better to use {@linkplain #entrySize(int) entry size} in "chunk" mode and - * configure it directly. + *

If value size varies moderately, specify the size higher than average, but lower than the maximum + * possible, to minimize average memory overuse. If value size varies in a wide range, it's better to use + * {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it directly. * * @param valueSize number of bytes, taken by serialized form of values * @return this {@code ChronicleMapBuilder} back @@ -249,15 +251,15 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Configures the constant number of bytes, taken by serialized form of values, put into maps, - * created by this builder. This is done by providing the {@code sampleValue}, all values should - * take the same number of bytes in serialized form, as this sample object. + * Configures the constant number of bytes, taken by serialized form of values, put into maps, created by + * this builder. This is done by providing the {@code sampleValue}, all values should take the same number + * of bytes in serialized form, as this sample object. * - *

If values are of boxed primitive type or {@link Byteable} subclass, i. e. if value size - * is known statically, it is automatically accounted and this method shouldn't be called. + *

If values are of boxed primitive type or {@link Byteable} subclass, i. e. if value size is known + * statically, it is automatically accounted and this method shouldn't be called. * - *

If value size varies, method {@link #valueSize(int)} or {@link #entrySize(int)} should be - * called instead of this one. + *

If value size varies, method {@link #valueSize(int)} or {@link #entrySize(int)} should be called + * instead of this one. * * @param sampleValue the sample value * @return this builder back @@ -286,18 +288,18 @@ public class ChronicleMapBuilder implements Cloneable, * {@inheritDoc} * *

In fully default case you can expect entry size to be about 256 bytes. But it is strongly - * recommended always to configure {@linkplain #keySize(int) key size} and - * {@linkplain #valueSize(int) value size}, if they couldn't be derived statically. + * recommended always to configure {@linkplain #keySize(int) key size} and {@linkplain #valueSize(int) + * value size}, if they couldn't be derived statically. * - *

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 {@linkplain #valueSize(int) value size}, plus a few bytes required by - * implementations, with respect to {@linkplain #entryAndValueAlignment(Alignment) alignment}. + *

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 + * {@linkplain #valueSize(int) value size}, plus a few bytes required by implementations, with respect to + * {@linkplain #entryAndValueAlignment(Alignment) alignment}. * *

Note that the actual entrySize will be aligned to 4 (default {@linkplain - * #entryAndValueAlignment(Alignment) entry alignment}). I. e. if you set entry size to 30, - * the actual entry size will be 32 (30 aligned to 4 bytes). If you don't want entry size - * to be aligned, set {@code entryAndValueAlignment(Alignment.NO_ALIGNMENT)}. + * #entryAndValueAlignment(Alignment) entry alignment}). I. e. if you set entry size to 30, the actual + * entry size will be 32 (30 aligned to 4 bytes). If you don't want entry size to be aligned, set {@code + * entryAndValueAlignment(Alignment.NO_ALIGNMENT)}. * * @see #entryAndValueAlignment(Alignment) * @see #entries(long) @@ -317,7 +319,7 @@ public class ChronicleMapBuilder implements Cloneable, int keySize = keySize(); size += keyBuilder.sizeMarshaller().sizeEncodingSize(keySize); size += keySize; - if (replicatedImpl()) + if (useReplication()) size += ReplicatedChronicleMap.ADDITIONAL_ENTRY_BYTES; int valueSize = valueSize(); size += valueBuilder.sizeMarshaller().sizeEncodingSize(valueSize); @@ -339,16 +341,16 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Configures alignment strategy of address in memory of entries and independently of address - * in memory of values within entries in ChronicleMaps, created by this builder. + * Configures alignment strategy of address in memory of entries and independently of address in memory of + * values within entries in ChronicleMaps, created by this builder. * - *

Useful when values of the map are updated intensively, particularly fields with - * volatile access, because it doesn't work well if the value crosses cache lines. Also, on some - * (nowadays rare) architectures any misaligned memory access is more expensive than aligned. + *

Useful when values of the map are updated intensively, particularly fields with volatile access, + * because it doesn't work well if the value crosses cache lines. Also, on some (nowadays rare) + * architectures any misaligned memory access is more expensive than aligned. * - *

Note that {@linkplain #entrySize(int) entry size} will be aligned according to this - * alignment. I. e. if you set {@code entrySize(20)} and {@link Alignment#OF_8_BYTES}, actual - * entry size will be 24 (20 aligned to 8 bytes). + *

Note that {@linkplain #entrySize(int) entry size} will be aligned according to this alignment. I. e. + * if you set {@code entrySize(20)} and {@link Alignment#OF_8_BYTES}, actual entry size will be 24 (20 + * aligned to 8 bytes). * *

Default is {@link Alignment#OF_4_BYTES}. * @@ -406,7 +408,6 @@ public class ChronicleMapBuilder implements Cloneable, return (int) Maths.nextPower2(Math.max((entries >> 30) + 1, minSegments()), 1); } - @Override public ChronicleMapBuilder lockTimeOut(long lockTimeOut, TimeUnit unit) { this.lockTimeOut = lockTimeOut; lockTimeOutUnit = unit; @@ -428,20 +429,20 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Configures if the maps created by this builder should return {@code null} instead of previous - * mapped values on {@link ChronicleMap#put(Object, Object) ChornicleMap.put(key, value)} calls. + * Configures if the maps created by this builder should return {@code null} instead of previous mapped + * values on {@link ChronicleMap#put(Object, Object) ChornicleMap.put(key, value)} calls. * - *

{@link Map#put(Object, Object) Map.put()} returns the previous value, functionality which - * is rarely used but fairly cheap for {@link HashMap}. In the case, for an off heap collection, - * it has to create a new object and deserialize the data from off-heap memory. It's expensive - * for something you probably don't use. + *

{@link Map#put(Object, Object) Map.put()} returns the previous value, functionality which is rarely + * used but fairly cheap for {@link HashMap}. In the case, for an off heap collection, it has to create a + * new object and deserialize the data from off-heap memory. It's expensive for something you probably + * don't use. * - *

By default, of cause, {@code ChronicleMap} conforms the general {@code Map} contract - * and returns the previous mapped value on {@code put()} calls. + *

By default, of cause, {@code ChronicleMap} conforms the general {@code Map} contract and returns the + * previous mapped value on {@code put()} calls. * * @param putReturnsNull {@code true} if you want {@link ChronicleMap#put(Object, Object) - * ChronicleMap.put()} to not return the value that was replaced - * but instead return {@code null} + * ChronicleMap.put()} to not return the value that was replaced but instead return + * {@code null} * @return an instance of the map builder back * @see #removeReturnsNull(boolean) */ @@ -455,17 +456,17 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Configures if the maps created by this builder should return {@code null} instead of - * the last mapped value on {@link ChronicleMap#remove(Object) ChronicleMap.remove(key)} calls. + * Configures if the maps created by this builder should return {@code null} instead of the last mapped + * value on {@link ChronicleMap#remove(Object) ChronicleMap.remove(key)} calls. * - *

{@link Map#remove(Object) Map.remove()} returns the previous value, functionality which - * is rarely used but fairly cheap for {@link HashMap}. In the case, for an off heap collection, - * it has to create a new object and deserialize the data from off-heap memory. It's expensive - * for something you probably don't use. + *

{@link Map#remove(Object) Map.remove()} returns the previous value, functionality which is rarely + * used but fairly cheap for {@link HashMap}. In the case, for an off heap collection, it has to create a + * new object and deserialize the data from off-heap memory. It's expensive for something you probably + * don't use. * * @param removeReturnsNull {@code true} if you want {@link ChronicleMap#remove(Object) - * ChronicleMap.remove()} to not return the value of the removed entry - * but instead return {@code null} + * ChronicleMap.remove()} to not return the value of the removed entry but + * instead return {@code null} * @return an instance of the map builder back * @see #putReturnsNull(boolean) */ @@ -496,6 +497,7 @@ public class ChronicleMapBuilder implements Cloneable, return this; } + int metaDataBytes() { return metaDataBytes; } @@ -549,32 +551,23 @@ public class ChronicleMapBuilder implements Cloneable, return toString().hashCode(); } - @Override - public ChronicleMapBuilder addReplicator(Replicator replicator) { - if (firstReplicator == null) { - firstReplicator = replicator; - replicators.put(replicator.getClass(), replicator); - } else { - if (replicator.identifier() != firstReplicator.identifier()) { - throw new IllegalArgumentException( - "Identifiers of all replicators of the map should be the same"); - } - if (replicators.containsKey(replicator.getClass())) { - throw new IllegalArgumentException("Replicator of " + replicator.getClass() + - " class has already to the map"); - } - replicators.put(replicator.getClass(), replicator); + public ChronicleMapBuilder replicators(byte identifier, Replicator... replicators) { + + this.identifier = identifier; + this.replicators.clear(); + for (Replicator replicator : replicators) { + this.replicators.put(replicator.getClass(), replicator); } + return this; } - @Override - public ChronicleMapBuilder setReplicators(Replicator... replicators) { - firstReplicator = null; + + public ChronicleMapBuilder channel(ChannelProvider.ChronicleChannel chronicleChannel) { + this.identifier = chronicleChannel.identifier(); this.replicators.clear(); - for (Replicator replicator : replicators) { - addReplicator(replicator); - } + replicators.put(chronicleChannel.getClass(), chronicleChannel); + return this; } @@ -613,13 +606,6 @@ public class ChronicleMapBuilder implements Cloneable, return this; } - /** - * For testing - */ - ChronicleMapBuilder forceReplicatedImpl() { - this.forceReplicatedImpl = true; - return this; - } @Override public ChronicleMapBuilder keyMarshaller(@NotNull BytesMarshaller keyMarshaller) { @@ -662,19 +648,18 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Specifies the value to be put for each key queried in {@link ChronicleMap#get get()} and - * {@link ChronicleMap#getUsing(Object, Object) getUsing()} methods, if the key is absent - * in the map. Then this default value is returned from query method. + * Specifies the value to be put for each key queried in {@link ChronicleMap#get get()} and {@link + * ChronicleMap#getUsing(Object, Object) getUsing()} methods, if the key is absent in the map. Then this + * default value is returned from query method. * - *

Setting default value to {@code null} is interpreted as map shouldn't put any default - * value for absent keys. This is by default. + *

Setting default value to {@code null} is interpreted as map shouldn't put any default value for + * absent keys. This is by default. * - *

This configuration overrides any previous - * {@link #defaultValueProvider(DefaultValueProvider)} configuration to this - * {@code ChronicleMapBuilder}. + *

This configuration overrides any previous {@link #defaultValueProvider(DefaultValueProvider)} + * configuration to this {@code ChronicleMapBuilder}. * - * @param defaultValue the default value to be put to the map for absent keys during - * {@code get()} and {@code getUsing()} calls and returned from these calls + * @param defaultValue the default value to be put to the map for absent keys during {@code get()} and + * {@code getUsing()} calls and returned from these calls * @return this builder object back */ public ChronicleMapBuilder defaultValue(V defaultValue) { @@ -684,13 +669,13 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Specifies the function to obtain a value for the key during {@link ChronicleMap#get get()} - * and {@link ChronicleMap#getUsing(Object, Object) getUsing()} calls, if the key is absent - * in the map. If the obtained value is non-null, it is put for the key in the map and then - * returned from current {@code get()} or {@code getUsing()} call. + * Specifies the function to obtain a value for the key during {@link ChronicleMap#get get()} and {@link + * ChronicleMap#getUsing(Object, Object) getUsing()} calls, if the key is absent in the map. If the + * obtained value is non-null, it is put for the key in the map and then returned from current {@code + * get()} or {@code getUsing()} call. * - *

This configuration overrides any previous {@link #defaultValue(Object)} configuration - * to this {@code ChronicleMapBuilder}. + *

This configuration overrides any previous {@link #defaultValue(Object)} configuration to this {@code + * ChronicleMapBuilder}. * * @param defaultValueProvider the strategy to obtain a default value by the absent key * @return this builder object back @@ -778,7 +763,7 @@ public class ChronicleMapBuilder implements Cloneable, private VanillaChronicleMap newMap() throws IOException { preMapConstruction(); - if (!replicatedImpl()) { + if (!useReplication()) { return new VanillaChronicleMap, V, Object, MetaBytesWriter>(this); } else { @@ -787,8 +772,8 @@ public class ChronicleMapBuilder implements Cloneable, } } - private boolean replicatedImpl() { - return firstReplicator != null || forceReplicatedImpl; + private boolean useReplication() { + return identifier != -1; } private void preMapConstruction() { @@ -833,5 +818,9 @@ public class ChronicleMapBuilder implements Cloneable, return (int) Math.min(Math.max(2L, entries() >> 10), VanillaChronicleMap.MAX_ENTRY_OVERSIZE_FACTOR); } + + public byte identifier() { + return identifier; + } } diff --git a/src/main/java/net/openhft/chronicle/map/Config.java b/src/main/java/net/openhft/chronicle/map/Config.java new file mode 100644 index 000000000..6e0d9a436 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/Config.java @@ -0,0 +1,7 @@ +package net.openhft.chronicle.map; + +/** + * @author Rob Austin. + */ +public interface Config { +} diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index 90b15382b..26736295c 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -212,7 +212,7 @@ public class NodeDiscovery { return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000L) - .addReplicator(tcp(identifier, tcpConfig)).create(); + .replicators(identifier, tcp(tcpConfig)).create(); } diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index 821d4ab28..fcc698273 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -55,14 +55,11 @@ public interface Replica extends Closeable { ModificationNotifier modificationNotifier); /** - * Returns the timestamp of the last change from the specified remote node, already replicated - * to this Replica. - *

- *

Used in conjunction with replication, to back fill data from a remote node. This node + * Returns the timestamp of the last change from the specified remote node, already replicated to this + * Replica.

Used in conjunction with replication, to back fill data from a remote node. This node * may have missed updates while it was not been running or connected via TCP. * - * @param remoteIdentifier the identifier of the remote node to check last replicated update - * time from + * @param remoteIdentifier the identifier of the remote node to check last replicated update time from * @return a timestamp of the last modification to an entry, or 0 if there are no entries. * @see #identifier() */ @@ -102,7 +99,7 @@ public interface Replica extends Closeable { * A non-blocking call that provides the entry that has changed to {@code callback.onEntry()}. * * @param callback a callback which will be called when a new entry becomes available. - * @param chronicleId only assigned when clustering + * @param chronicleId only assigned when using chronicle channels * @return {@code true} if the entry was accepted by the {@code callback.onEntry()} method, {@code * false} if the entry was not accepted or was not available */ @@ -111,10 +108,8 @@ public interface Replica extends Closeable { /** * Dirties all entries with a modification time equal to {@code fromTimeStamp} or newer. It means all * these entries will be considered as "new" by this ModificationIterator and iterated once again no - * matter if they have already been. - *

- *

This functionality is used to publish recently modified entries to a new remote node as it - * connects. + * matter if they have already been.

This functionality is used to publish recently modified + * entries to a new remote node as it connects. * * @param fromTimeStamp the timestamp from which all entries should be dirty */ @@ -133,7 +128,7 @@ public interface Replica extends Closeable { * @param entry the byte location of the entry to be stored * @param destination a buffer the entry will be written to, the segment may reject this operation and * add zeroBytes, if the identifier in the entry did not match the maps local - * @param chronicleId used in cluster into identify the canonical map or queue + * @param chronicleId is the channel id used to identify the canonical map or queue */ void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, int chronicleId); @@ -180,20 +175,20 @@ public interface Replica extends Closeable { } /** - * Implemented typically by a replicator, This interface provides the event, which will get - * called whenever a put() or remove() has occurred to the map + * Implemented typically by a replicator, This interface provides the event, which will get called + * whenever a put() or remove() has occurred to the map */ abstract class EntryCallback { /** * Called whenever a put() or remove() has occurred to a replicating map. * - * @param entry the entry you will receive, this does not have to be locked, - * as locking is already provided from the caller. + * @param entry the entry you will receive, this does not have to be locked, as locking is + * already provided from the caller. * @param chronicleId only assigned when clustering - * @return {@code false} if this entry should be ignored because the identifier - * of the source node is not from one of our changes, WARNING even though we check - * the identifier in the ModificationIterator the entry may have been updated. + * @return {@code false} if this entry should be ignored because the identifier of the source node is + * not from one of our changes, WARNING even though we check the identifier in the + * ModificationIterator the entry may have been updated. */ public abstract boolean onEntry(final Bytes entry, final int chronicleId); diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index bcf0a5d0c..6c768546e 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -18,16 +18,16 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.TimeProvider; +import net.openhft.chronicle.map.serialization.BytesReader; +import net.openhft.chronicle.map.serialization.MetaBytesInterop; +import net.openhft.chronicle.map.serialization.MetaBytesWriter; +import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import net.openhft.lang.Maths; import net.openhft.lang.collection.ATSDirectBitSet; import net.openhft.lang.io.Bytes; import net.openhft.lang.io.MultiStoreBytes; import net.openhft.lang.io.NativeBytes; -import net.openhft.chronicle.map.serialization.BytesReader; -import net.openhft.chronicle.map.serialization.MetaBytesInterop; -import net.openhft.chronicle.map.serialization.MetaBytesWriter; import net.openhft.lang.model.Byteable; -import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -100,8 +100,13 @@ class ReplicatedChronicleMap, throws IOException { super(builder); this.timeProvider = builder.timeProvider(); - Replicator firstReplicator = builder.firstReplicator; - this.localIdentifier = firstReplicator != null ? firstReplicator.identifier() : 1; + + this.localIdentifier = builder.identifier(); + + if (localIdentifier==-1){ + throw new IllegalStateException("localIdentifier should not be -1"); + } + } private int assignedModIterBitSetSizeInBytes() { @@ -882,15 +887,15 @@ class ReplicatedChronicleMap, } /** - * Puts entry. If {@code value} implements {@link Byteable} interface and - * {@code usingValue} is {@code true}, the value is backed with the bytes of this entry. + * Puts entry. If {@code value} implements {@link Byteable} interface and {@code usingValue} is {@code + * true}, the value is backed with the bytes of this entry. * * @param hash2 a hash of the {@code keyBytes}. Caller was searching for the key in the * {@code searchedHashLookup} using this hash. * @param value the value to put * @param usingValue {@code true} if the value should be backed with the bytes of the entry, - * if it implements {@link Byteable} interface, - * {@code false} if it should put itself + * if it implements {@link Byteable} interface, {@code false} if it should + * put itself * @param identifier the identifier of the outer CHM node * @param timestamp the timestamp when the entry was put (this could be later if it was a * remote put) this method is called only from usual put or acquire diff --git a/src/main/java/net/openhft/chronicle/map/Replicator.java b/src/main/java/net/openhft/chronicle/map/Replicator.java index 0911c8ba0..86ef2ed82 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicator.java +++ b/src/main/java/net/openhft/chronicle/map/Replicator.java @@ -21,7 +21,7 @@ import java.io.IOException; /** * @see Replicators - * @see ChronicleMapBuilder#addReplicator(Replicator) + * @see ChronicleMapBuilder#replicator(byte, Replicator) */ public abstract class Replicator { @@ -31,14 +31,6 @@ public abstract class Replicator { protected Replicator() { } - /** - * Returns this node identifier within the replication topology. - * - * @return the unique identifier among nodes - * @see Replica#identifier() - */ - public abstract byte identifier(); - /** * Applies the replicator to the map instance and returns a Closeable token to manage resources, * associated with the replication. diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java index f8b44bf52..faa766e85 100644 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ b/src/main/java/net/openhft/chronicle/map/Replicators.java @@ -25,13 +25,8 @@ public final class Replicators { private Replicators() { } - public static Replicator tcp(final byte identifier, - final TcpReplicationConfig replicationConfig) { + public static Replicator tcp(final TcpReplicationConfig replicationConfig) { return new Replicator() { - @Override - public byte identifier() { - return identifier; - } @Override protected Closeable applyTo(ChronicleMapBuilder builder, @@ -43,13 +38,9 @@ public final class Replicators { }; } - public static Replicator udp(final byte identifier, - final UdpReplicationConfig replicationConfig) { + public static Replicator udp( + final UdpReplicationConfig replicationConfig) { return new Replicator() { - @Override - public byte identifier() { - return identifier; - } @Override protected Closeable applyTo(ChronicleMapBuilder builder, diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java index b213f4455..5ea8cb1c1 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java @@ -29,7 +29,7 @@ import static java.util.Collections.unmodifiableSet; import static java.util.concurrent.TimeUnit.SECONDS; -public class TcpReplicationConfig { +public class TcpReplicationConfig implements Config { private static final int DEFAULT_PACKET_SIZE = 1024 * 8; private static final long DEFAULT_HEART_BEAT_INTERVAL = 20; diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java index 0e02d9e9d..a1e8ad0aa 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java @@ -23,7 +23,7 @@ import java.net.InetAddress; import java.net.NetworkInterface; -public abstract class UdpReplicationConfig { +public abstract class UdpReplicationConfig implements Config { /** * Package-private constructor forbids subclassing from outside of the package @@ -37,7 +37,6 @@ public abstract class UdpReplicationConfig { throw new IllegalArgumentException(); - return create(address, port, null, ThrottlingConfig.noThrottling()); } diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index 5473edcb5..d4d2b47c1 100644 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -39,7 +39,8 @@ public class ChronicleSetBuilder ChronicleSetBuilder(Class keyClass) { chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, DummyValue.class) .entryAndValueAlignment(Alignment.NO_ALIGNMENT) - .valueMarshallerAndFactory(DummyValueMarshaller.INSTANCE, NullObjectFactory.of()); + .valueMarshallerAndFactory(DummyValueMarshaller.INSTANCE, + NullObjectFactory.of()); } public static ChronicleSetBuilder of(Class keyClass) { @@ -85,7 +86,8 @@ public class ChronicleSetBuilder * .entries(50000) * .keySize(10) * .create();} - * (Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded + * + *

(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}.) * @@ -121,9 +123,9 @@ public class ChronicleSetBuilder * strongly recommended always to configure {@linkplain #keySize(int) key size}, * if they couldn't be derived statically. * - *

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. + *

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. */ @Override public ChronicleSetBuilder entrySize(int entrySize) { @@ -161,6 +163,12 @@ public class ChronicleSetBuilder return this; } + @Override + public ChronicleSetBuilder replicators(byte identifier, Replicator... replicators) { + chronicleMapBuilder.replicators(identifier, replicators); + return this; + } + @Override public String toString() { return " ChronicleSetBuilder{" + @@ -181,15 +189,11 @@ public class ChronicleSetBuilder return toString().hashCode(); } - public ChronicleSetBuilder addReplicator(Replicator replicator) { - chronicleMapBuilder.addReplicator(replicator); + public ChronicleSetBuilder replicators(byte identifier, Replicator replicator) { + chronicleMapBuilder.replicators(identifier, replicator); return this; } - public ChronicleSetBuilder setReplicators(Replicator... replicators) { - chronicleMapBuilder.setReplicators(replicators); - return this; - } @Override public ChronicleSetBuilder timeProvider(TimeProvider timeProvider) { diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java index bf60a5ac2..1799e16c3 100644 --- a/src/test/java/net/openhft/chronicle/map/Builder.java +++ b/src/test/java/net/openhft/chronicle/map/Builder.java @@ -73,7 +73,7 @@ public class Builder { .heartBeatInterval(1L, SECONDS); return ChronicleMapBuilder.of(kClass, vClass) .entries(20000L) - .addReplicator(tcp(identifier, tcpConfig)); + .replicators(identifier, tcp( tcpConfig)); } diff --git a/src/test/java/net/openhft/chronicle/map/ClusterReplicationTest.java b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java similarity index 81% rename from src/test/java/net/openhft/chronicle/map/ClusterReplicationTest.java rename to src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java index 935661045..086dffb86 100644 --- a/src/test/java/net/openhft/chronicle/map/ClusterReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java @@ -17,7 +17,10 @@ package net.openhft.chronicle.map; -import org.junit.*; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; import java.io.Closeable; import java.io.IOException; @@ -32,7 +35,7 @@ import static org.junit.Assert.assertTrue; * * @author Rob Austin. */ -public class ClusterReplicationTest { +public class ChannelReplicationTest { private ChronicleMap map1a; private ChronicleMap map2a; @@ -40,8 +43,8 @@ public class ClusterReplicationTest { private ChronicleMap map1b; private ChronicleMap map2b; - private ReplicatingCluster clusterB; - private ReplicatingCluster clusterA; + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; @Before @@ -52,12 +55,13 @@ public class ClusterReplicationTest { .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); - clusterA = new ReplicatingClusterBuilder((byte) 1, 1024).tcpReplication(tcpConfig) + channelProviderA = new ChannelProviderBuilder() + .replicators((byte) 1, tcpConfig) .create(); map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterA.channelReplicator((short) 1)) + .channel(channelProviderA.createChannel((short) 1)) .create(getPersistenceFile()); } @@ -65,12 +69,12 @@ public class ClusterReplicationTest { TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(8087).heartBeatInterval(1, SECONDS); - clusterB = new ReplicatingClusterBuilder((byte) 2, 1024).tcpReplication(tcpConfig) + channelProviderB = new ChannelProviderBuilder().replicators((byte) 2, tcpConfig) .create(); map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterB.channelReplicator((short) 1)) + .channel(channelProviderB.createChannel((short) 1)) .create(getPersistenceFile()); } } @@ -78,7 +82,7 @@ public class ClusterReplicationTest { @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{clusterA, clusterB}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) { @@ -93,13 +97,13 @@ public class ClusterReplicationTest { map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterB.channelReplicator((short) 2)) + .channel(channelProviderB.createChannel((short) 2)) .create(getPersistenceFile()); map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterA.channelReplicator((short) 2)) + .channel(channelProviderA.createChannel((short) 2)) .create(getPersistenceFile()); map2a.put(1, "EXAMPLE-2"); diff --git a/src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java similarity index 79% rename from src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java rename to src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java index 09a491cb2..f5292974c 100644 --- a/src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java +++ b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java @@ -17,7 +17,10 @@ package net.openhft.chronicle.map; -import org.junit.*; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; import java.io.Closeable; import java.io.File; @@ -32,7 +35,7 @@ import static org.junit.Assert.assertTrue; * * @author Rob Austin. */ -public class PostClusterMapCreationTest { +public class PostChannelCreationTest { private ChronicleMap map1a; private ChronicleMap map2a; @@ -40,11 +43,8 @@ public class PostClusterMapCreationTest { private ChronicleMap map1b; private ChronicleMap map2b; - private ReplicatingCluster clusterB; - private ReplicatingCluster clusterA; - - private ReplicatingClusterBuilder replicatingClusterBuilder1; - + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; public static File getPersistenceFile() { String TMP = System.getProperty("java.io.tmpdir"); @@ -59,43 +59,43 @@ public class PostClusterMapCreationTest { public void setup() throws IOException { { - final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig + final TcpReplicationConfig tcpConfig = TcpReplicationConfig .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); - clusterA = new ReplicatingClusterBuilder((byte) 1, 1024) - .tcpReplication(tcpReplicationConfig).create(); + channelProviderA = new ChannelProviderBuilder() + .replicators((byte) 1, tcpConfig).create(); // this is how you add maps after the custer is created map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterA.channelReplicator((short) 1)) + .channel(channelProviderA.createChannel((short) 1)) .create(getPersistenceFile()); map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterA.channelReplicator((short) 2)) + .channel(channelProviderA.createChannel((short) 2)) .create(getPersistenceFile()); } { - final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig + final TcpReplicationConfig tcpConfig = TcpReplicationConfig .of(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); - clusterB = new ReplicatingClusterBuilder((byte) 2, 1024) - .tcpReplication(tcpReplicationConfig).create(); + channelProviderB = new ChannelProviderBuilder() + .replicators((byte) 2, tcpConfig).create(); // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterB.channelReplicator((short) 1)) + .channel(channelProviderB.createChannel((short) 1)) .create(getPersistenceFile()); map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterB.channelReplicator((short) 2)) + .channel(channelProviderB.createChannel((short) 2)) .create(getPersistenceFile()); } @@ -105,7 +105,7 @@ public class PostClusterMapCreationTest { @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{clusterA, clusterB}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) { diff --git a/src/test/java/net/openhft/chronicle/map/ReplicatedChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/ReplicatedChronicleMapTest.java index 404bb4ef1..39738c5d3 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicatedChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicatedChronicleMapTest.java @@ -18,8 +18,6 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.map.jrs166.JSR166TestCase; import org.junit.Test; -import org.junit.runner.RunWith; -import org.junit.runners.Parameterized; import java.io.File; import java.io.IOException; @@ -44,48 +42,51 @@ public class ReplicatedChronicleMapTest extends JSR166TestCase { } ChronicleMap newShmIntString(int size) throws IOException { + ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(size).replicators((byte) 1); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(size) - .forceReplicatedImpl() .create(getPersistenceFile()); } ChronicleMap newShmListBoolean(int size) throws IOException { + ChronicleMapBuilder.of(ArrayList.class, CharSequence.class) + .entries(size).replicators((byte) 1); return ChronicleMapBuilder.of(ArrayList.class, CharSequence.class) .entries(size) - .forceReplicatedImpl() .create(getPersistenceFile()); } ChronicleMap newShmListBoolean() throws IOException { + ChronicleMapBuilder.of(ArrayList.class, CharSequence.class).replicators((byte) 1); return ChronicleMapBuilder.of(ArrayList.class, CharSequence.class) - .forceReplicatedImpl() .create(getPersistenceFile()); } ChronicleMap newShmStringString(int size) throws IOException { + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(size).replicators((byte) 1); return ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .entries(size) - .forceReplicatedImpl() .create(getPersistenceFile()); } ChronicleMap newShmIntString() throws IOException { + ChronicleMapBuilder.of(Integer.class, CharSequence.class).replicators((byte) 1); return ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .forceReplicatedImpl() .create(getPersistenceFile()); } ChronicleMap newShmBiBoolean() throws IOException { - return ChronicleMapBuilder.of(BI.class, Boolean.class) - .forceReplicatedImpl().create(getPersistenceFile()); + ChronicleMapBuilder.of(BI.class, Boolean.class).replicators((byte) 1); + return ChronicleMapBuilder.of(BI.class, Boolean.class).create(getPersistenceFile()); } diff --git a/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java b/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java index 21d1ffa8d..82ede45f6 100644 --- a/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java +++ b/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java @@ -41,9 +41,8 @@ public class SingleMapClusterTest { private ChronicleMap map1b; - private ReplicatingCluster clusterB; - private ReplicatingCluster clusterA; - + private ChannelProvider channelProviderA; + private ChannelProvider channelProviderB; public static File getPersistenceFile() { String TMP = System.getProperty("java.io.tmpdir"); @@ -61,12 +60,12 @@ public class SingleMapClusterTest { .of(8086, new InetSocketAddress("localhost", 8087)) .heartBeatInterval(1, SECONDS); - clusterA = new ReplicatingClusterBuilder((byte) 1, 1024) - .tcpReplication(tcpReplicationConfig).create(); + channelProviderA = new ChannelProviderBuilder() + .replicators((byte) 1, tcpReplicationConfig).create(); // this is how you add maps after the custer is created map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(clusterA.channelReplicator((short) 1)) + .channel(channelProviderA.createChannel((short) 1)) .create(getPersistenceFile()); } @@ -76,11 +75,12 @@ public class SingleMapClusterTest { .of(8087, new InetSocketAddress("localhost", 8086)) .heartBeatInterval(1, SECONDS); - clusterB = new ReplicatingClusterBuilder((byte) 2, 1024) - .tcpReplication(tcpReplicationConfig).create(); + channelProviderB = new ChannelProviderBuilder() + .replicators((byte) 2, tcpReplicationConfig).create(); + // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .addReplicator(clusterB.channelReplicator((short) 1)) + .channel(channelProviderB.createChannel((short) 1)) .entries(1000) .create(getPersistenceFile()); } @@ -89,7 +89,7 @@ public class SingleMapClusterTest { @After public void tearDown() throws InterruptedException { - for (final Closeable closeable : new Closeable[]{clusterA, clusterB}) { + for (final Closeable closeable : new Closeable[]{channelProviderA, channelProviderB}) { try { closeable.close(); } catch (IOException e) { diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java index 922fd26f2..d4bd0a076 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java @@ -49,7 +49,7 @@ public class TCPSocketReplicationTest3wayPutReturnsNull { .entries(1000) .putReturnsNull(true) - .addReplicator(Replicators.tcp(identifier, tcpConfig)) + .replicators(identifier, Replicators.tcp(tcpConfig)) .create(Builder.getPersistenceFile()); } diff --git a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTests.java b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTests.java index d5ffc4545..51d1b54f8 100644 --- a/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTests.java +++ b/src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTests.java @@ -48,9 +48,10 @@ public class TimeBasedReplicationTests extends JSR166TestCase { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .forceReplicatedImpl() + .replicators((byte) 1) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -78,10 +79,10 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALatePutIfAbsent() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .forceReplicatedImpl() - .entries(10) + .replicators((byte) 1) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -110,10 +111,10 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALateReplace() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .forceReplicatedImpl() - .entries(10) + .replicators((byte) 1) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -145,10 +146,10 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALateReplaceWithValue() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - .entries(10) - .forceReplicatedImpl() - .entries(10) + .replicators((byte) 1) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -177,9 +178,9 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALateRemoveWithValue() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - - .forceReplicatedImpl() + .replicators((byte) 1) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -207,9 +208,9 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreALateRemove() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - - .forceReplicatedImpl() + .replicators((byte) 1) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -237,10 +238,9 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testIgnoreWithRemoteRemove() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class).replicators((byte) 1); ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - - .forceReplicatedImpl() - .entries(10) + .replicators((byte) 1) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -269,10 +269,9 @@ public class TimeBasedReplicationTests extends JSR166TestCase { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class).replicators((byte) 1); ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) - - .forceReplicatedImpl() - .entries(10) + .replicators((byte) 1) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -303,7 +302,7 @@ public class TimeBasedReplicationTests extends JSR166TestCase { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); ReplicatedChronicleMap map = (ReplicatedChronicleMap) ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .entries(10) - .forceReplicatedImpl() + .replicators((byte) 1) .entries(10) .timeProvider(timeProvider) .create(getPersistenceFile()); @@ -331,9 +330,10 @@ public class TimeBasedReplicationTests extends JSR166TestCase { public void testPutRemovePut() throws IOException { final TimeProvider timeProvider = Mockito.mock(TimeProvider.class); + ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) + .entries(10).replicators((byte) 1); ChronicleMap map = ChronicleMapBuilder.of(CharSequence.class, CharSequence.class) .entries(10) - .forceReplicatedImpl() .entries(10) .timeProvider(timeProvider) .create(getPersistenceFile()); diff --git a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java index 91671206d..59199827d 100644 --- a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java +++ b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java @@ -51,12 +51,12 @@ public class TwoMapOnDifferentServers { File persistenceFile = getPersistenceFile(); map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000) - .addReplicator(tcp((byte) 1, tcpReplicationConfig)) + .replicators((byte) 1, tcp(tcpReplicationConfig)) .create(persistenceFile); map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(20000) - .addReplicator(tcp((byte) 2, + .replicators((byte) 2, tcp( TcpReplicationConfig.of(8077).heartBeatInterval(1, SECONDS))) .create(persistenceFile); } diff --git a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java index f95cf91e4..8bec68d2c 100644 --- a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java @@ -44,7 +44,7 @@ public class UDPSocketReplicationTest { return ChronicleMapBuilder.of(Integer.class, CharSequence.class) .entries(1000) - .addReplicator(udp((byte) identifier, udpConfig)) + .replicators((byte) identifier, udp(udpConfig)) .create(Builder.getPersistenceFile()); } diff --git a/src/test/java/net/openhft/chronicle/map/fromdocs/GettingStartedTest.java b/src/test/java/net/openhft/chronicle/map/fromdocs/GettingStartedTest.java deleted file mode 100644 index 8e338784a..000000000 --- a/src/test/java/net/openhft/chronicle/map/fromdocs/GettingStartedTest.java +++ /dev/null @@ -1,65 +0,0 @@ -/* - * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package net.openhft.chronicle.map.fromdocs; - -import net.openhft.chronicle.map.ChronicleMapBuilder; -import org.junit.Test; - -import java.io.Closeable; -import java.io.File; -import java.io.IOException; -import java.util.Map; - -import static org.junit.Assert.assertEquals; - -/** - * Created by peter on 28/02/14. - */ -public class GettingStartedTest { - - @Test - public void testTheCodeInGuide() throws IOException { - String tmpdir = System.getProperty("java.io.tmpdir"); - Map map = ChronicleMapBuilder.of(String.class, String.class) - - .create(new File(tmpdir + "/shared.map")); - map.put("some.json", "missing"); - map.get("some.json"); - - map.put("Hello", "World"); - String hi = map.get("Hello"); - map.put("Long String", "0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789"); - String json = "{\"menu\": {\n" + - " \"id\": \"file\",\n" + - " \"value\": \"File\",\n" + - " \"popup\": {\n" + - " \"menuitem\": [\n" + - " {\"value\": \"New\", \"onclick\": \"CreateNew()\"},\n" + - " {\"value\": \"Open\", \"onclick\": \"OpenDoc()\"},\n" + - " {\"value\": \"Close\", \"onclick\": \"CloseDoc()\"}\n" + - " ]\n" + - " }\n" + - "}}"; - map.put("some.json", json); - - String s2 = map.get("some.json"); - assertEquals(json, s2); - - ((Closeable) map).close(); - - } -} diff --git a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java index c62a03e5c..0212db7f6 100644 --- a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java +++ b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java @@ -51,7 +51,7 @@ public class SetForTCPSocketReplicationTest3way { final int serverPort, final InetSocketAddress... endpoints) throws IOException { return (T) ChronicleSetBuilder.of(Integer.class) - .addReplicator(tcp(identifier, of(serverPort, endpoints). + .replicators(identifier, tcp(of(serverPort, endpoints). heartBeatInterval(1L, SECONDS))).create(getPersistenceFile()); }