diff --git a/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java b/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java index 209b0110a..517009267 100644 --- a/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java +++ b/src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java @@ -357,10 +357,10 @@ public interface ChronicleHashBuilder{@code * ChronicleMap map = ChronicleMapBuilder * .of(Key.class, Value.class) @@ -71,17 +68,18 @@ import static net.openhft.chronicle.map.Objects.builderEquals; * documentation. * *

Later in this documentation, "ChronicleMap" means "ChronicleMaps, created by {@code - * ChronicleMapBuilder}", unless specified different, because theoretically someone might provide {@code - * ChronicleMap} implementations with completely different properties. + * ChronicleMapBuilder}", unless specified different, because theoretically someone might provide + * {@code ChronicleMap} implementations with completely different properties. * - *

{@code ChronicleMap} ("ChronicleMaps, created by {@code ChronicleMapBuilder}") currently doesn't support - * resizing. That is why you should always configure {@linkplain #entries(long) number of entries} you - * are going to insert into the created map at most. See {@link #entries(long)} method documentation - * for more information on this. + *

{@code ChronicleMap} ("ChronicleMaps, created by {@code ChronicleMapBuilder}") currently + * doesn't support resizing. That is why you should always configure {@linkplain + * #entries(long) number of entries} you are going to insert into the created map at most. + * See {@link #entries(long)} method documentation for more information on this. * *

{@code ChronicleMap} allocates memory by equally sized chunks. This size is called {@linkplain * #entrySize(int) entry size}, you are strongly recommended to configure it to achieve least memory - * consumption and best speed. See {@link #entrySize(int)} method documentation for more information on this. + * consumption and best speed. See {@link #entrySize(int)} method documentation for more information + * on this. * * @param key type of the maps, produced by this builder * @param value type of the maps, produced by this builder @@ -130,7 +128,8 @@ public class ChronicleMapBuilder implements Cloneable, private V defaultValue = null; private DefaultValueProvider defaultValueProvider = null; private byte identifier = -1; - private boolean isStateless; + + private StatelessBuilder statelessBuilder; ChronicleMapBuilder(Class keyClass, Class valueClass) { keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); @@ -191,9 +190,9 @@ public class ChronicleMapBuilder implements Cloneable, * .keySize(10) * // shouldn't specify valueSize(), because it is statically known * .create();} - * (Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded (and each - * character takes 2 bytes on-heap), because default off-heap {@link String} encoding is UTF-8 in {@code - * ChronicleMap}.) + * (Note that 10 is chosen as key size in bytes despite strings in Java are UTF-16 encoded (and + * each character takes 2 bytes on-heap), because default off-heap {@link String} encoding is + * UTF-8 in {@code ChronicleMap}.) * * @see #constantKeySizeBySample(Object) * @see #valueSize(int) @@ -230,16 +229,17 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Configures the optimal number of bytes, taken by serialized form of values, put into maps, created by - * this builder. If value size is always the same, call {@link #constantValueSizeBySample(Object)} method - * instead of this one. + * Configures the optimal number of bytes, taken by serialized form of values, put into maps, + * created by this builder. If value size is always the same, call {@link + * #constantValueSizeBySample(Object)} method instead of this one. * - *

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

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

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

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

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

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

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

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

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

If entry size is not configured explicitly by calling this method, it is computed based on - * {@linkplain #metaDataBytes(int) meta data bytes}, plus {@linkplain #keySize(int) key size}, plus - * {@linkplain #valueSize(int) value size}, plus a few bytes required by implementations, with respect to - * {@linkplain #entryAndValueAlignment(Alignment) alignment}. + * {@linkplain #metaDataBytes(int) meta data bytes}, plus {@linkplain #keySize(int) key size}, + * plus {@linkplain #valueSize(int) value size}, plus a few bytes required by implementations, + * with respect to {@linkplain #entryAndValueAlignment(Alignment) alignment}. * *

Note that the actual entrySize will be aligned to 4 (default {@linkplain - * #entryAndValueAlignment(Alignment) entry alignment}). I. e. if you set entry size to 30, the actual - * entry size will be 32 (30 aligned to 4 bytes). If you don't want entry size to be aligned, set {@code - * entryAndValueAlignment(Alignment.NO_ALIGNMENT)}. + * #entryAndValueAlignment(Alignment) entry alignment}). I. e. if you set entry size to 30, the + * actual entry size will be 32 (30 aligned to 4 bytes). If you don't want entry size to be + * aligned, set {@code entryAndValueAlignment(Alignment.NO_ALIGNMENT)}. * * @see #entryAndValueAlignment(Alignment) * @see #entries(long) @@ -344,16 +344,16 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Configures alignment strategy of address in memory of entries and independently of address in memory of - * values within entries in ChronicleMaps, created by this builder. + * Configures alignment strategy of address in memory of entries and independently of address in + * memory of values within entries in ChronicleMaps, created by this builder. * - *

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

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

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

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

Default is {@link Alignment#OF_4_BYTES}. * @@ -437,20 +437,21 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Configures if the maps created by this builder should return {@code null} instead of previous mapped - * values on {@link ChronicleMap#put(Object, Object) ChornicleMap.put(key, value)} calls. + * Configures if the maps created by this builder should return {@code null} instead of previous + * mapped values on {@link ChronicleMap#put(Object, Object) ChornicleMap.put(key, value)} + * calls. * - *

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

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

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

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

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

{@link Map#remove(Object) Map.remove()} returns the previous value, functionality which is + * rarely used but fairly cheap for {@link HashMap}. In the case, for an off heap collection, it + * has to create a new object and deserialize the data from off-heap memory. It's expensive for + * something you probably don't use. * * @param removeReturnsNull {@code true} if you want {@link ChronicleMap#remove(Object) - * ChronicleMap.remove()} to not return the value of the removed entry but - * instead return {@code null} + * ChronicleMap.remove()} to not return the value of the removed entry + * but instead return {@code null} * @return an instance of the map builder back * @see #putReturnsNull(boolean) */ @@ -637,9 +638,9 @@ public class ChronicleMapBuilder implements Cloneable, * .create();} * *

This serializer is used to serialize both keys and values, if they both require this: - * loosely typed, nullable, and custom {@linkplain #keyMarshaller(BytesMarshaller) key} - * and {@linkplain #valueMarshallerAndFactory(BytesMarshaller, ObjectFactory) value} marshallers - * are not configured. + * loosely typed, nullable, and custom {@linkplain #keyMarshaller(BytesMarshaller) key} and + * {@linkplain #valueMarshallerAndFactory(BytesMarshaller, ObjectFactory) value} marshallers are + * not configured. */ @Override public ChronicleMapBuilder objectSerializer(ObjectSerializer objectSerializer) { @@ -665,6 +666,12 @@ public class ChronicleMapBuilder implements Cloneable, return this; } + @Override + public ChronicleMapBuilder stateless(StatelessBuilder statelessBuilder) { + this.statelessBuilder = statelessBuilder; + return this; + } + public ChronicleMapBuilder valueMarshallerAndFactory( @NotNull BytesMarshaller valueMarshaller, @NotNull ObjectFactory valueFactory) { valueBuilder.marshaller(valueMarshaller, valueFactory); @@ -694,18 +701,18 @@ public class ChronicleMapBuilder implements Cloneable, } /** - * Specifies the value to be put for each key queried in {@link ChronicleMap#get get()} and {@link - * ChronicleMap#getUsing(Object, Object) getUsing()} methods, if the key is absent in the map. Then this - * default value is returned from query method. + * Specifies the value to be put for each key queried in {@link ChronicleMap#get get()} and + * {@link ChronicleMap#getUsing(Object, Object) getUsing()} methods, if the key is absent in the + * map. Then this default value is returned from query method. * - *

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

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

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

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

This configuration overrides any previous {@link #defaultValue(Object)} configuration to + * this {@code ChronicleMapBuilder}. * * @param defaultValueProvider the strategy to obtain a default value by the absent key * @return this builder object back @@ -733,13 +740,6 @@ public class ChronicleMapBuilder implements Cloneable, } - - @Override - public ChronicleMapBuilder stateless(boolean isStateless) { - this.isStateless = isStateless; - return this; - } - /** * Non-public because should be called only after {@link #preMapConstruction()} */ @@ -806,8 +806,14 @@ public class ChronicleMapBuilder implements Cloneable, return establishReplication(map); } + @Override public ChronicleMap create() throws IOException { + + if (statelessBuilder != null) { + return createStatelessMap(statelessBuilder); + } + VanillaChronicleMap map = newMap(); BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE, map.sizeInBytes(), true); @@ -815,6 +821,15 @@ public class ChronicleMapBuilder implements Cloneable, return establishReplication(map); } + private ChronicleMap createStatelessMap(StatelessBuilder statelessBuilder) throws IOException { + preMapConstruction(); + + KeyValueSerializer keyValueSerializer = new KeyValueSerializer(this.keyBuilder, + this.valueBuilder); + + return new StatelessMapClient(keyValueSerializer, + statelessBuilder); + } private VanillaChronicleMap newMap() throws IOException { preMapConstruction(); @@ -831,7 +846,7 @@ public class ChronicleMapBuilder implements Cloneable, return identifier != -1; } - void preMapConstruction() { + void preMapConstruction() { keyBuilder.objectSerializer(objectSerializer()); valueBuilder.objectSerializer(objectSerializer()); diff --git a/src/main/java/net/openhft/chronicle/map/ConcurrentExpiryMap.java b/src/main/java/net/openhft/chronicle/map/ConcurrentExpiryMap.java deleted file mode 100644 index 426bc94ba..000000000 --- a/src/main/java/net/openhft/chronicle/map/ConcurrentExpiryMap.java +++ /dev/null @@ -1,14 +0,0 @@ -package net.openhft.chronicle.map; - -import java.util.Map; -import java.util.Queue; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; -import java.util.concurrent.ConcurrentMap; - -/** - * wraps a concurrent map and queue so that we can easily expire old entries - * - * @author Rob Austin. - */ - diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java index b10c35541..7fa9449b8 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java @@ -1525,4 +1525,74 @@ class ReplicatedChronicleMap, } +abstract class Replicator { + /** + * A constructor for use in subclasses. + */ + protected Replicator() { + } + + /** + * Applies the replicator to the map instance and returns a Closeable token to manage resources, + * associated with the replication. + *

+ *

This method isn't intended to be called from the client code. + * + * @param builder the builder from which the map was constructed. The replicator may obtain some + * 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 + * @throws IllegalStateException if this replicator doesn't support application to more than + * one map (or the specified number of maps), and this replicator has already been + * applied to a map (or the specified number of maps) + */ + protected abstract Closeable applyTo(ChronicleMapBuilder builder, + Replica map, Replica.EntryExternalizable entryExternalizable, final ChronicleMap chronicleMap) + throws IOException; +} + + +final class Replicators { + + private Replicators() { + } + + static Replicator tcp(final TcpReplicationConfig replicationConfig) { + return new Replicator() { + + @Override + protected Closeable applyTo(ChronicleMapBuilder builder, + Replica replica, Replica.EntryExternalizable entryExternalizable, + final ChronicleMap chronicleMap) throws IOException { + + final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder + .keyBuilder, builder.valueBuilder); + + StatelessServerConnector statelessServer = new StatelessServerConnector + (keyValueSerializer, chronicleMap, builder.entrySize()); + + return new TcpReplicator(replica, entryExternalizable, replicationConfig, + builder.entrySize(), statelessServer); + } + }; + } + + static Replicator udp( + final UdpReplicationConfig replicationConfig) { + return new Replicator() { + + @Override + protected Closeable applyTo(ChronicleMapBuilder builder, + 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/Replicator.java b/src/main/java/net/openhft/chronicle/map/Replicator.java deleted file mode 100644 index 3dede414d..000000000 --- a/src/main/java/net/openhft/chronicle/map/Replicator.java +++ /dev/null @@ -1,55 +0,0 @@ -/* - * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package net.openhft.chronicle.map; - -import java.io.Closeable; -import java.io.IOException; - -/** - * @see Replicators - * @see ChronicleMapBuilder#replicator(byte, Replicator) - */ -public abstract class Replicator { - - /** - * A constructor for use in subclasses. - */ - protected Replicator() { - } - - /** - * Applies the replicator to the map instance and returns a Closeable token to manage resources, - * associated with the replication. - *

- *

This method isn't intended to be called from the client code. - * - * @param builder the builder from which the map was constructed. The replicator may obtain some - * 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 - * @throws IllegalStateException if this replicator doesn't support application to more than - * one map (or the specified number of maps), and this replicator has already been - * applied to a map (or the specified number of maps) - */ - protected abstract Closeable applyTo(ChronicleMapBuilder builder, - 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 deleted file mode 100644 index aeb5118d3..000000000 --- a/src/main/java/net/openhft/chronicle/map/Replicators.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Copyright 2014 Higher Frequency Trading http://www.higherfrequencytrading.com - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package net.openhft.chronicle.map; - - -import java.io.Closeable; -import java.io.IOException; - -final class Replicators { - - private Replicators() { - } - - static Replicator tcp(final TcpReplicationConfig replicationConfig) { - return new Replicator() { - - @Override - protected Closeable applyTo(ChronicleMapBuilder builder, - Replica replica, Replica.EntryExternalizable entryExternalizable, - final ChronicleMap chronicleMap) - throws IOException { - - final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder - .keyBuilder, builder.valueBuilder); - - StatelessServerConnector statelessServerConnector = new StatelessServerConnector - (keyValueSerializer, chronicleMap); - - return new TcpReplicator(replica, entryExternalizable, replicationConfig, - builder.entrySize(), statelessServerConnector); - } - }; - } - - static Replicator udp( - final UdpReplicationConfig replicationConfig) { - return new Replicator() { - - @Override - protected Closeable applyTo(ChronicleMapBuilder builder, - 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/StatelessMapClient.java b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java index c986ee1ce..03aefd365 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java @@ -1,5 +1,6 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.StatelessBuilder; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; @@ -9,12 +10,9 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; 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.*; @@ -28,10 +26,10 @@ class StatelessMapClient implements ChronicleMap, Closeable { 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}); private final CloseablesManager closeables = new CloseablesManager(); + private final StatelessBuilder builder; public static enum EventId { HEARTBEAT, @@ -52,22 +50,23 @@ class StatelessMapClient implements ChronicleMap, Closeable { REMOVE_WITH_VALUE } - private final ThreadLocal sourceBuffer = new ThreadLocal(); - private final Map transactionIDs = new HashMap(); + // private final ThreadLocal sourceBuffer = new ThreadLocal(); + private final AtomicLong transactionID = new AtomicLong(); - private long timeoutMs = TimeUnit.SECONDS.toMillis(20); + // private long timeoutMs = TimeUnit.SECONDS.toMillis(20); - public StatelessMapClient(final KeyValueSerializer keyValueSerializer, final InetSocketAddress remote) throws IOException { + public StatelessMapClient(final KeyValueSerializer keyValueSerializer, + final StatelessBuilder builder) throws IOException { this.keyValueSerializer = keyValueSerializer; + this.builder = builder; try { clientChannel = AbstractChannelReplicator.openSocketChannel(closeables); - clientChannel.connect(remote); + clientChannel.connect(builder.remoteAddress()); doHandShaking(); } catch (IOException e) { @@ -97,7 +96,6 @@ class StatelessMapClient implements ChronicleMap, Closeable { } - public File file() { return null; } @@ -117,12 +115,10 @@ class StatelessMapClient implements ChronicleMap, Closeable { 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) { long sizeLocation = writeEvent(PUT_IF_ABSENT); @@ -138,7 +134,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { writeValue((V) value); // get the data back from the server - return blockingFetch(sizeLocation, 20000).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } @@ -150,7 +146,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { writeValue(newValue); // get the data back from the server - return blockingFetch(sizeLocation, 20000).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } @@ -169,7 +165,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { long sizeLocation = writeEvent(SIZE); // get the data back from the server - return blockingFetch(sizeLocation, 20000).readInt(); + return blockingFetch(sizeLocation).readInt(); } @@ -177,7 +173,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { long sizeLocation = writeEvent(IS_EMPTY); // get the data back from the server - return blockingFetch(sizeLocation, 20000).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } @@ -186,7 +182,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { writeKey((K) key); // get the data back from the server - return blockingFetch(sizeLocation, 20000).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } @@ -196,14 +192,14 @@ class StatelessMapClient implements ChronicleMap, Closeable { writeValue((V) value); // get the data back from the server - return blockingFetch(sizeLocation, 20000).readBoolean(); + return blockingFetch(sizeLocation).readBoolean(); } public synchronized long longSize() { long sizeLocation = writeEvent(LONG_SIZE); // get the data back from the server - return blockingFetch(sizeLocation, 20000).readLong(); + return blockingFetch(sizeLocation).readLong(); } @@ -278,7 +274,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { long sizeLocation = writeEvent(CLEAR); // get the data back from the server - blockingFetch(sizeLocation, 20000); + blockingFetch(sizeLocation); } @@ -288,7 +284,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { long sizeLocation = writeEvent(KEY_SET); // get the data back from the server - return readKeySet(blockingFetch(sizeLocation, 20000)); + return readKeySet(blockingFetch(sizeLocation)); } @@ -298,7 +294,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { long sizeLocation = writeEvent(VALUES); // get the data back from the server - final Bytes in = blockingFetch(sizeLocation, 20000); + final Bytes in = blockingFetch(sizeLocation); final long size = in.readStopBit(); @@ -320,7 +316,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { long sizeLocation = writeEvent(ENTRY_SET); // get the data back from the server - Bytes in = blockingFetch(sizeLocation, 20000); + Bytes in = blockingFetch(sizeLocation); long size = in.readStopBit(); @@ -345,62 +341,6 @@ class StatelessMapClient implements ChronicleMap, Closeable { 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; @@ -455,9 +395,9 @@ class StatelessMapClient implements ChronicleMap, Closeable { } } - private Bytes blockingFetch(long sizeLocation, long timeOutMs) { + private Bytes blockingFetch(long sizeLocation) { try { - return blockingFetchThrowable(sizeLocation, timeOutMs); + return blockingFetchThrowable(sizeLocation, this.builder.timeoutMs()); } catch (IOException e) { closeables.closeQuietly(); throw new RuntimeIOException(e); @@ -475,7 +415,9 @@ class StatelessMapClient implements ChronicleMap, Closeable { long transactionId = nextUniqueTransaction(startTime); long timeoutTime = startTime + timeOutMs; - LOG.info("sending data with transactionId=" + transactionId); + if (LOG.isDebugEnabled()) + LOG.debug("sending data with transactionId=" + transactionId); + out.writeLong(transactionId); writeSizeAt(sizeLocation); @@ -525,7 +467,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { private void checkTimeout(long timeoutTime) { if (timeoutTime < System.currentTimeMillis()) - throw new RuntimeTimeOutException(); + throw new TimeoutException(); } private void writeSizeAt(long locationOfSize) { @@ -544,7 +486,7 @@ class StatelessMapClient implements ChronicleMap, Closeable { } private V readKey(final long sizeLocation) { - return keyValueSerializer.readValue(blockingFetch(sizeLocation, 20000)); + return keyValueSerializer.readValue(blockingFetch(sizeLocation)); } private void writeValue(V value) { @@ -580,9 +522,9 @@ class RuntimeIOException extends RuntimeException { } -class RuntimeTimeOutException extends RuntimeException { +class TimeoutException extends RuntimeException { - public RuntimeTimeOutException() { + public TimeoutException() { } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java deleted file mode 100644 index dc5d24dfb..000000000 --- a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java +++ /dev/null @@ -1,368 +0,0 @@ -package net.openhft.chronicle.map; - -import net.openhft.lang.io.Bytes; -import net.openhft.lang.model.constraints.NotNull; - -import java.util.*; - -/** - * @author Rob Austin. - */ -class StatelessServerConnector { - - - private final KeyValueSerializer keyValueSerializer; - private final ChronicleMap map; - private double maxEntrySizeBytes; - - - public StatelessServerConnector(KeyValueSerializer keyValueSerializer, ChronicleMap map) { - this.keyValueSerializer = keyValueSerializer; - this.map = map; - } - - public Work processStatelessEvent(byte eventId, @NotNull Bytes in, @NotNull Bytes out) { - - final StatelessMapClient.EventId event = StatelessMapClient.EventId.values()[eventId]; - - switch (event) { - - case LONG_SIZE: - return longSize(in, out); - - case IS_EMPTY: - return isEmpty(in, out); - - case CONTAINS_KEY: - return containsKey(in, out); - - case CONTAINS_VALUE: - return containsValue(in, out); - - case GET: - return get(in, out); - - case PUT: - return put(in, out); - - case REMOVE: - return remove(in, out); - - case CLEAR: - return clear(in, out); - - case KEY_SET: - return keySet(in, out); - - case VALUES: - return values(in, out); - - case ENTRY_SET: - return entrySet(in, out); - - case REPLACE: - return replace(in, out); - - case REPLACE_WITH_OLD_AND_NEW_VALUE: - return replaceWithOldAndNew(in, out); - - case PUT_IF_ABSENT: - return putIfAbsent(in, out); - - case REMOVE_WITH_VALUE: - return removeWithValue(in, out); - - case SIZE: - return size(in, out); - - default: - throw new IllegalStateException("unsupported event=" + event); - - } - - } - - private Work removeWithValue(Bytes in, Bytes out) { - boolean result = map.remove(readKey(in), readValue(in)); - out.writeBoolean(result); - return null; - } - - private Work replaceWithOldAndNew(Bytes in, Bytes out) { - - final K key = readKey(in); - V oldValue = readValue(in); - V newValue = readValue(in); - - long sizeLocation = reflectTransactionId(in, out); - try { - map.replace(key, oldValue, newValue); - } catch (RuntimeException e) { - writeException(e, out); - writeSizeAndFlags(sizeLocation, true, out); - return null; - } - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - - public Work longSize(Bytes in, Bytes out) { - long sizeLocation = reflectTransactionId(in, out); - out.writeLong(map.longSize()); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work size(Bytes in, Bytes out) { - long sizeLocation = reflectTransactionId(in, out); - out.writeInt(map.size()); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work isEmpty(Bytes in, Bytes out) { - long sizeLocation = reflectTransactionId(in, out); - out.writeBoolean(map.isEmpty()); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work containsKey(Bytes in, Bytes out) { - K k = readKey(in); - long sizeLocation = reflectTransactionId(in, out); - out.writeBoolean(map.containsKey(k)); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work containsValue(Bytes in, Bytes out) { - V v = readValue(in); - long sizeLocation = reflectTransactionId(in, out); - out.writeBoolean(map.containsValue(v)); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work get(Bytes in, Bytes out) { - K k = readKey(in); - long sizeLocation = reflectTransactionId(in, out); - try { - writeValue(map.get(k), out); - } catch (RuntimeException e) { - writeException(e, out); - writeSizeAndFlags(sizeLocation, true, out); - return null; - } - writeSizeAndFlags(sizeLocation, false, out); - return null; - - } - - - public Work put(Bytes in, Bytes out) { - K k = readKey(in); - V v = readValue(in); - long sizeLocation = reflectTransactionId(in, out); - writeValue(map.put(k, v), out); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work remove(Bytes in, Bytes out) { - final V value = map.remove(readKey(in)); - long sizeLocation = reflectTransactionId(in, out); - writeValue(value, out); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work putAll(Bytes in, Bytes out) { - map.putAll(readEntries(in)); - long sizeLocation = reflectTransactionId(in, out); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - - public Work clear(Bytes in, Bytes out) { - map.clear(); - long sizeLocation = reflectTransactionId(in, out); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work keySet(Bytes in, final Bytes out) { - final long sizeLocation = reflectTransactionId(in, out); - - final Set ks = map.keySet(); - out.writeStopBit(ks.size()); - - final Iterator iterator = ks.iterator(); - - // this allows us to write more data than the buffer will allow - return new Work() { - - @Override - public boolean doWork(Bytes out) { - - while (iterator.hasNext()) { - - // we've filled up the buffer, so lets give another channel a chance to send - // some data, we don't know the max key size, we will use the entrySize instead - if (out.remaining() <= maxEntrySizeBytes) - return false; - - writeKey(iterator.next(), out); - } - - writeSizeAndFlags(sizeLocation, false, out); - return true; - } - }; - } - - public Work 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); - } - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - public Work 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); - } - return null; - } - - public Work putIfAbsent(Bytes in, Bytes out) { - K key = readKey(in); - V v = readValue(in); - long sizeLocation = reflectTransactionId(in, out); - writeValue(map.putIfAbsent(key, v), out); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - - public Work replace(Bytes in, Bytes out) { - K k = readKey(in); - V v = readValue(in); - - long sizeLocation = reflectTransactionId(in, out); - map.replace(k, v); - writeSizeAndFlags(sizeLocation, false, out); - return null; - } - - /** - * 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(3); - 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 writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { - long size = out.position() - locationOfSize; - out.writeUnsignedShort(0L, (int) size); - out.writeBoolean(2L, isException); - } - - private void writeException(RuntimeException e, Bytes out) { - - long start = out.position(); - out.skip(2); - out.writeObject(e); - long len = out.position() - (start + 2L); - out.writeUnsignedShort(start, (int) len); - } - - - 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; - } -} - -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/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index ca5ed10d7..809955418 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -27,8 +27,7 @@ import java.io.IOException; import java.net.*; import java.nio.ByteBuffer; import java.nio.channels.*; -import java.util.BitSet; -import java.util.Set; +import java.util.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -1188,3 +1187,367 @@ interface Work { } +/** + * @author Rob Austin. + */ +class StatelessServerConnector { + + + private final KeyValueSerializer keyValueSerializer; + private final ChronicleMap map; + private double maxEntrySizeBytes; + + StatelessServerConnector(KeyValueSerializer keyValueSerializer, ChronicleMap map, int maxEntrySizeBytes) { + this.keyValueSerializer = keyValueSerializer; + this.map = map; + this.maxEntrySizeBytes = maxEntrySizeBytes; + } + + Work processStatelessEvent(byte eventId, @net.openhft.lang.model.constraints.NotNull Bytes in, @net.openhft.lang.model.constraints.NotNull Bytes out) { + + final StatelessMapClient.EventId event = StatelessMapClient.EventId.values()[eventId]; + + switch (event) { + + case LONG_SIZE: + return longSize(in, out); + + case IS_EMPTY: + return isEmpty(in, out); + + case CONTAINS_KEY: + return containsKey(in, out); + + case CONTAINS_VALUE: + return containsValue(in, out); + + case GET: + return get(in, out); + + case PUT: + return put(in, out); + + case REMOVE: + return remove(in, out); + + case CLEAR: + return clear(in, out); + + case KEY_SET: + return keySet(in, out); + + case VALUES: + return values(in, out); + + case ENTRY_SET: + return entrySet(in, out); + + case REPLACE: + return replace(in, out); + + case REPLACE_WITH_OLD_AND_NEW_VALUE: + return replaceWithOldAndNew(in, out); + + case PUT_IF_ABSENT: + return putIfAbsent(in, out); + + case REMOVE_WITH_VALUE: + return removeWithValue(in, out); + + case SIZE: + return size(in, out); + + default: + throw new IllegalStateException("unsupported event=" + event); + + } + + } + + private Work removeWithValue(Bytes in, Bytes out) { + boolean result = map.remove(readKey(in), readValue(in)); + out.writeBoolean(result); + return null; + } + + private Work replaceWithOldAndNew(Bytes in, Bytes out) { + + final K key = readKey(in); + V oldValue = readValue(in); + V newValue = readValue(in); + + long sizeLocation = reflectTransactionId(in, out); + try { + map.replace(key, oldValue, newValue); + } catch (RuntimeException e) { + writeException(e, out); + writeSizeAndFlags(sizeLocation, true, out); + return null; + } + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + + public Work longSize(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + out.writeLong(map.longSize()); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work size(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + out.writeInt(map.size()); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work isEmpty(Bytes in, Bytes out) { + long sizeLocation = reflectTransactionId(in, out); + out.writeBoolean(map.isEmpty()); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work containsKey(Bytes in, Bytes out) { + K k = readKey(in); + long sizeLocation = reflectTransactionId(in, out); + out.writeBoolean(map.containsKey(k)); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work containsValue(Bytes in, Bytes out) { + V v = readValue(in); + long sizeLocation = reflectTransactionId(in, out); + out.writeBoolean(map.containsValue(v)); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work get(Bytes in, Bytes out) { + K k = readKey(in); + long sizeLocation = reflectTransactionId(in, out); + try { + writeValue(map.get(k), out); + } catch (RuntimeException e) { + writeException(e, out); + writeSizeAndFlags(sizeLocation, true, out); + return null; + } + writeSizeAndFlags(sizeLocation, false, out); + return null; + + } + + + public Work put(Bytes in, Bytes out) { + K k = readKey(in); + V v = readValue(in); + long sizeLocation = reflectTransactionId(in, out); + writeValue(map.put(k, v), out); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work remove(Bytes in, Bytes out) { + final V value = map.remove(readKey(in)); + long sizeLocation = reflectTransactionId(in, out); + writeValue(value, out); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work putAll(Bytes in, Bytes out) { + map.putAll(readEntries(in)); + long sizeLocation = reflectTransactionId(in, out); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + + public Work clear(Bytes in, Bytes out) { + map.clear(); + long sizeLocation = reflectTransactionId(in, out); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work keySet(Bytes in, final Bytes out) { + final long sizeLocation = reflectTransactionId(in, out); + + final Set ks = map.keySet(); + out.writeStopBit(ks.size()); + + final Iterator iterator = ks.iterator(); + + // this allows us to write more data than the buffer will allow + return new Work() { + + @Override + public boolean doWork(Bytes out) { + + while (iterator.hasNext()) { + + // we've filled up the buffer, so lets give another channel a chance to send + // some data, we don't know the max key size, we will use the entrySize instead + if (out.remaining() <= maxEntrySizeBytes) + return false; + + writeKey(iterator.next(), out); + } + + writeSizeAndFlags(sizeLocation, false, out); + return true; + } + }; + } + + public Work 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); + } + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + public Work 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); + } + return null; + } + + public Work putIfAbsent(Bytes in, Bytes out) { + K key = readKey(in); + V v = readValue(in); + long sizeLocation = reflectTransactionId(in, out); + writeValue(map.putIfAbsent(key, v), out); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + + public Work replace(Bytes in, Bytes out) { + K k = readKey(in); + V v = readValue(in); + + long sizeLocation = reflectTransactionId(in, out); + map.replace(k, v); + writeSizeAndFlags(sizeLocation, false, out); + return null; + } + + /** + * 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(3); + 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 writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { + long size = out.position() - locationOfSize; + out.writeUnsignedShort(0L, (int) size); + out.writeBoolean(2L, isException); + } + + private void writeException(RuntimeException e, Bytes out) { + + long start = out.position(); + out.skip(2); + out.writeObject(e); + long len = out.position() - (start + 2L); + out.writeUnsignedShort(start, (int) len); + } + + + 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; + } +} + +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/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index 1988dd27a..3f2cef16f 100644 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.set; import net.openhft.chronicle.ChronicleHashBuilder; import net.openhft.chronicle.ChronicleHashErrorListener; +import net.openhft.chronicle.StatelessBuilder; import net.openhft.chronicle.TimeProvider; import net.openhft.chronicle.map.*; import net.openhft.lang.io.serialization.BytesMarshaller; @@ -254,8 +255,8 @@ public class ChronicleSetBuilder } @Override - public ChronicleSetBuilder stateless(boolean isStateless) { - chronicleMapBuilder.stateless(isStateless); + public ChronicleSetBuilder stateless(StatelessBuilder statelessBuilder) { + chronicleMapBuilder.stateless(statelessBuilder); return this; } diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 52e386d19..402fb0322 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -1,6 +1,5 @@ package net.openhft.chronicle.map; -import org.junit.Ignore; import org.junit.Test; import org.testng.Assert; @@ -8,6 +7,7 @@ import java.io.IOException; import java.net.InetSocketAddress; import static java.util.concurrent.TimeUnit.SECONDS; +import static net.openhft.chronicle.StatelessBuilder.remoteAddress; import static net.openhft.chronicle.map.Builder.getPersistenceFile; import static net.openhft.chronicle.map.TcpReplicationConfig.of; @@ -18,26 +18,38 @@ public class StatelessClientTest { @Test - @Ignore 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 serverMap; + final ChronicleMap statelessMap; + + // server { - final ChronicleMap map2a = - builder.create(getPersistenceFile()); - map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2 + + serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .entries(20000L) + .replicators((byte) 2, of(8076).heartBeatInterval(1L, SECONDS)) + .create(getPersistenceFile()); + + serverMap.put(10, "EXAMPLE-10"); } - KeyValueSerializer keyValueSerializer = new KeyValueSerializer(builder.keyBuilder, builder.valueBuilder); - InetSocketAddress remote = new InetSocketAddress("localhost", 8076); - StatelessMapClient map = new StatelessMapClient(keyValueSerializer, remote); + // stateless client + { + statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .stateless(remoteAddress(new InetSocketAddress("localhost", 8076))).create(); + + Assert.assertEquals("EXAMPLE-10", statelessMap.get(10)); + Assert.assertEquals(1, statelessMap.size()); + } + + serverMap.close(); + statelessMap.close(); - Assert.assertEquals("EXAMPLE-10", map.get(10)); - Assert.assertEquals(1, map.size()); } + + }