diff --git a/src/main/java/net/openhft/chronicle/ChronicleHash.java b/src/main/java/net/openhft/chronicle/ChronicleHash.java
new file mode 100644
index 000000000..996213937
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/ChronicleHash.java
@@ -0,0 +1,64 @@
+/*
+ * 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;
+
+import net.openhft.chronicle.map.ChronicleMap;
+import net.openhft.chronicle.set.ChronicleSet;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+
+/**
+ * This interface defines common {@link ChronicleMap} and {@link ChronicleSet}, related to off-heap
+ * memory management and file-mapping. Not usable by itself.
+ */
+public interface ChronicleHash extends Closeable {
+ /**
+ * Returns the file this hash container mapped to, i. e. when it is created by
+ * {@link ChronicleHashBuilder#create()} call, or {@code null} if it is purely in-memory,
+ * i. e. if it is created by {@link ChronicleHashBuilder#create()} call.
+ *
+ * @return the file this {@link ChronicleMap} or {@link ChronicleSet} is mapped to,
+ * or {@code null} if it is not mapped to any file
+ * @see ChronicleHashBuilder#create(File)
+ * @see ChronicleHashBuilder#create()
+ */
+ File file();
+
+ /**
+ * Releases the off-heap memory, used by this hash container and resources, used by replication,
+ * if any. However, if hash container (hence off-heap memory, used by it) is mapped to the file
+ * and there are other instances mapping the same data on the server across JVMs, the memory
+ * won't be actually freed on operation system level. I. e. this method call doesn't affect
+ * other {@link ChronicleMap} or {@link ChronicleSet} instances mapping the same data.
+ *
+ * If you won't call this method, memory would be held at least until next garbage
+ * collection. This could be a problem if, for example, you target rare garbage collections,
+ * but load and drop {@code ChronicleHash}es regularly.
+ *
+ *
+ * TODO what about commit guarantees, when ChronicleMap is used with memory-mapped files, if
+ * {@code ChronicleMap}/{@code ChronicleSet} closed/not?
+ *
+ *
After this method call behaviour of all methods of {@code ChronicleMap}
+ * or {@code ChronicleSet} is undefined. Any method call on the map might throw
+ * any exception, or even JVM crash. Shortly speaking, don't call use map closing.
+ */
+ @Override
+ void close() throws IOException;
+}
diff --git a/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java
index 20cd595b5..e4a21799a 100644
--- a/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java
+++ b/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java
@@ -1,11 +1,16 @@
package net.openhft.chronicle;
-import net.openhft.chronicle.map.*;
+import net.openhft.chronicle.map.ChannelProvider;
+import net.openhft.chronicle.map.ChronicleMap;
+import net.openhft.chronicle.map.ChronicleMapBuilder;
+import net.openhft.chronicle.map.ReplicationConfig;
+import net.openhft.chronicle.set.ChronicleSet;
import net.openhft.chronicle.set.ChronicleSetBuilder;
-import org.jetbrains.annotations.NotNull;
-
-import net.openhft.lang.io.serialization.*;
+import net.openhft.lang.io.serialization.BytesMarshaller;
+import net.openhft.lang.io.serialization.BytesMarshallerFactory;
+import net.openhft.lang.io.serialization.ObjectSerializer;
import net.openhft.lang.model.Byteable;
+import org.jetbrains.annotations.NotNull;
import java.io.File;
import java.io.IOException;
@@ -17,17 +22,15 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
- * This interface defines the meaning of configurations, common to {@link ChronicleMapBuilder}
- * and {@link ChronicleSetBuilder}, i. e. Chronicle hash container configurations.
+ * This interface defines the meaning of configurations, common to {@link ChronicleMapBuilder} and {@link
+ * ChronicleSetBuilder}, i. e. Chronicle hash container configurations.
*
* @param the type of keys in hash containers, created by this builder
- * @param the container type, created by this builder, i. e. {@link ChronicleMap}
- * or {@link ChronicleSet}
- * @param the concrete builder type, i. e. {@link ChronicleMapBuilder}
- * or {@link ChronicleSetBuilder}
+ * @param the container type, created by this builder, i. e. {@link ChronicleMap} or {@link ChronicleSet}
+ * @param the concrete builder type, i. e. {@link ChronicleMapBuilder} or {@link ChronicleSetBuilder}
*/
-public interface ChronicleHashBuilder>
- extends Cloneable {
+public interface ChronicleHashBuilder> extends Cloneable {
B clone();
@@ -40,14 +43,13 @@ public interface ChronicleHashBuilderIf key size varies moderately, specify the size higher than average, but lower than
- * the maximum possible, to minimize average memory overuse. If key size varies in a wide range,
- * it's better to use {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it
- * directly.
+ * If key size varies moderately, specify the size higher than average, but lower than the maximum
+ * possible, to minimize average memory overuse. If key size varies in a wide range, it's better to use
+ * {@linkplain #entrySize(int) entry size} in "chunk" mode and configure it directly.
*
* @param keySize number of bytes, taken by serialized form of keys
* @return this builder back
@@ -57,15 +59,15 @@ public interface ChronicleHashBuilderIf keys are of boxed primitive type or {@link Byteable} subclass, i. e. if key size
- * is known statically, it is automatically accounted and this method shouldn't be called.
+ * If keys are of boxed primitive type or {@link Byteable} subclass, i. e. if key size is known
+ * statically, it is automatically accounted and this method shouldn't be called.
*
- *
If key size varies, method {@link #keySize(int)} or {@link #entrySize(int)} should be
- * called instead of this one.
+ *
If key size varies, method {@link #keySize(int)} or {@link #entrySize(int)} should be called instead
+ * of this one.
*
* @param sampleKey the sample key
* @return this builder back
@@ -74,55 +76,43 @@ public interface ChronicleHashBuilder{@link ChronicleMap} and {@link ChronicleSet} store their data off-heap, so it is required
- * to serialize key (and values, in {@code ChronicleMap} case) (unless they are direct
- * {@link Byteable} instances). Serialized key bytes (+ serialized value bytes,
- * in {@code ChronicleMap} case) + some metadata bytes comprise "entry space", which
- * {@code ChronicleMap} or {@code ChronicleSet} should allocate. So entry size
- * is a minimum allocation portion in the hash containers, created by this builder.
- * E. g. if entry size is 100, the created container could only allocate 100, 200, 300... bytes
- * for an entry. If say 150 bytes of entry space are required by the entry, 200 bytes
- * will be allocated, 150 used and 50 wasted. To minimize memory overuse and improve speed, you
- * should pay decent attention to this configuration.
+ * {@link ChronicleMap} and {@link ChronicleSet} store their data off-heap, so it is required to
+ * serialize key (and values, in {@code ChronicleMap} case) (unless they are direct {@link Byteable}
+ * instances). Serialized key bytes (+ serialized value bytes, in {@code ChronicleMap} case) + some
+ * metadata bytes comprise "entry space", which {@code ChronicleMap} or {@code ChronicleSet} should
+ * allocate. So entry size is a minimum allocation portion in the hash containers, created by this
+ * builder. E. g. if entry size is 100, the created container could only allocate 100, 200, 300... bytes
+ * for an entry. If say 150 bytes of entry space are required by the entry, 200 bytes will be allocated,
+ * 150 used and 50 wasted. To minimize memory overuse and improve speed, you should pay decent attention
+ * to this configuration.
*
- *
There are three major patterns of this configuration usage:
- *
- * - Key (and value, in {@code ChronicleMap} case) sizes are constant. Configure them
- * via {@link #constantKeySizeBySample(Object)} and
- * {@link ChronicleMapBuilder#constantValueSizeBySample(Object)} methods, and you will
- * experience no memory waste at all.
- * - Key (and/or value size, in {@code ChronicleMap} case) varies moderately. Specify them
- * using corresponding methods, or specify entry size directly by calling this method,
- * by sizes somewhere between average and maximum possible. The idea is to have most
- * (90% or more) entries to fit a single "entry size" with moderate memory waste
- * (10-20% on average), rest 10% or less of entries should take 2 "entry sizes", thus with
- * ~50% memory overuse.
- * - Key (and/or value size, in {@code ChronicleMap} case) varies in a wide range.
- * Then it's best to use entry size configuration in chunk mode. Specify entry size
- * so that most entries should take from 5 to several dozens of "chunks". With this
- * approach, average memory waste should be very low.
+ *
There are three major patterns of this configuration usage:
- Key (and value, in {@code
+ * ChronicleMap} case) sizes are constant. Configure them via {@link #constantKeySizeBySample(Object)} and
+ * {@link ChronicleMapBuilder#constantValueSizeBySample(Object)} methods, and you will experience no
+ * memory waste at all.
- Key (and/or value size, in {@code ChronicleMap} case) varies moderately.
+ * Specify them using corresponding methods, or specify entry size directly by calling this method, by
+ * sizes somewhere between average and maximum possible. The idea is to have most (90% or more) entries to
+ * fit a single "entry size" with moderate memory waste (10-20% on average), rest 10% or less of entries
+ * should take 2 "entry sizes", thus with ~50% memory overuse.
- Key (and/or value size, in {@code
+ * ChronicleMap} case) varies in a wide range. Then it's best to use entry size configuration in chunk
+ * mode. Specify entry size so that most entries should take from 5 to several dozens of "chunks".
+ * With this approach, average memory waste should be very low.
*
- *
However, remember that
- *
- * - Operations with entries that span several "entry sizes" are a bit slower, than
- * entries which take a single "entry size". That is why "chunk" approach
- * is not recommended, when key (and/or value size) varies moderately.
- * - The maximum number of chunks could be taken by an entry is 64.
- * {@link IllegalArgumentException} is thrown on attempt to insert too large entry,
- * compared to the configured or computed entry size.
- * - The number of "entries" {@linkplain #entries(long) in the whole map} and
- * {@linkplain #actualEntriesPerSegment(int) per segment} is actually the number
- * "chunks".
- *
+ * However, remember that
- Operations with entries that span several "entry sizes" are a bit
+ * slower, than entries which take a single "entry size". That is why "chunk" approach is not recommended,
+ * when key (and/or value size) varies moderately.
- The maximum number of chunks could be taken by
+ * an entry is 64. {@link IllegalArgumentException} is thrown on attempt to insert too large entry,
+ * compared to the configured or computed entry size.
- The number of "entries" {@linkplain
+ * #entries(long) in the whole map} and {@linkplain #actualEntriesPerSegment(int) per segment} is actually
+ * the number "chunks".
*
- * Example: if values in your {@code ChronicleMap} are adjacency lists of some social
- * graph, where nodes are represented as {@code long} ids, and adjacency lists
- * are serialized in efficient manner, for example as {@code long[]} arrays. Typical number
- * of connections is 100-300, maximum is 3000. In this case entry size of
- * 50 * (8 bytes for each id) = 400 bytes would be a good choice:
{@code
+ * Example: if values in your {@code ChronicleMap} are adjacency lists of some social graph, where
+ * nodes are represented as {@code long} ids, and adjacency lists are serialized in efficient manner, for
+ * example as {@code long[]} arrays. Typical number of connections is 100-300, maximum is 3000. In this
+ * case entry size of
+ * 50 * (8 bytes for each id) = 400 bytes would be a good choice:
{@code
* Map socialGraph = ChronicleMapBuilder
* .of(Long.class, long[].class)
* // given that graph should have of 1 billion nodes, and 150 average adjacency list size
@@ -130,11 +120,8 @@ public interface ChronicleHashBuilder
- * It is minimum possible (because 3000 friends / 50 friends = 60 is close to 64 "max
- * chunks by single entry" limit, and ensures moderate average memory overuse (not more
- * than 20%).
- *
- *
+ * It is minimum possible (because 3000 friends / 50 friends = 60 is close to 64 "max chunks by single
+ * entry" limit, and ensures moderate average memory overuse (not more than 20%).
*
* @param entrySize the "chunk size" in bytes
* @return this builder back
@@ -143,35 +130,28 @@ public interface ChronicleHashBuilder
- * If key size (and value size, in {@code ChronicleMap} case) is constant, this number
- * is equal to the maximum number of entries (because each entry takes exactly one
- * "entry size" memory unit).
- * If key (and/or value, in {@code ChronicleMap} case) size varies moderately, you
- * should pass to this method the maximum number of entries + 5-25%, depending on your data
- * properties and configured {@linkplain #keySize(int) key}/{@linkplain
- * ChronicleMapBuilder#valueSize(int) value}/{@linkplain #entrySize(int) entry} sizes.
- * If your data size varies in a wide range, pass the maximum number of entries
- * multiplied by average data size and divided by the configured "entry size" (i. e. chunk
- * size). See an example in the documentation to {@link #entrySize(int)} method.
- *
+ * - If key size (and value size, in {@code ChronicleMap} case) is constant, this number is equal
+ * to the maximum number of entries (because each entry takes exactly one "entry size" memory unit).
+ * - If key (and/or value, in {@code ChronicleMap} case) size varies moderately, you should pass to this
+ * method the maximum number of entries + 5-25%, depending on your data properties and configured
+ * {@linkplain #keySize(int) key}/{@linkplain ChronicleMapBuilder#valueSize(int) value}/{@linkplain
+ * #entrySize(int) entry} sizes.
- If your data size varies in a wide range, pass the maximum
+ * number of entries multiplied by average data size and divided by the configured "entry size" (i. e.
+ * chunk size). See an example in the documentation to {@link #entrySize(int)} method.
*
- * You shouldn't put additional margin over the number, computed according the rules above.
- * This bad practice was popularized by {@link HashMap#HashMap(int)}
- * and {@link HashSet#HashSet(int)} constructors, which accept "capacity", that should
- * be multiplied by "load factor" to obtain actual maximum expected number of entries.
- * {@code ChronicleMap} and {@code ChronicleSet} don't have a notion of load factor.
+ *
You shouldn't put additional margin over the number, computed according the rules above. This bad
+ * practice was popularized by {@link HashMap#HashMap(int)} and {@link HashSet#HashSet(int)} constructors,
+ * which accept "capacity", that should be multiplied by "load factor" to obtain actual maximum expected
+ * number of entries. {@code ChronicleMap} and {@code ChronicleSet} don't have a notion of load factor.
*
*
Default value is 2^20 (~ 1 million).
*
- * @param entries maximum size of the created maps, in memory allocation units,
- * so-called "entry size"
+ * @param entries maximum size of the created maps, in memory allocation units, so-called "entry size"
* @return this builder back
* @see #entrySize(int)
*/
@@ -189,7 +169,9 @@ public interface ChronicleHashBuilder keyMarshaller);
/**
- * Specifies that key objects, queried with the hash containers, created by this builder,
- * are inherently immutable. Keys in {@link ChronicleMap} or {@link ChronicleSet}
- * are not required to be immutable, as in ordinary {@link Map} or {@link Set} implementations,
- * because they are serialized off-heap. However, {@code ChronicleMap} and {@code ChronicleSet}
- * implementations can benefit from the knowledge that keys are not mutated between queries.
+ * Specifies that key objects, queried with the hash containers, created by this builder, are inherently
+ * immutable. Keys in {@link ChronicleMap} or {@link ChronicleSet} are not required to be immutable, as in
+ * ordinary {@link Map} or {@link Set} implementations, because they are serialized off-heap. However,
+ * {@code ChronicleMap} and {@code ChronicleSet} implementations can benefit from the knowledge that keys
+ * are not mutated between queries.
*
- * By default, {@code ChronicleHashBuilder}s detects immutability automatically
- * only for very few standard JDK types (for example, for {@link String}), it is not recommended
- * to rely on {@code ChronicleHashBuilder} to be smart enough about this.
+ *
By default, {@code ChronicleHashBuilder}s detects immutability automatically only for very few
+ * standard JDK types (for example, for {@link String}), it is not recommended to rely on {@code
+ * ChronicleHashBuilder} to be smart enough about this.
*
* @return this builder back
*/
diff --git a/src/main/java/net/openhft/chronicle/TimeProvider.java b/src/main/java/net/openhft/chronicle/TimeProvider.java
index 5f48dcb51..8f5885286 100644
--- a/src/main/java/net/openhft/chronicle/TimeProvider.java
+++ b/src/main/java/net/openhft/chronicle/TimeProvider.java
@@ -17,7 +17,7 @@
package net.openhft.chronicle;
import net.openhft.chronicle.map.ChronicleMap;
-import net.openhft.chronicle.map.ChronicleSet;
+import net.openhft.chronicle.set.ChronicleSet;
import java.io.ObjectStreamException;
import java.io.Serializable;
diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
index 89e741a0b..1f9cfba3e 100644
--- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
@@ -197,8 +197,9 @@ abstract class AbstractChannelReplicator implements Closeable {
/**
- * checks the number of bytes written in this interval, if this number of bytes exceeds a threshold,
- * the selected will de-register the socket that is being written to, until the interval is finished.
+ * checks the number of bytes written in this interval, if this number of bytes exceeds a
+ * threshold, the selected will de-register the socket that is being written to, until the
+ * interval is finished.
*
* @throws ClosedChannelException
*/
@@ -261,12 +262,22 @@ abstract class AbstractChannelReplicator implements Closeable {
@Override
public boolean onEntry(final Bytes entry, final int chronicleId) {
+
+ long pos0 = in.position();
+
+ // used to denote that this is not a stateless map event
+ in.writeByte(StatelessMapClient.EventId.STATEFUL_UPDATE.ordinal());
+
+ long sizeLocation = in.position();
+
+ // this is where we will store the size of the entry
in.skip(SIZE_OF_SHORT);
+
final long start = in.position();
externalizable.writeExternalEntry(entry, in, chronicleId);
if (in.position() == start) {
- in.position(in.position() - SIZE_OF_SHORT);
+ in.position(pos0);
return false;
}
@@ -276,7 +287,10 @@ abstract class AbstractChannelReplicator implements Closeable {
if (entrySize > MAX_UNSIGNED_SHORT)
throw new IllegalStateException("entry too large, the entry size=" + entrySize + ", " +
"entries are limited to a size of " + MAX_UNSIGNED_SHORT);
- in.writeUnsignedShort(start - SIZE_OF_SHORT, entrySize);
+
+ if (LOG.isDebugEnabled())
+ LOG.debug("sending entry of entrySize=" + entrySize);
+ in.writeUnsignedShort(sizeLocation, entrySize);
return true;
}
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatingChannel.java b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
similarity index 98%
rename from src/main/java/net/openhft/chronicle/map/ReplicatingChannel.java
rename to src/main/java/net/openhft/chronicle/map/ChannelProvider.java
index 13ae91e3b..a59519c97 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatingChannel.java
+++ b/src/main/java/net/openhft/chronicle/map/ChannelProvider.java
@@ -45,8 +45,8 @@ import static net.openhft.chronicle.map.Replica.ModificationNotifier.NOP;
/**
* @author Rob Austin.
*/
-public final class ReplicatingChannel implements Closeable {
- private static final Logger LOG = LoggerFactory.getLogger(ReplicatingChannel.class.getName());
+public final class ChannelProvider implements Closeable {
+ private static final Logger LOG = LoggerFactory.getLogger(ChannelProvider.class.getName());
private static final byte BOOTSTRAP_MESSAGE = 'B';
final EntryExternalizable asEntryExternalizable = new EntryExternalizable() {
@@ -152,7 +152,7 @@ public final class ReplicatingChannel implements Closeable {
@Override
public void close() throws IOException {
- ReplicatingChannel.this.close();
+ ChannelProvider.this.close();
}
};
private final int maxEntrySize;
@@ -169,7 +169,7 @@ public final class ReplicatingChannel implements Closeable {
private final Set replicators =
new CopyOnWriteArraySet();
- ReplicatingChannel(ReplicatingChannelBuilder builder) {
+ ChannelProvider(ChannelProviderBuilder builder) {
localIdentifier = builder.identifier;
maxEntrySize = builder.maxEntrySize;
chronicleChannels = new Replica[builder.maxNumberOfChronicles];
@@ -407,7 +407,7 @@ public final class ReplicatingChannel implements Closeable {
@Override
protected Closeable applyTo(ChronicleMapBuilder builder,
- Replica map, EntryExternalizable entryExternalizable) {
+ Replica map, EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) {
if (builder.entrySize() > maxEntrySize) {
throw new IllegalArgumentException("During ReplicatingChannelBuilder setup, " +
"maxEntrySize=" + maxEntrySize + " was specified, but map with " +
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatingChannelBuilder.java b/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java
similarity index 64%
rename from src/main/java/net/openhft/chronicle/map/ReplicatingChannelBuilder.java
rename to src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java
index 23e095896..38726cbff 100644
--- a/src/main/java/net/openhft/chronicle/map/ReplicatingChannelBuilder.java
+++ b/src/main/java/net/openhft/chronicle/map/ChannelProviderBuilder.java
@@ -21,46 +21,54 @@ import java.io.IOException;
/**
* @author Rob Austin.
*/
-public final class ReplicatingChannelBuilder {
+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;
- ReplicatingChannelBuilder(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, ReplicationConfig... replicationConfigs) {
+ this.identifier = identifier;
+
+
+ for (ReplicationConfig replicationConfig : replicationConfigs) {
+
+ if (replicationConfig instanceof TcpReplicationConfig) {
+ this.tcpReplicationConfig = (TcpReplicationConfig) replicationConfig;
+ } else if (replicationConfig instanceof UdpReplicationConfig) {
+ this.udpReplicationConfig = (UdpReplicationConfig) replicationConfig;
+ } else
+ throw new UnsupportedOperationException();
}
- }
-
- public ReplicatingChannelBuilder udpReplication(UdpReplicationConfig replicationConfig) {
- this.udpReplicationConfig = replicationConfig;
return this;
}
- public ReplicatingChannelBuilder tcpReplication(TcpReplicationConfig replicationConfig) {
- this.tcpReplicationConfig = replicationConfig;
- return this;
- }
-
- public ReplicatingChannelBuilder maxNumberOfChronicles(int maxNumberOfChronicles) {
+ public ChannelProviderBuilder maxNumberOfChronicles(int maxNumberOfChronicles) {
this.maxNumberOfChronicles = maxNumberOfChronicles;
return this;
}
- public ReplicatingChannel create() throws IOException {
- final ReplicatingChannel replicatingCluster = new ReplicatingChannel(this);
+ public ChannelProvider create() throws IOException {
+ final ChannelProvider replicatingCluster = new ChannelProvider(this);
if (tcpReplicationConfig != null) {
final TcpReplicator tcpReplicator = new TcpReplicator(
replicatingCluster.asReplica,
replicatingCluster.asEntryExternalizable,
tcpReplicationConfig,
- maxEntrySize);
+ maxEntrySize, null);
replicatingCluster.add(tcpReplicator);
}
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
index e6858b4b8..57385aa98 100644
--- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java
@@ -16,24 +16,103 @@
package net.openhft.chronicle.map;
+import net.openhft.chronicle.ChronicleHash;
+import net.openhft.chronicle.map.serialization.BytesReader;
+import net.openhft.lang.io.Bytes;
+import net.openhft.lang.io.serialization.BytesMarshaller;
+import net.openhft.lang.io.serialization.ObjectFactory;
+import net.openhft.lang.model.Byteable;
+
import java.io.Closeable;
import java.io.File;
+import java.io.IOException;
+import java.util.Map;
import java.util.concurrent.ConcurrentMap;
-public interface ChronicleMap extends ConcurrentMap, Closeable {
+/**
+ * Extension of {@link ConcurrentMap} interface, stores the data off-heap.
+ *
+ * For information on
+ *
+ * - how to construct a {@code ChronicleMap}
+ * - {@code ChronicleMap} flavors and properties
+ * - available configurations
+ *
+ * see {@link ChronicleMapBuilder} documentation.
+ *
+ * Functionally this interface defines some methods supporting garbage-free off-heap programming:
+ * {@link #getUsing(Object, Object)}, {@link #acquireUsing(Object, Object)}.
+ *
+ *
Note that {@code ChronicleMap} extends {@link Closeable}, don't forget
+ * to {@linkplain #close() close} map when it is no longer needed.
+ *
+ * @param the map key type
+ * @param the map value type
+ */
+public interface ChronicleMap extends ConcurrentMap, ChronicleHash {
/**
- * @return size as a long.
- * @see java.util.Map size()
+ * Returns the number of entries in this map.
+ *
+ * @return number of entries in this map
+ * @see Map#size()
*/
long longSize();
/**
- * Get a value for a key if available. If the value is Byteable, it will be assigned to
- * reference the value, instead of copying the data for zero copy access to the collection.
+ * Returns the value to which the specified key is mapped, or {@code null} if this map contains
+ * no mapping for the key.
*
- * @param key to lookup.
- * @param value to reuse if possible. If null, a new object will be created.
- * @return value found or null if not.
+ * If the specified key if absent in the map, {@linkplain
+ * ChronicleMapBuilder#defaultValue(Object) default value} is taken or {@linkplain
+ * ChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) default value provider}
+ * is called (depending on what was configured last in {@code ChronicleMapBuilder}).
+ * If it results to something different from {@code null}, this object is put to this map for
+ * the specified key, and then returned from this {@code get()} call. Otherwise, if default
+ * value is {@code null} and default value provider returns {@code null}, the map remains
+ * unchanged and {@code null} is returned.
+ *
+ *
If the value class allows reusing, particularly if it is a {@link Byteable} subclass,
+ * consider {@link #getUsing(Object, Object)} method instead of this to reduce garbage creation.
+ *
+ * @param key the key whose associated value is to be returned
+ * @return the value to which the specified key is mapped after this method call,
+ * or {@code null} if no value is mapped
+ * @see #getUsing(Object, Object)
+ */
+ @Override
+ V get(Object key);
+
+ /**
+ * Returns the value to which the specified key is mapped, read to the provided {@code value}
+ * object, if possible, or returns {@code null}, if this map contains no mapping for the key.
+ *
+ *
If the specified key is present in the map, the value data is read to the provided
+ * {@code value} object via value marshaller's {@link BytesMarshaller#read(Bytes, Object)
+ * read(Bytes, value)} or value reader's {@link BytesReader#read(Bytes, long, Object)
+ * read(Bytes, size, value)} method, depending on what deserialization strategy is configured
+ * on this map builder. If the value deserializer is able to reuse the given {@code value}
+ * object, calling this method instead of {@link #get(Object)} could help to reduce garbage
+ * creation.
+ *
+ *
The provided {@code value} object is allowed to be {@code null}, in this case
+ * {@code map.getUsing(key, null)} call is semantically equivalent to simple
+ * {@code map.get(key)} call.
+ *
+ *
If the specified key if absent in the map, {@linkplain
+ * ChronicleMapBuilder#defaultValue(Object) default value} is taken or {@linkplain
+ * ChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) default value provider}
+ * is called. If it results to something different from {@code null}, this object is put
+ * to this map for the specified key, and then a new value is returned from this
+ * {@code getUsing()} call. Otherwise, if default value is {@code null} and default value
+ * provider, if specified, returns {@code null}, the map remains unchanged and {@code null}
+ * is returned. The provided {@code value} is untouched anyway in this case.
+ *
+ * @param key the key whose associated value is to be returned
+ * @param value the object to read value data in, if possible
+ * @return the value mapped to the specified key after this method call, or {@code null} if no
+ * value is mapped
+ * @see #get(Object)
+ * @see ChronicleMapBuilder#valueMarshallerAndFactory(BytesMarshaller, ObjectFactory)
*/
V getUsing(K key, V value);
@@ -41,15 +120,10 @@ public interface ChronicleMap extends ConcurrentMap, Closeable {
* Acquire a value for a key, creating if absent. If the value is Byteable, it will be assigned
* to reference the value, instead of copying the data.
*
- * @param key to lookup.
+ * @param key the key whose associated value is to be returned
* @param value to reuse if possible. If null, a new object will be created.
* @return value created or found.
*/
V acquireUsing(K key, V value);
- /**
- * @return The file or directory for this ChronicleMap
- */
- File file();
-
}
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
index 71c83fe3f..c400f0f43 100644
--- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
@@ -20,6 +20,7 @@ import net.openhft.chronicle.ChronicleHashBuilder;
import net.openhft.chronicle.ChronicleHashErrorListener;
import net.openhft.chronicle.ChronicleHashErrorListeners;
import net.openhft.chronicle.TimeProvider;
+import net.openhft.chronicle.map.serialization.AgileBytesMarshaller;
import net.openhft.chronicle.map.serialization.MetaBytesInterop;
import net.openhft.chronicle.map.serialization.MetaBytesWriter;
import net.openhft.chronicle.map.serialization.MetaProvider;
@@ -555,11 +556,21 @@ public class ChronicleMapBuilder implements Cloneable,
return toString().hashCode();
}
- public ChronicleMapBuilder replicators(byte identifier, Replicator... replicators) {
+ public ChronicleMapBuilder replicators(byte identifier, ReplicationConfig... replicationConfigs) {
this.identifier = identifier;
this.replicators.clear();
- for (Replicator replicator : replicators) {
+ for (ReplicationConfig replicationConfig : replicationConfigs) {
+
+ Replicator replicator;
+ if (replicationConfig instanceof TcpReplicationConfig) {
+ replicator = Replicators.tcp((TcpReplicationConfig) replicationConfig);
+ } else if (replicationConfig instanceof UdpReplicationConfig) {
+ replicator = Replicators.udp((UdpReplicationConfig) replicationConfig);
+ } else
+ throw new UnsupportedOperationException();
+
+
this.replicators.put(replicator.getClass(), replicator);
}
@@ -567,10 +578,10 @@ public class ChronicleMapBuilder implements Cloneable,
}
- public ChronicleMapBuilder channel(ReplicatingChannel.ChronicleChannel replicator) {
- this.identifier = replicator.identifier();
+ public ChronicleMapBuilder channel(ChannelProvider.ChronicleChannel chronicleChannel) {
+ this.identifier = chronicleChannel.identifier();
this.replicators.clear();
- replicators.put(replicator.getClass(), replicator);
+ replicators.put(chronicleChannel.getClass(), chronicleChannel);
return this;
}
@@ -629,6 +640,13 @@ public class ChronicleMapBuilder implements Cloneable,
return this;
}
+ public ChronicleMapBuilder valueMarshallerAndFactory(
+ @NotNull AgileBytesMarshaller valueMarshaller,
+ @NotNull ObjectFactory valueFactory) {
+ valueBuilder.agileMarshaller(valueMarshaller, valueFactory);
+ return this;
+ }
+
public ChronicleMapBuilder valueFactory(@NotNull ObjectFactory valueFactory) {
valueBuilder.factory(valueFactory);
return this;
@@ -773,7 +791,7 @@ public class ChronicleMapBuilder implements Cloneable,
return identifier != -1;
}
- private void preMapConstruction() {
+ void preMapConstruction() {
keyBuilder.objectSerializer(objectSerializer());
valueBuilder.objectSerializer(objectSerializer());
@@ -792,7 +810,7 @@ public class ChronicleMapBuilder implements Cloneable,
if (map instanceof ReplicatedChronicleMap) {
ReplicatedChronicleMap result = (ReplicatedChronicleMap) map;
for (Replicator replicator : replicators.values()) {
- Closeable token = replicator.applyTo(this, result, result);
+ Closeable token = replicator.applyTo(this, result, result, map);
if (replicators.size() == 1 && token.getClass() == UdpReplicator.class) {
LOG.warn(
"MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " +
diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleSet.java b/src/main/java/net/openhft/chronicle/map/ChronicleSet.java
deleted file mode 100644
index 5d3b92abf..000000000
--- a/src/main/java/net/openhft/chronicle/map/ChronicleSet.java
+++ /dev/null
@@ -1,8 +0,0 @@
-package net.openhft.chronicle.map;
-
-import java.io.Closeable;
-import java.util.Set;
-
-public interface ChronicleSet extends Set, Closeable {
- public long longSize();
-}
diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
index 26736295c..ad2723a86 100644
--- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
@@ -25,7 +25,6 @@ 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;
/**
@@ -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();
}
diff --git a/src/main/java/net/openhft/chronicle/map/ReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/ReplicationConfig.java
new file mode 100644
index 000000000..a72abed48
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/ReplicationConfig.java
@@ -0,0 +1,9 @@
+package net.openhft.chronicle.map;
+
+/**
+ * @author Rob Austin.
+ */
+public interface ReplicationConfig {
+
+
+}
diff --git a/src/main/java/net/openhft/chronicle/map/Replicator.java b/src/main/java/net/openhft/chronicle/map/Replicator.java
index 86ef2ed82..3dede414d 100644
--- a/src/main/java/net/openhft/chronicle/map/Replicator.java
+++ b/src/main/java/net/openhft/chronicle/map/Replicator.java
@@ -41,6 +41,7 @@ public abstract class Replicator {
* map configurations, not accessible via the map instance.
* @param map a replicated map instance. Provides basic tools for replication implementation.
* @param entryExternalizable the callback for ser/deser implementation in the replicator
+ * @param chronicleMap
* @return a {@code Closeable} token to control replication resources.
* It should be closed on closing the replicated map.
* @throws java.io.IOException if an io error occurred during the replicator setup
@@ -49,6 +50,6 @@ public abstract class Replicator {
* applied to a map (or the specified number of maps)
*/
protected abstract Closeable applyTo(ChronicleMapBuilder builder,
- Replica map, Replica.EntryExternalizable entryExternalizable)
+ Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap)
throws IOException;
}
diff --git a/src/main/java/net/openhft/chronicle/map/Replicators.java b/src/main/java/net/openhft/chronicle/map/Replicators.java
index faa766e85..aeb5118d3 100644
--- a/src/main/java/net/openhft/chronicle/map/Replicators.java
+++ b/src/main/java/net/openhft/chronicle/map/Replicators.java
@@ -20,31 +20,39 @@ package net.openhft.chronicle.map;
import java.io.Closeable;
import java.io.IOException;
-public final class Replicators {
+final class Replicators {
private Replicators() {
}
- public static Replicator tcp(final TcpReplicationConfig replicationConfig) {
+ static Replicator tcp(final TcpReplicationConfig replicationConfig) {
return new Replicator() {
@Override
protected Closeable applyTo(ChronicleMapBuilder builder,
- Replica map, Replica.EntryExternalizable entryExternalizable)
+ Replica replica, Replica.EntryExternalizable entryExternalizable,
+ final ChronicleMap chronicleMap)
throws IOException {
- return new TcpReplicator(map, entryExternalizable, replicationConfig,
- builder.entrySize());
+
+ final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder
+ .keyBuilder, builder.valueBuilder);
+
+ StatelessServerConnector statelessServerConnector = new StatelessServerConnector
+ (keyValueSerializer, chronicleMap);
+
+ return new TcpReplicator(replica, entryExternalizable, replicationConfig,
+ builder.entrySize(), statelessServerConnector);
}
};
}
- public static Replicator udp(
+ static Replicator udp(
final UdpReplicationConfig replicationConfig) {
return new Replicator() {
@Override
protected Closeable applyTo(ChronicleMapBuilder builder,
- Replica map, Replica.EntryExternalizable entryExternalizable)
+ Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap)
throws IOException {
return new UdpReplicator(map, entryExternalizable, replicationConfig,
builder.entrySize());
diff --git a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java
index c55306fd3..e95967aa9 100644
--- a/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java
+++ b/src/main/java/net/openhft/chronicle/map/SerializationBuilder.java
@@ -80,7 +80,7 @@ final class SerializationBuilder implements Cloneable {
if (role == Role.VALUE) {
factory = concreteClass(eClass) && marshallerUseFactory(eClass) ?
new NewInstanceObjectFactory(eClass) :
- NullObjectFactory.INSTANCE;
+ (ObjectFactory) NullObjectFactory.of();
}
if (concreteClass(eClass) && Byteable.class.isAssignableFrom(eClass)) {
diff --git a/src/main/java/net/openhft/chronicle/map/Serializer.java b/src/main/java/net/openhft/chronicle/map/Serializer.java
new file mode 100644
index 000000000..d30bb388a
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/Serializer.java
@@ -0,0 +1,81 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.map.serialization.*;
+import net.openhft.chronicle.map.threadlocal.Provider;
+import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies;
+import net.openhft.lang.io.Bytes;
+import net.openhft.lang.model.Byteable;
+import net.openhft.lang.model.constraints.NotNull;
+
+/**
+ * uses a {@code SerializationBuilder} to read and write object(s) as bytes
+ *
+ * @author Rob Austin.
+ */
+class Serializer {
+
+ final SizeMarshaller sizeMarshaller;
+ final BytesReader originalReader;
+ transient Provider> readerProvider;
+ final Object originalWriter;
+ transient Provider writerProvider;
+ final MetaBytesInterop originalMetaValueWriter;
+ final MetaProvider metaValueWriterProvider;
+
+ Serializer(final SerializationBuilder serializationBuilder) {
+
+ sizeMarshaller = serializationBuilder.sizeMarshaller();
+ originalMetaValueWriter = serializationBuilder.metaInterop();
+ metaValueWriterProvider = serializationBuilder.metaInteropProvider();
+
+ originalReader = serializationBuilder.reader();
+ originalWriter = serializationBuilder.interop();
+
+ readerProvider = Provider.of((Class) originalReader.getClass());
+ writerProvider = Provider.of((Class) originalWriter.getClass());
+ }
+
+ public O readMarshallable(@NotNull Bytes in) throws IllegalStateException {
+
+ final ThreadLocalCopies copies = readerProvider.getCopies(null);
+ final BytesReader valueReader = readerProvider.get(copies, originalReader);
+
+ try {
+ long valueSize = sizeMarshaller.readSize(in);
+ return valueReader.read(in, valueSize, null);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void writeMarshallable(O value, @NotNull Bytes out) {
+
+ ThreadLocalCopies copies = readerProvider.getCopies(null);
+ Object valueWriter = writerProvider.get(copies, originalWriter);
+ copies = writerProvider.getCopies(copies);
+
+ final long valueSize;
+
+ MetaBytesWriter metaValueWriter = null;
+
+ if ((value instanceof Byteable)) {
+ valueSize = ((Byteable) value).maxSize();
+ } else {
+ copies = writerProvider.getCopies(copies);
+ valueWriter = writerProvider.get(copies, originalWriter);
+ copies = metaValueWriterProvider.getCopies(copies);
+ metaValueWriter = metaValueWriterProvider.get(
+ copies, originalMetaValueWriter, valueWriter, value);
+ valueSize = metaValueWriter.size(valueWriter, value);
+ }
+
+ sizeMarshaller.writeSize(out, valueSize);
+
+ if (metaValueWriter != null)
+ metaValueWriter.write(valueWriter, out, value);
+ else
+ throw new UnsupportedOperationException("");
+
+ }
+
+}
diff --git a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java
new file mode 100644
index 000000000..397a69ee3
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java
@@ -0,0 +1,533 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.io.ByteBufferBytes;
+import net.openhft.lang.io.Bytes;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static net.openhft.chronicle.map.StatelessMapClient.EventId.*;
+
+/**
+ * **** THIS IS VERY MUCH IN DRAFT ***
+ *
+ * @author Rob Austin.
+ */
+class StatelessMapClient {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StatelessMapClient.class);
+
+
+ private final SocketChannel clientChannel;
+ public static final ByteBuffer STATELESS_CLIENT_IDENTIFER = ByteBuffer.wrap(new byte[]{-127});
+
+ public static enum EventId {
+ HEARTBEAT,
+ STATEFUL_UPDATE,
+ LONG_SIZE, SIZE,
+ IS_EMPTY,
+ CONTAINS_KEY,
+ CONTAINS_VALUE,
+ GET, PUT,
+ REMOVE,
+ CLEAR,
+ KEY_SET,
+ VALUES,
+ ENTRY_SET,
+ REPLACE,
+ REPLACE_WITH_OLD_AND_NEW_VALUE,
+ PUT_IF_ABSENT,
+ REMOVE_WITH_VALUE
+ }
+
+ private final ThreadLocal sourceBuffer = new ThreadLocal();
+ private final Map transactionIDs = new HashMap();
+
+ private final AtomicLong transactionID = new AtomicLong();
+
+ private long timeoutMs = TimeUnit.SECONDS.toMillis(20);
+
+
+ public StatelessMapClient(final KeyValueSerializer keyValueSerializer, final InetSocketAddress remote) throws IOException {
+
+ this.keyValueSerializer = keyValueSerializer;
+
+ clientChannel = SocketChannel.open();
+ clientChannel.connect(remote);
+
+ doHandShaking();
+ }
+
+ private void doHandShaking() throws IOException {
+
+
+ while (STATELESS_CLIENT_IDENTIFER.remaining() > 0) {
+ clientChannel.write(STATELESS_CLIENT_IDENTIFER);
+ }
+
+ in.buffer().clear();
+
+ while (in.buffer().position() <= 0) {
+ clientChannel.read(in.buffer());
+ }
+
+ in.limit(in.buffer().position());
+ in.position(0);
+ byte remoteIdentifier = in.readByte();
+
+ LOG.info("Attached to a map with a remote identifier=" + remoteIdentifier);
+
+ }
+
+
+ public File file() {
+ return null;
+ }
+
+ public void close() throws IOException {
+ throw new UnsupportedOperationException("This is not supported in the " + this.getClass()
+ .getSimpleName());
+ }
+
+ long nextUniqueTransaction() {
+ long time = System.currentTimeMillis();
+
+ long l = transactionID.get();
+ if (time > l) {
+ boolean b = transactionID.compareAndSet(l, time);
+ if (b) return time;
+ }
+
+ return transactionID.incrementAndGet();
+ }
+
+
+ private final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocate(1024));
+ private final ByteBufferBytes in = new ByteBufferBytes(ByteBuffer.allocate(1024));
+ private final KeyValueSerializer keyValueSerializer;
+
+
+ public synchronized V putIfAbsent(K key, V value) throws IOException {
+
+ long sizeLocation = writeEvent(PUT_IF_ABSENT);
+ writeKey(key);
+ writeValue(value);
+ return readKey(sizeLocation);
+ }
+
+
+ public synchronized boolean remove(Object key, Object value) throws IOException {
+ long sizeLocation = writeEvent(REMOVE_WITH_VALUE);
+ writeKey((K) key);
+ writeValue((V) value);
+
+ // get the data back from the server
+ return blockingFetch(sizeLocation).readBoolean();
+
+ }
+
+
+ public synchronized boolean replace(K key, V oldValue, V newValue) throws IOException {
+ long sizeLocation = writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE);
+ writeKey(key);
+ writeValue(oldValue);
+ writeValue(newValue);
+
+ // get the data back from the server
+ return blockingFetch(sizeLocation).readBoolean();
+ }
+
+
+ public synchronized V replace(K key, V value) throws IOException {
+ long sizeLocation = writeEvent(REPLACE);
+ writeKey(key);
+ writeValue(value);
+
+ // get the data back from the server
+ return readKey(sizeLocation);
+ }
+
+
+ public synchronized int size() throws IOException {
+
+ long sizeLocation = writeEvent(SIZE);
+
+ // get the data back from the server
+ return blockingFetch(sizeLocation).readInt();
+ }
+
+
+ public synchronized boolean isEmpty() throws IOException {
+ long sizeLocation = writeEvent(IS_EMPTY);
+
+ // get the data back from the server
+ return blockingFetch(sizeLocation).readBoolean();
+ }
+
+
+ public synchronized boolean containsKey(Object key) throws IOException {
+ long sizeLocation = writeEvent(CONTAINS_KEY);
+ writeKey((K) key);
+
+ // get the data back from the server
+ return blockingFetch(sizeLocation).readBoolean();
+
+ }
+
+
+ public synchronized boolean containsValue(Object value) throws IOException {
+ long sizeLocation = writeEvent(CONTAINS_VALUE);
+ writeValue((V) value);
+
+ // get the data back from the server
+ return blockingFetch(sizeLocation).readBoolean();
+ }
+
+
+ public synchronized long longSize() throws IOException {
+ long sizeLocation = writeEvent(LONG_SIZE);
+ // get the data back from the server
+ return blockingFetch(sizeLocation).readLong();
+ }
+
+
+ public synchronized V get(Object key) throws IOException {
+ long sizeLocation = writeEvent(GET);
+ writeKey((K) key);
+
+ // get the data back from the server
+ return readKey(sizeLocation);
+
+ }
+
+
+ public synchronized V getUsing(K key, V value) {
+ throw new UnsupportedOperationException("getUsing is not supported for stateless clients");
+ }
+
+
+ public synchronized V acquireUsing(K key, V value) {
+ throw new UnsupportedOperationException("getUsing is not supported for stateless clients");
+ }
+
+
+ public synchronized V put(K key, V value) throws IOException {
+
+ long sizeLocation = writeEvent(PUT);
+ writeKey(key);
+ writeValue(value);
+
+ // get the data back from the server
+ return readKey(sizeLocation);
+
+ }
+
+
+ public synchronized V remove(Object key) throws IOException {
+ long sizeLocation = writeEvent(REMOVE);
+ writeKey((K) key);
+
+ // get the data back from the server
+ return readKey(sizeLocation);
+
+ }
+
+
+ public synchronized void putAll(Map map) throws IOException {
+
+ for (Map.Entry e : map.entrySet()) {
+ put(e.getKey(), e.getValue());
+ }
+
+ }
+
+ private void writeEntries(Map map) {
+
+ final HashMap safeCopy = new HashMap(map);
+
+ out.writeStopBit(safeCopy.size());
+
+ final Set entries = (Set) safeCopy.entrySet();
+
+ for (Entry e : entries) {
+ writeKey(e.getKey());
+ writeValue(e.getValue());
+ }
+ }
+
+
+ public synchronized void clear() throws IOException {
+ long sizeLocation = writeEvent(CLEAR);
+
+ // get the data back from the server
+ blockingFetch(sizeLocation);
+
+ }
+
+ @NotNull
+
+ public synchronized Set keySet() throws IOException {
+ long sizeLocation = writeEvent(KEY_SET);
+
+ // get the data back from the server
+ return readKeySet(blockingFetch(sizeLocation));
+ }
+
+
+ @NotNull
+
+ public synchronized Collection values() throws IOException {
+ long sizeLocation = writeEvent(VALUES);
+
+ // get the data back from the server
+ final Bytes in = blockingFetch(sizeLocation);
+
+ final long size = in.readStopBit();
+
+ if (size > Integer.MAX_VALUE)
+ throw new IllegalStateException("size=" + size + " is too large.");
+
+ final ArrayList result = new ArrayList((int) size);
+
+ for (int i = 0; i < size; i++) {
+ result.add(keyValueSerializer.readValue(in));
+ }
+ return result;
+ }
+
+
+ @NotNull
+
+ public synchronized Set> entrySet() throws IOException {
+ long sizeLocation = writeEvent(ENTRY_SET);
+
+ // get the data back from the server
+ Bytes in = blockingFetch(sizeLocation);
+
+ long size = in.readStopBit();
+
+ Set> result = new HashSet>();
+
+ for (int i = 0; i < size; i++) {
+ K k = keyValueSerializer.readKey(in);
+ V v = keyValueSerializer.readValue(in);
+ result.add(new Entry(k, v));
+ }
+
+ return result;
+ }
+
+
+ private long writeEvent(EventId event) {
+ out.clear();
+ out.buffer().clear();
+
+ out.write((byte) event.ordinal());
+ long sizeLocation = markSizeLocation();
+ return sizeLocation;
+ }
+
+ public Buffer buffer() {
+ Buffer result = sourceBuffer.get();
+
+
+ if (result != null) {
+ return result;
+ }
+
+ result = new Buffer() {
+
+ volatile ByteBufferBytes buffer = null;
+
+
+ public void set(ByteBufferBytes source) {
+ buffer = source;
+ }
+
+
+ public ByteBufferBytes get() {
+ return buffer;
+ }
+ };
+
+ sourceBuffer.set(result);
+ return result;
+ }
+
+
+ /**
+ * blocks until a message is received from the server or a timeout is reached
+ *
+ * @param transactionId
+ * @return
+ */
+ private Bytes blockingFetchWithTransaction(long transactionId) {
+ final Buffer buffer = buffer();
+ buffer.set(null);
+
+ synchronized (this) {
+ transactionIDs.put(transactionId, buffer);
+ try {
+ buffer.wait(timeoutMs);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ final ByteBufferBytes bufferBytes = buffer.get();
+
+ if (bufferBytes == null)
+ throw new RuntimeException("Timed-out", new TimeoutException());
+
+ return bufferBytes;
+ }
+
+
+ class Entry implements Map.Entry {
+
+ final K key;
+ final V value;
+
+ /**
+ * Creates new entry.
+ */
+ Entry(K k1, V v) {
+ value = v;
+ key = k1;
+ }
+
+ public final K getKey() {
+ return key;
+ }
+
+ public final V getValue() {
+ return value;
+ }
+
+ public final V setValue(V newValue) {
+ V oldValue = value;
+ try {
+ StatelessMapClient.this.put((K) getKey(), (V) newValue);
+ } catch (IOException e) {
+ LOG.error("", e);
+ }
+ return oldValue;
+ }
+
+ public final boolean equals(Object o) {
+ if (!(o instanceof Map.Entry))
+ return false;
+ Map.Entry e = (Map.Entry) o;
+ Object k1 = getKey();
+ Object k2 = e.getKey();
+ if (k1 == k2 || (k1 != null && k1.equals(k2))) {
+ Object v1 = getValue();
+ Object v2 = e.getValue();
+ if (v1 == v2 || (v1 != null && v1.equals(v2)))
+ return true;
+ }
+ return false;
+ }
+
+ public final int hashCode() {
+ return (key == null ? 0 : key.hashCode()) ^
+ (value == null ? 0 : value.hashCode());
+ }
+
+ public final String toString() {
+ return getKey() + "=" + getValue();
+ }
+ }
+
+ private Bytes blockingFetch(long sizeLocation) throws IOException {
+ long transactionId = nextUniqueTransaction();
+
+ LOG.info("sending data with transactionId=" + transactionId);
+ out.writeLong(transactionId);
+ writeSizeAt(sizeLocation);
+
+
+ out.buffer().limit((int) out.position());
+ while (out.buffer().remaining() > 0) {
+ int write = clientChannel.write(out.buffer());
+ System.out.println(write);
+ }
+
+ in.clear();
+ in.buffer().clear();
+
+ // read size
+ while (in.buffer().position() < 2) {
+ clientChannel.read(in.buffer());
+ }
+
+ int size = in.readUnsignedShort();
+
+ while (in.buffer().position() < size + 2) {
+ clientChannel.read(in.buffer());
+ }
+
+ long inTransactionId = in.readLong();
+
+ if (inTransactionId != transactionId) {
+ throw new IllegalStateException("the received transaction-id=" + inTransactionId +
+ ", does not match the expected transaction-id=" + transactionId);
+ }
+
+ return in;
+ }
+
+ private void writeSizeAt(long locationOfSize) {
+ long size = out.position() - locationOfSize;
+ out.writeUnsignedShort(locationOfSize, (int) size - 2);
+ }
+
+ private long markSizeLocation() {
+ long position = out.position();
+ out.skip(2);
+ return position;
+ }
+
+ private void writeKey(K key) {
+ keyValueSerializer.writeKey(key, out);
+ }
+
+ private V readKey(final long sizeLocation) throws IOException {
+ return keyValueSerializer.readValue(blockingFetch(sizeLocation));
+ }
+
+ private void writeValue(V value) {
+ keyValueSerializer.writeValue(value, out);
+ }
+
+
+ private Set readKeySet(Bytes in) {
+ long size = in.readStopBit();
+ final HashSet result = new HashSet<>();
+
+ for (long i = 0; i < size; i++) {
+ result.add(keyValueSerializer.readKey(in));
+ }
+ return result;
+ }
+
+}
+
+
+interface Buffer {
+
+ void set(ByteBufferBytes source);
+
+ ByteBufferBytes get();
+}
diff --git a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java
new file mode 100644
index 000000000..ca3f949e4
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java
@@ -0,0 +1,317 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.io.Bytes;
+import net.openhft.lang.model.constraints.NotNull;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * @author Rob Austin.
+ */
+class StatelessServerConnector {
+
+
+ private final KeyValueSerializer keyValueSerializer;
+ private final ChronicleMap map;
+
+
+ public StatelessServerConnector(KeyValueSerializer keyValueSerializer, ChronicleMap map) {
+ this.keyValueSerializer = keyValueSerializer;
+ this.map = map;
+ }
+
+
+ public void processStatelessEvent(byte eventId, @NotNull Bytes in, @NotNull Bytes out) {
+
+
+ StatelessMapClient.EventId event = StatelessMapClient.EventId.values()[eventId];
+ switch (event) {
+
+ case LONG_SIZE:
+ longSize(in, out);
+ break;
+
+ case IS_EMPTY:
+ isEmpty(in, out);
+ break;
+
+ case CONTAINS_KEY:
+ containsKey(in, out);
+ break;
+ case CONTAINS_VALUE:
+ containsValue(in, out);
+ break;
+ case GET:
+ get(in, out);
+ break;
+ case PUT:
+ put(in, out);
+ break;
+ case REMOVE:
+ remove(in, out);
+ break;
+ case CLEAR:
+ clear(in, out);
+ break;
+ case KEY_SET:
+ keySet(in, out);
+ break;
+ case VALUES:
+ values(in, out);
+ break;
+ case ENTRY_SET:
+ entrySet(in, out);
+ break;
+ case REPLACE:
+ replace(in, out);
+ break;
+ case REPLACE_WITH_OLD_AND_NEW_VALUE:
+ replaceWithOldAndNew(in, out);
+ break;
+ case PUT_IF_ABSENT:
+ putIfAbsent(in, out);
+ break;
+ case REMOVE_WITH_VALUE:
+ removeWithValue(in, out);
+ break;
+
+ case SIZE:
+ size(in, out);
+ break;
+
+ default:
+ throw new IllegalStateException("unsupported event=" + event);
+
+ }
+
+ }
+
+ private void removeWithValue(Bytes in, Bytes out) {
+ boolean result = map.remove(readKey(in), readValue(in));
+ out.writeBoolean(result);
+ }
+
+ private void replaceWithOldAndNew(Bytes in, Bytes out) {
+
+ final K key = readKey(in);
+ V oldValue = readValue(in);
+ V newValue = readValue(in);
+
+ long sizeLocation = reflectTransactionId(in, out);
+ map.replace(key, oldValue, newValue);
+
+ writeSizeAt(sizeLocation, out);
+ }
+
+
+ public void longSize(Bytes in, Bytes out) {
+ long sizeLocation = reflectTransactionId(in, out);
+ out.writeLong(map.longSize());
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void size(Bytes in, Bytes out) {
+ long sizeLocation = reflectTransactionId(in, out);
+ out.writeInt(map.size());
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void isEmpty(Bytes in, Bytes out) {
+ long sizeLocation = reflectTransactionId(in, out);
+ out.writeBoolean(map.isEmpty());
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void containsKey(Bytes in, Bytes out) {
+ K k = readKey(in);
+ long sizeLocation = reflectTransactionId(in, out);
+ out.writeBoolean(map.containsKey(k));
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void containsValue(Bytes in, Bytes out) {
+ V v = readValue(in);
+ long sizeLocation = reflectTransactionId(in, out);
+ out.writeBoolean(map.containsValue(v));
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void get(Bytes in, Bytes out) {
+ K k = readKey(in);
+ long sizeLocation = reflectTransactionId(in, out);
+ writeValue(map.get(k), out);
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void put(Bytes in, Bytes out) {
+ K k = readKey(in);
+ V v = readValue(in);
+ long sizeLocation = reflectTransactionId(in, out);
+ writeValue(map.put(k, v), out);
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void remove(Bytes in, Bytes out) {
+ final V value = map.remove(readKey(in));
+ long sizeLocation = reflectTransactionId(in, out);
+ writeValue(value, out);
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void putAll(Bytes in, Bytes out) {
+ map.putAll(readEntries(in));
+ long sizeLocation = reflectTransactionId(in, out);
+ writeSizeAt(sizeLocation, out);
+ }
+
+ private Map readEntries(Bytes in) {
+
+ long size = in.readStopBit();
+ final HashMap result = new HashMap();
+
+ for (long i = 0; i < size; i++) {
+ result.put(readKey(in), readValue(in));
+ }
+ return result;
+
+ }
+
+ public void clear(Bytes in, Bytes out) {
+ map.clear();
+ long sizeLocation = reflectTransactionId(in, out);
+ writeSizeAt(sizeLocation, out);
+ }
+
+
+ public void keySet(Bytes in, Bytes out) {
+ long sizeLocation = reflectTransactionId(in, out);
+
+ final Set ks = map.keySet();
+ out.writeStopBit(ks.size());
+ for (K key : ks) {
+ writeKey(key, out);
+ }
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void values(Bytes in, Bytes out) {
+ long sizeLocation = reflectTransactionId(in, out);
+ final Collection values = map.values();
+ out.writeStopBit(values.size());
+ for (final V value : values) {
+ writeValue(value, out);
+ }
+ writeSizeAt(sizeLocation, out);
+ }
+
+ public void entrySet(Bytes in, Bytes out) {
+ long sizeLocation = reflectTransactionId(in, out);
+
+ final Set> entries = map.entrySet();
+ out.writeStopBit(entries.size());
+ for (Map.Entry e : entries) {
+ writeKey(e.getKey(), out);
+ writeValue(e.getValue(), out);
+ }
+ }
+
+ public void putIfAbsent(Bytes in, Bytes out) {
+ K key = readKey(in);
+ V v = readValue(in);
+ long sizeLocation = reflectTransactionId(in, out);
+ writeValue(map.putIfAbsent(key, v), out);
+ writeSizeAt(sizeLocation, out);
+ }
+
+
+ public void replace(Bytes in, Bytes out) {
+ K k = readKey(in);
+ V v = readValue(in);
+
+ long sizeLocation = reflectTransactionId(in, out);
+ map.replace(k, v);
+ writeSizeAt(sizeLocation, out);
+ }
+
+ /**
+ * write the keysize and the key to the the {@code target} buffer
+ *
+ * @param key the key of the map
+ */
+ private void writeKey(K key, Bytes out) {
+ keyValueSerializer.writeKey(key, out);
+ }
+
+ private long reflectTransactionId(Bytes in, Bytes out) {
+ long sizeLocation = out.position();
+ out.skip(2);
+ long transactionId = in.readLong();
+ out.writeLong(transactionId);
+ return sizeLocation;
+ }
+
+ private void writeValue(V value, final Bytes out) {
+ keyValueSerializer.writeValue(value, out);
+ }
+
+ private K readKey(Bytes in) {
+ return keyValueSerializer.readKey(in);
+ }
+
+ private V readValue(Bytes in) {
+ return keyValueSerializer.readValue(in);
+ }
+
+ private void writeSizeAt(long locationOfSize, Bytes out) {
+ long size = out.position() - locationOfSize;
+ out.writeUnsignedShort((int) size);
+ }
+}
+
+class KeyValueSerializer {
+
+ private final Serializer valueSerializer;
+ private final Serializer keySerializer;
+
+ KeyValueSerializer(SerializationBuilder keySerializer,
+ SerializationBuilder valueSerializer) {
+ this.keySerializer = new Serializer(keySerializer);
+ this.valueSerializer = new Serializer(valueSerializer);
+ }
+
+ V readValue(Bytes in) {
+ if (in.readBoolean())
+ return null;
+ return valueSerializer.readMarshallable(in);
+ }
+
+ K readKey(Bytes in) {
+ if (in.readBoolean())
+ return null;
+
+ return keySerializer.readMarshallable(in);
+ }
+
+ /**
+ * write the keysize and the key to the the {@code target} buffer
+ *
+ * @param key the key of the map
+ */
+ void writeKey(K key, Bytes out) {
+
+ out.writeBoolean(key == null);
+ if (key != null)
+ keySerializer.writeMarshallable(key, out);
+ }
+
+ void writeValue(V value, Bytes out) {
+ out.writeBoolean(value == null);
+ if (value != null)
+ valueSerializer.writeMarshallable(value, out);
+ }
+
+
+}
diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/TcpReplicationConfig.java
index b213f4455..5bce434ab 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 ReplicationConfig {
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/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
index 61a013ce3..be8fd7b70 100644
--- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java
@@ -17,6 +17,7 @@
package net.openhft.chronicle.map;
import net.openhft.lang.io.ByteBufferBytes;
+import net.openhft.lang.io.Bytes;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -32,10 +33,11 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import static java.nio.channels.SelectionKey.*;
+import static net.openhft.chronicle.map.StatelessMapClient.EventId.HEARTBEAT;
/**
- * Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a socket
- * connection {@see net.openhft.map.OutSocketReplicator}
+ * Used with a {@see net.openhft.map.ReplicatedSharedHashMap} to send data between the maps using a
+ * socket connection {@see net.openhft.map.OutSocketReplicator}
*
* @author Rob Austin.
*/
@@ -43,6 +45,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName());
private static final int BUFFER_SIZE = 0x100000; // 1MB
+ public static final int STATELESS_CLIENT = -127;
+ public static final byte NOT_SET = (byte) HEARTBEAT.ordinal();
private final SelectionKey[] selectionKeysStore = new SelectionKey[Byte.MAX_VALUE + 1];
// used to instruct the selector thread to set OP_WRITE on a key correlated by the bit index in the
@@ -57,22 +61,28 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
private final Replica.EntryExternalizable externalizable;
private final TcpReplicationConfig replicationConfig;
+ private final StatelessServerConnector statelessServerConnector;
+
private long selectorTimeout;
/**
- * @param maxEntrySizeBytes used to check that the last entry will fit into the buffer, it can not be
- * smaller than the size of and entry, if it is set smaller the buffer will over
- * flow, it can be larger then the entry, but setting it too large reduces the
- * workable space in the buffer.
+ * @param maxEntrySizeBytes used to check that the last entry will fit into the buffer,
+ * it can not be smaller than the size of and entry, if it is
+ * set smaller the buffer will over flow, it can be larger then
+ * the entry, but setting it too large reduces the workable
+ * space in the buffer.
+ * @param statelessServerConnector
* @throws IOException
*/
TcpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable externalizable,
@NotNull final TcpReplicationConfig replicationConfig,
- final int maxEntrySizeBytes) throws IOException {
+ final int maxEntrySizeBytes,
+ StatelessServerConnector statelessServerConnector) throws IOException {
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(),
maxEntrySizeBytes);
+ this.statelessServerConnector = statelessServerConnector;
final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig();
long throttleBucketInterval = TimeUnit.MILLISECONDS.convert(throttlingConfig
@@ -251,7 +261,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
}
/**
- * check to see if we have lost connection with the remote node and if we have attempts a reconnect.
+ * check to see if we have lost connection with the remote node and if we have attempts a
+ * reconnect.
*
* @param key the key relating to the heartbeat that we are checking
* @param approxTimeOutTime the approximate time in milliseconds
@@ -374,8 +385,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
}
/**
- * this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear all the
- * old bootstrap information
+ * this can be called when a new CHM is added to a cluster, we have to rebootstrap so will clear
+ * all the old bootstrap information
*
* @param key the nio SelectionKey
*/
@@ -387,7 +398,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
}
/**
- * used to exchange identifiers and timestamps and heartbeat intervals between the server and client
+ * used to exchange identifiers and timestamps and heartbeat intervals between the server and
+ * client
*
* @param key the SelectionKey relating to the this cha
* @param socketChannel
@@ -404,6 +416,11 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
final byte remoteIdentifier = reader.identifierFromBuffer();
+ if (remoteIdentifier == STATELESS_CLIENT) {
+ attached.handShakingComplete = true;
+ return;
+ }
+
if (remoteIdentifier == Byte.MIN_VALUE)
return;
@@ -425,7 +442,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
final SocketAddress remoteAddress = socketChannel.getRemoteAddress();
- if ( (identifierListener != null && !identifierListener.isIdentifierUnique(remoteIdentifier,
+ if ((identifierListener != null && !identifierListener.isIdentifierUnique(remoteIdentifier,
remoteAddress)) || remoteIdentifier == localIdentifier)
// throwing this exception will cause us to disconnect, both the client and server
@@ -477,7 +494,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
// now we're finished we can get on with reading the entries
attached.handShakingComplete = true;
attached.remoteModificationIterator.dirtyEntries(attached.remoteBootstrapTimestamp);
- reader.entriesFromBuffer();
+ reader.entriesFromBuffer(attached);
}
}
@@ -531,17 +548,17 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
attached.entryReader.lastHeartBeatReceived = approxTime;
if (attached.isHandShakingComplete()) {
- attached.entryReader.entriesFromBuffer();
+ attached.entryReader.entriesFromBuffer(attached);
} else {
doHandShaking(key, socketChannel);
}
}
/**
- * sets interestOps to "selector keys",The change to interestOps much be on the same thread as the
- * selector. This class, allows via {@link AbstractChannelReplicator .KeyInterestUpdater#set(int)} to
- * holds a pending change in interestOps ( via a bitset ), this change is processed later on the same
- * thread as the selector
+ * sets interestOps to "selector keys",The change to interestOps much be on the same thread as
+ * the selector. This class, allows via {@link AbstractChannelReplicator
+ * .KeyInterestUpdater#set(int)} to holds a pending change in interestOps ( via a bitset ),
+ * this change is processed later on the same thread as the selector
*/
private static class KeyInterestUpdater {
@@ -572,8 +589,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
}
/**
- * @param keyIndex the index of the key that has changed, the list of keys is provided by the
- * constructor {@link KeyInterestUpdater(int, SelectionKey[])}
+ * @param keyIndex the index of the key that has changed, the list of keys is provided by
+ * the constructor {@link KeyInterestUpdater(int, SelectionKey[])}
*/
public void set(int keyIndex) {
changeOfOpWriteRequired.set(keyIndex);
@@ -773,12 +790,19 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
}
+ /**
+ * @return the buffer messages can be written to
+ */
+ Bytes buffer() {
+ return in;
+ }
+
/**
* writes the timestamp into the buffer
*
* @param localIdentifier the current nodes identifier
*/
- void identifierToBuffer(final int localIdentifier) {
+ void identifierToBuffer(final byte localIdentifier) {
in.writeByte(localIdentifier);
}
@@ -792,7 +816,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
}
/**
- * writes all the entries that have changed, to the buffer which will later be written to TCP/IP
+ * writes all the entries that have changed, to the buffer which will later be written to
+ * TCP/IP
*
* @param modificationIterator a record of which entries have modification
* @param selectionKey
@@ -804,7 +829,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
final SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
final Attached attached = (Attached) selectionKey.attachment();
- // this can occur when new CHM's are added to a cluster
+ // this can occur when new map are added to a channel
final boolean handShakingComplete = attached.isHandShakingComplete();
for (; ; ) {
@@ -867,9 +892,15 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
}
/**
- * used to send an single zero byte if we have not send any data for up to the localHeartbeatInterval
+ * used to send an single zero byte if we have not send any data for up to the
+ * localHeartbeatInterval
*/
private void writeHeartbeatToBuffer() {
+
+ // denotes the state - 0 for a heartbeat
+ in.writeByte(HEARTBEAT.ordinal());
+
+ // denotes the size in bytes
in.writeUnsignedShort(0);
}
@@ -879,8 +910,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
/**
- * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE will get
- * added back in as soon as we have data to write
+ * removes back in the OP_WRITE from the selector, otherwise it'll spin loop. The OP_WRITE
+ * will get added back in as soon as we have data to write
*
* @param socketChannel the socketChannel we wish to stop writing to
* @param attached data associated with the socketChannels key
@@ -914,8 +945,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
private final ByteBuffer in;
private final ByteBufferBytes out;
public long lastHeartBeatReceived = System.currentTimeMillis();
- // we use Integer.MIN_VALUE as N/A
- private int sizeOfNextEntry = Integer.MIN_VALUE;
+
+ private int sizeInBytes;
+ private byte state;
private TcpSocketChannelEntryReader() {
in = ByteBuffer.allocateDirect(replicationConfig.packetSize() + maxEntrySizeBytes);
@@ -943,45 +975,79 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
/**
* reads entries from the buffer till empty
*
+ * @param attached
* @throws InterruptedException
*/
- private void entriesFromBuffer() throws InterruptedException, IOException {
-
+ private void entriesFromBuffer(Attached attached) throws InterruptedException, IOException {
for (; ; ) {
out.limit(in.position());
// its set to MIN_VALUE when it should be read again
- if (sizeOfNextEntry == Integer.MIN_VALUE) {
- if (out.remaining() < SIZE_OF_SHORT) {
+ if (state == NOT_SET) {
+ if (out.remaining() < SIZE_OF_SHORT + 1 ) {
return;
}
- int value = out.readUnsignedShort();
+ // state is used for both heartbeat and stateless
+ state = out.readByte();
- // this is the heartbeat
- if (value == 0)
+
+
+ sizeInBytes = out.readUnsignedShort();
+
+ if (state == 0 && sizeInBytes != 0) {
+ int i = 1;
+ }
+
+ // this is the :
+ // -- heartbeat if its 0
+ // -- stateful update if its 1
+ // -- the id of the stateful event
+ if (state == NOT_SET)
continue;
-
- sizeOfNextEntry = value;
}
- if (out.remaining() < sizeOfNextEntry) {
+
+ if (out.remaining() < sizeInBytes) {
return;
}
- final long nextEntryPos = out.position() + sizeOfNextEntry;
+ if (sizeInBytes == 0) {
+ int j = 1;
+ }
+
+
+ final long nextEntryPos = out.position() + sizeInBytes;
final long limit = out.limit();
out.limit(nextEntryPos);
- externalizable.readExternalEntry(out);
+
+ boolean isStateless = (state != 1);
+
+ if (isStateless) {
+ if (statelessServerConnector == null) {
+
+ LOG.error("", new IllegalArgumentException("received an event " +
+ "from a stateless map, stateless maps are not " +
+ "currently supported when using Chronicle Channels"));
+ } else
+
+ statelessServerConnector.processStatelessEvent(state,
+ out, attached.entryWriter.buffer());
+ // to allow the sizeInBytes to be read the next time around
+
+
+ } else
+ externalizable.readExternalEntry(out);
out.limit(limit);
// skip onto the next entry
out.position(nextEntryPos);
- // to allow the sizeOfNextEntry to be read the next time around
- sizeOfNextEntry = Integer.MIN_VALUE;
+
+ state = NOT_SET;
+ sizeInBytes = 0;
}
}
@@ -1016,7 +1082,12 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
* @return the timestamp or -1 if unsuccessful
*/
long remoteBootstrapTimestamp() {
- return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE;
+ if (out.remaining() >= 8)
+ return out.readLong();
+ else
+ return Long.MIN_VALUE;
+
+ // return (out.remaining() >= 8) ? out.readLong() : Long.MIN_VALUE;
}
public long remoteHeartbeatIntervalFromBuffer() {
diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java b/src/main/java/net/openhft/chronicle/map/UdpReplicationConfig.java
index 0e02d9e9d..15a3382d7 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 ReplicationConfig {
/**
* 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/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
index 1a2189a09..d34852d43 100644
--- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
@@ -133,7 +133,8 @@ class VanillaChronicleMap,
this.removeReturnsNull = builder.removeReturnsNull();
this.actualSegments = builder.actualSegments();
- this.entriesPerSegment = builder.actualEntriesPerSegment();
+ // align by 8 because otherwise sizeOfBitSets() might address less blocks than specified
+ this.entriesPerSegment = align8(builder.actualEntriesPerSegment());
this.metaDataBytes = builder.metaDataBytes();
this.eventListener = builder.eventListener();
@@ -158,6 +159,10 @@ class VanillaChronicleMap,
return (l + 63) & ~63;
}
+ static int align8(int n) {
+ return (n + 7) & ~7;
+ }
+
void initTransients() {
keyReaderProvider = Provider.of((Class) originalKeyReader.getClass());
keyInteropProvider = Provider.of((Class) originalKeyInterop.getClass());
@@ -555,8 +560,8 @@ class VanillaChronicleMap,
long start = bytes.startAddr() + SEGMENT_HEADER;
createHashLookups(start);
start += align64(sizeOfMultiMap() + sizeOfMultiMapBitSet()) * multiMapsPerSegment();
- final NativeBytes bsBytes = new NativeBytes(
- tmpBytes.objectSerializer(), start, start + sizeOfBitSets(), null);
+ final NativeBytes bsBytes = new NativeBytes(tmpBytes.objectSerializer(),
+ start, start + ((entriesPerSegment + 7) / 8), null);
freeList = new SingleThreadedDirectBitSet(bsBytes);
start += numberOfBitSets() * sizeOfBitSets();
entriesOffset = start - bytes.startAddr();
@@ -914,15 +919,23 @@ class VanillaChronicleMap,
);
}
}
+ updateNextPosToSearchFrom(ret, blocks);
+ } else {
+ // if bit at nextPosToSearchFrom is clear, it was skipped because
+ // more than 1 block was requested. Don't move nextPosToSearchFrom
+ // in this case. blocks == 1 clause is just a fast path.
+ if (blocks == 1 || freeList.isSet(nextPosToSearchFrom)) {
+ updateNextPosToSearchFrom(ret, blocks);
+ }
}
- // if bit at nextPosToSearchFrom is clear, it was skipped because
- // more than 1 block was requested. Don't move nextPosToSearchFrom
- // in this case. blocks == 1 clause is just a fast path.
- if (blocks == 1 || freeList.isSet(nextPosToSearchFrom))
- nextPosToSearchFrom = ret + blocks;
return ret;
}
+ private void updateNextPosToSearchFrom(int allocated, int blocks) {
+ if ((nextPosToSearchFrom = allocated + blocks) >= freeList.size())
+ nextPosToSearchFrom = 0;
+ }
+
private boolean realloc(int fromPos, int oldBlocks, int newBlocks) {
if (freeList.allClear(fromPos + oldBlocks, fromPos + newBlocks)) {
freeList.set(fromPos + oldBlocks, fromPos + newBlocks);
@@ -1188,6 +1201,7 @@ class VanillaChronicleMap,
try {
hashLookup.clear();
freeList.clear();
+ nextPosToSearchFrom = 0;
resetSize();
} finally {
unlock();
diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSet.java b/src/main/java/net/openhft/chronicle/set/ChronicleSet.java
new file mode 100644
index 000000000..c167caad1
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/set/ChronicleSet.java
@@ -0,0 +1,10 @@
+package net.openhft.chronicle.set;
+
+import net.openhft.chronicle.ChronicleHash;
+
+import java.io.Closeable;
+import java.util.Set;
+
+public interface ChronicleSet extends Set, ChronicleHash {
+ public long longSize();
+}
diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
index ec344ca5b..1b2571609 100644
--- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
@@ -23,26 +23,24 @@ import net.openhft.chronicle.map.*;
import net.openhft.lang.io.serialization.BytesMarshaller;
import net.openhft.lang.io.serialization.BytesMarshallerFactory;
import net.openhft.lang.io.serialization.ObjectSerializer;
+import net.openhft.lang.io.serialization.impl.NullObjectFactory;
import org.jetbrains.annotations.NotNull;
import java.io.File;
import java.io.IOException;
-import java.io.Serializable;
-import java.util.AbstractSet;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.Set;
import java.util.concurrent.TimeUnit;
public class ChronicleSetBuilder
implements ChronicleHashBuilder, ChronicleSetBuilder> {
- private ChronicleMapBuilder chronicleMapBuilder;
+ private ChronicleMapBuilder chronicleMapBuilder;
ChronicleSetBuilder(Class keyClass) {
- chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, Void.class)
- .entryAndValueAlignment(Alignment.NO_ALIGNMENT);
+ chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, DummyValue.class)
+ .entryAndValueAlignment(Alignment.NO_ALIGNMENT)
+ .valueMarshallerAndFactory(DummyValueMarshaller.INSTANCE,
+ NullObjectFactory.of());
}
public static ChronicleSetBuilder of(Class keyClass) {
@@ -88,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 (and each
+ *
+ * (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}.)
*
@@ -165,11 +164,18 @@ public class ChronicleSetBuilder
}
@Override
- public ChronicleSetBuilder replicators(byte identifier, Replicator... replicators) {
+ public ChronicleSetBuilder replicators(byte identifier, ReplicationConfig... replicators) {
chronicleMapBuilder.replicators(identifier, replicators);
return this;
}
+ @Override
+ public ChronicleSetBuilder channel(ChannelProvider.ChronicleChannel chronicleChannel) {
+ chronicleMapBuilder.channel(chronicleChannel);
+ return this;
+ }
+
+
@Override
public String toString() {
return " ChronicleSetBuilder{" +
@@ -190,12 +196,6 @@ public class ChronicleSetBuilder
return toString().hashCode();
}
- public ChronicleSetBuilder replicators(byte identifier, Replicator replicator) {
- chronicleMapBuilder.replicators(identifier, replicator);
- return this;
- }
-
-
@Override
public ChronicleSetBuilder timeProvider(TimeProvider timeProvider) {
chronicleMapBuilder.timeProvider(timeProvider);
@@ -229,108 +229,15 @@ public class ChronicleSetBuilder
@Override
public ChronicleSet create(File file) throws IOException {
- final ChronicleMap map = chronicleMapBuilder.create(file);
+ final ChronicleMap map = chronicleMapBuilder.create(file);
return new SetFromMap(map);
}
@Override
public ChronicleSet create() throws IOException {
- final ChronicleMap map = chronicleMapBuilder.create();
+ final ChronicleMap map = chronicleMapBuilder.create();
return new SetFromMap(map);
}
- private static class SetFromMap extends AbstractSet
- implements ChronicleSet, Serializable {
-
- private final ChronicleMap m; // The backing map
- private transient Set s; // Its keySet
-
- SetFromMap(ChronicleMap map) {
- if (!map.isEmpty())
- throw new IllegalArgumentException("Map is non-empty");
- m = map;
- s = map.keySet();
- }
-
- public void clear() {
- m.clear();
- }
-
- public int size() {
- return m.size();
- }
-
- public boolean isEmpty() {
- return m.isEmpty();
- }
-
- public boolean contains(Object o) {
- return m.containsKey(o);
- }
-
- public boolean remove(Object o) {
- return m.remove(o) != null;
- }
-
- public boolean add(E e) {
- return m.put(e, (Void) null) == null;
- }
-
- public Iterator iterator() {
- return s.iterator();
- }
-
- public Object[] toArray() {
- return s.toArray();
- }
-
- public T[] toArray(T[] a) {
- return s.toArray(a);
- }
-
- public String toString() {
- return s.toString();
- }
-
- public int hashCode() {
- return s.hashCode();
- }
-
- public boolean equals(Object o) {
- return o == this || s.equals(o);
- }
-
- public boolean containsAll(Collection c) {
- return s.containsAll(c);
- }
-
- public boolean removeAll(Collection c) {
- return s.removeAll(c);
- }
-
- public boolean retainAll(Collection c) {
- return s.retainAll(c);
- }
- // addAll is the only inherited implementation
-
- private static final long serialVersionUID = 2454657854757543876L;
-
- private void readObject(java.io.ObjectInputStream stream)
- throws IOException, ClassNotFoundException {
- stream.defaultReadObject();
- s = m.keySet();
- }
-
- @Override
- public long longSize() {
- return m.longSize();
- }
-
- @Override
- public void close() throws IOException {
- m.close();
- }
- }
-
}
diff --git a/src/main/java/net/openhft/chronicle/set/DummyValue.java b/src/main/java/net/openhft/chronicle/set/DummyValue.java
new file mode 100644
index 000000000..1d73a070b
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/set/DummyValue.java
@@ -0,0 +1,21 @@
+/*
+ * 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.set;
+
+enum DummyValue {
+ DUMMY_VALUE
+}
diff --git a/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java b/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java
new file mode 100644
index 000000000..85bb16462
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/set/DummyValueMarshaller.java
@@ -0,0 +1,71 @@
+/*
+ * 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.set;
+
+import net.openhft.chronicle.map.serialization.AgileBytesMarshaller;
+import net.openhft.lang.io.Bytes;
+
+import static net.openhft.chronicle.set.DummyValue.DUMMY_VALUE;
+
+enum DummyValueMarshaller implements AgileBytesMarshaller {
+ INSTANCE;
+
+ @Override
+ public boolean startsWith(Bytes bytes, DummyValue dummyValue) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long hash(DummyValue dummyValue) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public DummyValue read(Bytes bytes, long size) {
+ return DUMMY_VALUE;
+ }
+
+ @Override
+ public DummyValue read(Bytes bytes, long size, DummyValue dummyValue) {
+ return DUMMY_VALUE;
+ }
+
+ @Override
+ public long size(DummyValue dummyValue) {
+ return 0L;
+ }
+
+ @Override
+ public void write(Bytes bytes, DummyValue dummyValue) {
+ // do nothing
+ }
+
+ @Override
+ public int sizeEncodingSize(long size) {
+ return 0;
+ }
+
+ @Override
+ public void writeSize(Bytes bytes, long size) {
+ // do nothing
+ }
+
+ @Override
+ public long readSize(Bytes bytes) {
+ return 0L;
+ }
+}
diff --git a/src/main/java/net/openhft/chronicle/set/SetFromMap.java b/src/main/java/net/openhft/chronicle/set/SetFromMap.java
new file mode 100644
index 000000000..cb513154c
--- /dev/null
+++ b/src/main/java/net/openhft/chronicle/set/SetFromMap.java
@@ -0,0 +1,127 @@
+/*
+ * 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.set;
+
+import net.openhft.chronicle.map.ChronicleMap;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.AbstractSet;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Set;
+
+import static net.openhft.chronicle.set.DummyValue.DUMMY_VALUE;
+
+class SetFromMap extends AbstractSet
+ implements ChronicleSet, Serializable {
+
+ private final ChronicleMap m; // The backing map
+ private transient Set s; // Its keySet
+
+ SetFromMap(ChronicleMap map) {
+ if (!map.isEmpty())
+ throw new IllegalArgumentException("Map is non-empty");
+ m = map;
+ s = map.keySet();
+ }
+
+ public void clear() {
+ m.clear();
+ }
+
+ public int size() {
+ return m.size();
+ }
+
+ public boolean isEmpty() {
+ return m.isEmpty();
+ }
+
+ public boolean contains(Object o) {
+ return m.containsKey(o);
+ }
+
+ public boolean remove(Object o) {
+ return m.remove(o) == DUMMY_VALUE;
+ }
+
+ public boolean add(E e) {
+ return m.putIfAbsent(e, DUMMY_VALUE) == null;
+ }
+
+ public Iterator iterator() {
+ return s.iterator();
+ }
+
+ public Object[] toArray() {
+ return s.toArray();
+ }
+
+ public T[] toArray(T[] a) {
+ return s.toArray(a);
+ }
+
+ public String toString() {
+ return s.toString();
+ }
+
+ public int hashCode() {
+ return s.hashCode();
+ }
+
+ public boolean equals(Object o) {
+ return o == this || s.equals(o);
+ }
+
+ public boolean containsAll(Collection c) {
+ return s.containsAll(c);
+ }
+
+ public boolean removeAll(Collection c) {
+ return s.removeAll(c);
+ }
+
+ public boolean retainAll(Collection c) {
+ return s.retainAll(c);
+ }
+ // addAll is the only inherited implementation
+
+ private static final long serialVersionUID = 2454657854757543876L;
+
+ private void readObject(java.io.ObjectInputStream stream)
+ throws IOException, ClassNotFoundException {
+ stream.defaultReadObject();
+ s = m.keySet();
+ }
+
+ @Override
+ public long longSize() {
+ return m.longSize();
+ }
+
+ @Override
+ public File file() {
+ return m.file();
+ }
+
+ @Override
+ public void close() throws IOException {
+ m.close();
+ }
+}
diff --git a/src/test/java/net/openhft/chronicle/map/Builder.java b/src/test/java/net/openhft/chronicle/map/Builder.java
index 1799e16c3..ae60b1915 100644
--- a/src/test/java/net/openhft/chronicle/map/Builder.java
+++ b/src/test/java/net/openhft/chronicle/map/Builder.java
@@ -21,9 +21,7 @@ import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
-
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static net.openhft.chronicle.map.Replicators.tcp;
+import java.util.concurrent.TimeUnit;
/**
* @author Rob Austin.
@@ -70,10 +68,10 @@ public class Builder {
final int serverPort,
final InetSocketAddress... endpoints) throws IOException {
TcpReplicationConfig tcpConfig = TcpReplicationConfig.of(serverPort, endpoints)
- .heartBeatInterval(1L, SECONDS);
+ .heartBeatInterval(1L, TimeUnit.SECONDS);
return ChronicleMapBuilder.of(kClass, vClass)
.entries(20000L)
- .replicators(identifier, tcp( tcpConfig));
+ .replicators(identifier, 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 941179aef..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 ReplicatingChannel clusterB;
- private ReplicatingChannel 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 ReplicatingChannelBuilder((byte) 1, 1024).tcpReplication(tcpConfig)
+ channelProviderA = new ChannelProviderBuilder()
+ .replicators((byte) 1, tcpConfig)
.create();
map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(1000)
- .channel(clusterA.createChannel((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 ReplicatingChannelBuilder((byte) 2, 1024).tcpReplication(tcpConfig)
+ channelProviderB = new ChannelProviderBuilder().replicators((byte) 2, tcpConfig)
.create();
map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(1000)
- .channel(clusterB.createChannel((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)
- .channel(clusterB.createChannel((short) 2))
+ .channel(channelProviderB.createChannel((short) 2))
.create(getPersistenceFile());
map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(1000)
- .channel(clusterA.createChannel((short) 2))
+ .channel(channelProviderA.createChannel((short) 2))
.create(getPersistenceFile());
map2a.put(1, "EXAMPLE-2");
diff --git a/src/test/java/net/openhft/chronicle/map/Issue3.java b/src/test/java/net/openhft/chronicle/map/Issue3.java
new file mode 100644
index 000000000..faf02b435
--- /dev/null
+++ b/src/test/java/net/openhft/chronicle/map/Issue3.java
@@ -0,0 +1,28 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.chronicle.set.ChronicleSet;
+import net.openhft.chronicle.set.ChronicleSetBuilder;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Random;
+import java.util.Set;
+
+public class Issue3 {
+
+ @Test
+ public void test() throws IOException {
+ Set set = ChronicleSetBuilder.of(Long.class)
+ .actualSegments(1)
+ .actualEntriesPerSegment(1000)
+ .create();
+
+ Random r = new Random();
+ for (int i = 0; i < 10; i++) {
+ for (int j = 0; j < 300; j++) {
+ set.add(r.nextLong());
+ }
+ set.clear();
+ }
+ }
+}
diff --git a/src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java
similarity index 80%
rename from src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java
rename to src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java
index e9c102f34..f5292974c 100644
--- a/src/test/java/net/openhft/chronicle/map/PostClusterMapCreationTest.java
+++ b/src/test/java/net/openhft/chronicle/map/PostChannelCreationTest.java
@@ -35,7 +35,7 @@ import static org.junit.Assert.assertTrue;
*
* @author Rob Austin.
*/
-public class PostClusterMapCreationTest {
+public class PostChannelCreationTest {
private ChronicleMap map1a;
private ChronicleMap map2a;
@@ -43,11 +43,8 @@ public class PostClusterMapCreationTest {
private ChronicleMap map1b;
private ChronicleMap map2b;
- private ReplicatingChannel clusterB;
- private ReplicatingChannel clusterA;
-
- private ReplicatingChannelBuilder replicatingClusterBuilder1;
-
+ private ChannelProvider channelProviderA;
+ private ChannelProvider channelProviderB;
public static File getPersistenceFile() {
String TMP = System.getProperty("java.io.tmpdir");
@@ -62,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 ReplicatingChannelBuilder((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)
- .channel(clusterA.createChannel((short) 1))
+ .channel(channelProviderA.createChannel((short) 1))
.create(getPersistenceFile());
map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(1000)
- .channel(clusterA.createChannel((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 ReplicatingChannelBuilder((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)
- .channel(clusterB.createChannel((short) 1))
+ .channel(channelProviderB.createChannel((short) 1))
.create(getPersistenceFile());
map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(1000)
- .channel(clusterB.createChannel((short) 2))
+ .channel(channelProviderB.createChannel((short) 2))
.create(getPersistenceFile());
}
@@ -108,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/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java
new file mode 100644
index 000000000..f6f1a00df
--- /dev/null
+++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java
@@ -0,0 +1,240 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.io.ByteBufferBytes;
+import net.openhft.lang.io.Bytes;
+import net.openhft.lang.io.serialization.BytesMarshallable;
+import net.openhft.lang.model.constraints.NotNull;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.*;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+
+public class SerializerTest {
+
+ @Test
+ public void testValueMarshallable() throws Exception {
+ testReadWriteValue("Test");
+ testReadWriteValue(1);
+ testReadWriteValue(1L);
+ testReadWriteValue(1.0);
+ testReadWriteValue(1.0f);
+ testReadWriteValue(Collections.singleton("Test"));
+ testReadWriteValue(Collections.EMPTY_MAP);
+ testReadWriteValue(new MyTestClass(3));
+ testReadWriteValue(new MyTestClassMarshallable(3));
+
+ testReadWriteValue(new MyTestClassExternalizable(3));
+ testReadWriteValue(new MyTestClassObjectGraph(3));
+ }
+
+
+ @Test
+ public void testKeyMarshallable() throws Exception {
+ testReadWriteKey("Test");
+ testReadWriteKey(1);
+ testReadWriteKey(1L);
+ testReadWriteKey(1.0);
+ testReadWriteKey(1.0f);
+ testReadWriteKey(Collections.singleton("Test"));
+ testReadWriteKey(Collections.EMPTY_MAP);
+ testReadWriteKey(new MyTestClass(3));
+ testReadWriteKey(new MyTestClassMarshallable(3));
+
+ testReadWriteKey(new MyTestClassExternalizable(3));
+ testReadWriteKey(new MyTestClassObjectGraph(3));
+ }
+
+
+ public void testReadWriteValue(Object value) throws Exception {
+
+ Class valueClass = value.getClass();
+
+ final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocateDirect(1024));
+ ByteBufferBytes in = out.slice();
+
+ ChronicleMapBuilder builder = ChronicleMapBuilder.of(Integer.class, valueClass);
+ builder.preMapConstruction();
+
+ Serializer v = new Serializer(builder.valueBuilder);
+
+ v.writeMarshallable(value, out);
+
+ long position = out.position();
+ in.limit(position);
+
+
+ Object actual = v.readMarshallable(in);
+ Assert.assertEquals(actual, value);
+ }
+
+
+ public void testReadWriteKey(Object key) throws Exception {
+
+ Class clazz = key.getClass();
+
+ final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocateDirect(1024));
+ ByteBufferBytes in = out.slice();
+
+ ChronicleMapBuilder builder = ChronicleMapBuilder.of(clazz, Integer.class);
+ builder.preMapConstruction();
+ {
+ Serializer v = new Serializer(builder.keyBuilder);
+
+
+ v.writeMarshallable(key, out);
+
+ long position = out.position();
+ in.limit(position);
+
+ Object actual = v.readMarshallable(in);
+ Assert.assertEquals(actual, key);
+
+ }
+
+
+ }
+
+ public static class MyTestClassExternalizable implements Externalizable {
+ int a;
+
+ public MyTestClassExternalizable() {
+ }
+
+
+ MyTestClassExternalizable(int a) {
+ this.a = a;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ MyTestClassExternalizable that = (MyTestClassExternalizable) o;
+
+ if (a != that.a) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return a;
+ }
+
+ @Override
+ public void writeExternal(ObjectOutput out) throws IOException {
+ out.write(a);
+ }
+
+ @Override
+ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+ a = in.readInt();
+ }
+ }
+
+ public static class MyTestClass implements Serializable {
+ int a;
+
+ MyTestClass(int a) {
+ this.a = a;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ MyTestClass myTestClass = (MyTestClass) o;
+
+ if (a != myTestClass.a) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return a;
+ }
+ }
+
+ public static class MyTestClassMarshallable implements BytesMarshallable {
+ int a;
+
+ public MyTestClassMarshallable() {
+ }
+
+ public MyTestClassMarshallable(int a) {
+ this.a = a;
+ }
+
+ public void setA(int a) {
+ this.a = a;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ MyTestClassMarshallable that = (MyTestClassMarshallable) o;
+
+ if (a != that.a) return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return a;
+ }
+
+ public void readMarshallable(@NotNull Bytes in) throws IllegalStateException {
+ a = in.readInt();
+ }
+
+ /**
+ * write an object to bytes
+ *
+ * @param out to write to
+ */
+ public void writeMarshallable(@NotNull Bytes out) {
+ out.writeInt(a);
+ }
+ }
+
+ public static class MyTestClassObjectGraph implements Serializable {
+
+ MyTestClass delegate;
+
+ public MyTestClassObjectGraph() {
+
+ }
+
+ MyTestClassObjectGraph(int a) {
+ delegate = new MyTestClass(a);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ MyTestClassObjectGraph that = (MyTestClassObjectGraph) o;
+
+ if (delegate != null ? !delegate.equals(that.delegate) : that.delegate != null)
+ return false;
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return delegate != null ? delegate.hashCode() : 0;
+ }
+ }
+
+
+}
\ No newline at end of file
diff --git a/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java b/src/test/java/net/openhft/chronicle/map/SingleMapClusterTest.java
index af26c450f..99c41eefa 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 ReplicatingChannel clusterB;
- private ReplicatingChannel clusterA;
-
+ private ChannelProvider channelProviderA;
+ private ChannelProvider channelProviderB;
public static File getPersistenceFile() {
String TMP = System.getProperty("java.io.tmpdir");
@@ -60,13 +59,15 @@ public class SingleMapClusterTest {
final TcpReplicationConfig tcpReplicationConfig = TcpReplicationConfig
.of(8086, new InetSocketAddress("localhost", 8087))
.heartBeatInterval(1, SECONDS);
-
- clusterA = new ReplicatingChannelBuilder((byte) 1, 1024)
- .tcpReplication(tcpReplicationConfig).create();
+ int maxEntrySize = 1024;
+ byte identifier= 1;
+ channelProviderA = new ChannelProviderBuilder()
+ .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)
- .channel(clusterA.createChannel((short) 1))
+ .channel(channelProviderA.createChannel((short) 1))
.create(getPersistenceFile());
}
@@ -76,11 +77,12 @@ public class SingleMapClusterTest {
.of(8087, new InetSocketAddress("localhost", 8086))
.heartBeatInterval(1, SECONDS);
- clusterB = new ReplicatingChannelBuilder((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)
- .channel(clusterB.createChannel((short) 1))
+ .channel(channelProviderB.createChannel((short) 1))
.entries(1000)
.create(getPersistenceFile());
}
@@ -89,7 +91,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/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
new file mode 100644
index 000000000..da229954a
--- /dev/null
+++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
@@ -0,0 +1,49 @@
+package net.openhft.chronicle.map;
+
+import net.openhft.lang.io.ByteBufferBytes;
+import org.junit.Test;
+import org.testng.Assert;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static net.openhft.chronicle.map.Builder.getPersistenceFile;
+import static net.openhft.chronicle.map.TcpReplicationConfig.of;
+
+/**
+ * @author Rob Austin.
+ */
+public class StatelessClientTest {
+
+
+ @Test
+ public void test() throws IOException, InterruptedException {
+
+ ChronicleMapBuilder builder =
+ ChronicleMapBuilder.of(Integer.class, CharSequence.class)
+ .entries(20000L)
+ .replicators((byte) 2, of(8076).heartBeatInterval(1L, SECONDS));
+ {
+ final ChronicleMap map2a =
+ builder.create(getPersistenceFile());
+ map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2
+ }
+
+ KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder.keyBuilder, builder.valueBuilder);
+
+ final ByteBufferBytes out = new ByteBufferBytes(ByteBuffer.allocateDirect(1024));
+ // ByteBufferBytes in = out.slice();
+
+ InetSocketAddress remote = new InetSocketAddress("localhost", 8076);
+ StatelessMapClient map = new StatelessMapClient(keyValueSerializer, remote);
+
+
+ Assert.assertEquals("EXAMPLE-10", map.get(10));
+ Assert.assertEquals("EXAMPLE-10", map.get(10));
+
+ Assert.assertEquals(1, map.size());
+ }
+}
+
diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplicationTest3wayPutReturnsNull.java
index d4bd0a076..e8925e4fb 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)
- .replicators(identifier, Replicators.tcp(tcpConfig))
+ .replicators(identifier, tcpConfig)
.create(Builder.getPersistenceFile());
}
diff --git a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java
index 59199827d..fd418c0c5 100644
--- a/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java
+++ b/src/test/java/net/openhft/chronicle/map/TwoMapOnDifferentServers.java
@@ -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.TcpReplicationConfig.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 TcpReplicationConfig 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);
}
diff --git a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java
index 8bec68d2c..8a2550b74 100644
--- a/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java
+++ b/src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java
@@ -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
*
@@ -44,7 +42,7 @@ public class UDPSocketReplicationTest {
return ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(1000)
- .replicators((byte) identifier, udp(udpConfig))
+ .replicators((byte) identifier, udpConfig)
.create(Builder.getPersistenceFile());
}
diff --git a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java
index 0212db7f6..057f70161 100644
--- a/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java
+++ b/src/test/java/net/openhft/chronicle/set/SetForTCPSocketReplicationTest3way.java
@@ -16,7 +16,6 @@
package net.openhft.chronicle.set;
-import net.openhft.chronicle.map.ChronicleSet;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -27,7 +26,6 @@ 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 org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -51,8 +49,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());
}