Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-167 now using the builder to create the Stateless client

Browse Source
This commit is contained in:
Rob Austin
2014-10-20 16:30:08 +01:00
parent c047e59d56
commit 1b40eb8147
12 changed files with 634 additions and 696 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/ChronicleHashBuilder.java
+2 -2
View File
@@ -357,10 +357,10 @@ public interface ChronicleHashBuilder<K, C extends ChronicleHash,
* consider to be a ClientProxy to {@link ChronicleMap} or {@link ChronicleSet} running on
* another host
*
* @param isStateless true if you want your container to hold no data
* @param statelessBuilder true if you want your container to hold no data
* @return this builder back
*/
B stateless(boolean isStateless);
B stateless(StatelessBuilder statelessBuilder);
/**
* Opens a hash container residing the specified file, or creates a new one if the file not yet
src/main/java/net/openhft/chronicle/StatelessBuilder.java
+34
View File
@@ -0,0 +1,34 @@
package net.openhft.chronicle;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;
/**
* @author Rob Austin.
*/
public class StatelessBuilder {
private long timeoutMs = TimeUnit.SECONDS.toMillis(5);
private InetSocketAddress remoteAddress;
public long timeoutMs() {
return timeoutMs;
}
public void timeout(long timeout, TimeUnit units) {
this.timeoutMs = units.toMillis(timeout);
}
private StatelessBuilder(InetSocketAddress remoteAddress) {
this.remoteAddress = remoteAddress;
}
public static StatelessBuilder remoteAddress(InetSocketAddress inetSocketAddress) {
return new StatelessBuilder(inetSocketAddress);
}
public InetSocketAddress remoteAddress() {
return remoteAddress;
}
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+105 -90
View File
@@ -16,10 +16,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.ChronicleHashBuilder;
import net.openhft.chronicle.ChronicleHashErrorListener;
import net.openhft.chronicle.ChronicleHashErrorListeners;
import net.openhft.chronicle.TimeProvider;
import net.openhft.chronicle.*;
import net.openhft.chronicle.serialization.AgileBytesMarshaller;
import net.openhft.chronicle.serialization.MetaBytesInterop;
import net.openhft.chronicle.serialization.MetaBytesWriter;
@@ -48,8 +45,8 @@ import java.util.concurrent.TimeUnit;
import static net.openhft.chronicle.map.Objects.builderEquals;
/**
* {@code ChronicleMapBuilder} manages the whole set of {@link ChronicleMap} configurations, could be used as
* a classic builder and/or factory. This means that in addition to the standard builder
* {@code ChronicleMapBuilder} manages the whole set of {@link ChronicleMap} configurations, could
* be used as a classic builder and/or factory. This means that in addition to the standard builder
* usage pattern: <pre>{@code
* ChronicleMap<Key, Value> map = ChronicleMapBuilder
* .of(Key.class, Value.class)
@@ -71,17 +68,18 @@ import static net.openhft.chronicle.map.Objects.builderEquals;
* documentation.
*
* <p>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.
*
* <p>{@code ChronicleMap} ("ChronicleMaps, created by {@code ChronicleMapBuilder}") currently doesn't support
* resizing. That is why you should <i>always</i> configure {@linkplain #entries(long) number of entries} you
* are going to insert into the created map <i>at most</i>. See {@link #entries(long)} method documentation
* for more information on this.
* <p>{@code ChronicleMap} ("ChronicleMaps, created by {@code ChronicleMapBuilder}") currently
* doesn't support resizing. That is why you should <i>always</i> configure {@linkplain
* #entries(long) number of entries} you are going to insert into the created map <i>at most</i>.
* See {@link #entries(long)} method documentation for more information on this.
*
* <p>{@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 <K> key type of the maps, produced by this builder
* @param <V> value type of the maps, produced by this builder
@@ -130,7 +128,8 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
private V defaultValue = null;
private DefaultValueProvider<K, V> defaultValueProvider = null;
private byte identifier = -1;
private boolean isStateless;
private StatelessBuilder statelessBuilder;
ChronicleMapBuilder(Class<K> keyClass, Class<V> valueClass) {
keyBuilder = new SerializationBuilder<K>(keyClass, SerializationBuilder.Role.KEY);
@@ -191,9 +190,9 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
* .keySize(10)
* // shouldn't specify valueSize(), because it is statically known
* .create();}</pre>
* (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<K, V> 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.
*
* <p>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.
* <p>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.
*
* <p>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.
* <p>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<K, V> 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.
*
* <p>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.
* <p>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.
*
* <p>If value size varies, method {@link #valueSize(int)} or {@link #entrySize(int)} should be called
* instead of this one.
* <p>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<K, V> implements Cloneable,
* {@inheritDoc}
*
* <p>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.
*
* <p>If entry size is not configured explicitly by calling this method, it is computed based on
* {@linkplain #metaDataBytes(int) meta data bytes}, plus {@linkplain #keySize(int) key size}, plus
* {@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}.
*
* <p>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<K, V> 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.
*
* <p>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.
* <p>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.
*
* <p>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).
* <p>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).
*
* <p>Default is {@link Alignment#OF_4_BYTES}.
*
@@ -437,20 +437,21 @@ public class ChronicleMapBuilder<K, V> 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.
*
* <p>{@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.
* <p>{@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.
*
* <p>By default, of cause, {@code ChronicleMap} conforms the general {@code Map} contract and returns the
* previous mapped value on {@code put()} calls.
* <p>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<K, V> 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.
*
* <p>{@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.
* <p>{@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<K, V> implements Cloneable,
* .create();}</pre>
*
* <p>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<K, V> objectSerializer(ObjectSerializer objectSerializer) {
@@ -665,6 +666,12 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
return this;
}
@Override
public ChronicleMapBuilder<K, V> stateless(StatelessBuilder statelessBuilder) {
this.statelessBuilder = statelessBuilder;
return this;
}
public ChronicleMapBuilder<K, V> valueMarshallerAndFactory(
@NotNull BytesMarshaller<V> valueMarshaller, @NotNull ObjectFactory<V> valueFactory) {
valueBuilder.marshaller(valueMarshaller, valueFactory);
@@ -694,18 +701,18 @@ public class ChronicleMapBuilder<K, V> 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.
*
* <p>Setting default value to {@code null} is interpreted as map shouldn't put any default value for
* absent keys. This is by default.
* <p>Setting default value to {@code null} is interpreted as map shouldn't put any default
* value for absent keys. This is by default.
*
* <p>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<K, V> defaultValue(V defaultValue) {
@@ -715,13 +722,13 @@ public class ChronicleMapBuilder<K, V> 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.
*
* <p>This configuration overrides any previous {@link #defaultValue(Object)} configuration to this {@code
* ChronicleMapBuilder}.
* <p>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<K, V> implements Cloneable,
}
@Override
public ChronicleMapBuilder<K, V> 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<K, V> implements Cloneable,
return establishReplication(map);
}
@Override
public ChronicleMap<K, V> create() throws IOException {
if (statelessBuilder != null) {
return createStatelessMap(statelessBuilder);
}
VanillaChronicleMap<K, ?, ?, V, ?, ?> map = newMap();
BytesStore bytesStore = new DirectStore(JDKObjectSerializer.INSTANCE,
map.sizeInBytes(), true);
@@ -815,6 +821,15 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
return establishReplication(map);
}
private ChronicleMap<K, V> createStatelessMap(StatelessBuilder statelessBuilder) throws IOException {
preMapConstruction();
KeyValueSerializer keyValueSerializer = new KeyValueSerializer(this.keyBuilder,
this.valueBuilder);
return new StatelessMapClient<K, V>(keyValueSerializer,
statelessBuilder);
}
private VanillaChronicleMap<K, ?, ?, V, ?, ?> newMap() throws IOException {
preMapConstruction();
@@ -831,7 +846,7 @@ public class ChronicleMapBuilder<K, V> implements Cloneable,
return identifier != -1;
}
void preMapConstruction() {
void preMapConstruction() {
keyBuilder.objectSerializer(objectSerializer());
valueBuilder.objectSerializer(objectSerializer());
src/main/java/net/openhft/chronicle/map/ConcurrentExpiryMap.java
-14
View File
@@ -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.
*/
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+70
View File
@@ -1525,4 +1525,74 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
}
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.
* <p/>
* <p>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());
}
};
}
}
src/main/java/net/openhft/chronicle/map/Replicator.java
-55
View File
@@ -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.
* <p/>
* <p>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;
}
src/main/java/net/openhft/chronicle/map/Replicators.java
-62
View File
@@ -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());
}
};
}
}
src/main/java/net/openhft/chronicle/map/StatelessMapClient.java
+29 -87
View File
@@ -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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Closeable {
REMOVE_WITH_VALUE
}
private final ThreadLocal<Buffer> sourceBuffer = new ThreadLocal<Buffer>();
private final Map<Long, Object> transactionIDs = new HashMap<Long, Object>();
// private final ThreadLocal<Buffer> sourceBuffer = new ThreadLocal<Buffer>();
private final AtomicLong transactionID = new AtomicLong();
private long timeoutMs = TimeUnit.SECONDS.toMillis(20);
// private long timeoutMs = TimeUnit.SECONDS.toMillis(20);
public StatelessMapClient(final KeyValueSerializer<K,
V> keyValueSerializer, final InetSocketAddress remote) throws IOException {
public StatelessMapClient(final KeyValueSerializer<K, V> 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<K, V> implements ChronicleMap<K, V>, Closeable {
}
public File file() {
return null;
}
@@ -117,12 +115,10 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V>, 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<K, V> keyValueSerializer;
public synchronized V putIfAbsent(K key, V value) {
long sizeLocation = writeEvent(PUT_IF_ABSENT);
@@ -138,7 +134,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, Closeable {
long sizeLocation = writeEvent(CLEAR);
// get the data back from the server
blockingFetch(sizeLocation, 20000);
blockingFetch(sizeLocation);
}
@@ -288,7 +284,7 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> {
final K key;
@@ -455,9 +395,9 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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() {
}
src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java
-368
View File
@@ -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<K, V> {
private final KeyValueSerializer<K, V> keyValueSerializer;
private final ChronicleMap<K, V> map;
private double maxEntrySizeBytes;
public StatelessServerConnector(KeyValueSerializer<K, V> keyValueSerializer, ChronicleMap<K, V> 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<K> ks = map.keySet();
out.writeStopBit(ks.size());
final Iterator<K> 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<V> 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<Map.Entry<K, V>> entries = map.entrySet();
out.writeStopBit(entries.size());
for (Map.Entry<K, V> 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<K, V> readEntries(Bytes in) {
long size = in.readStopBit();
final HashMap<K, V> result = new HashMap<K, V>();
for (long i = 0; i < size; i++) {
result.put(readKey(in), readValue(in));
}
return result;
}
}
class KeyValueSerializer<K, V> {
private final Serializer<V> valueSerializer;
private final Serializer<K> keySerializer;
KeyValueSerializer(SerializationBuilder<K> keySerializer,
SerializationBuilder<V> valueSerializer) {
this.keySerializer = new Serializer<K>(keySerializer);
this.valueSerializer = new Serializer<V>(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);
}
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+365 -2
View File
@@ -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<K, V> {
private final KeyValueSerializer<K, V> keyValueSerializer;
private final ChronicleMap<K, V> map;
private double maxEntrySizeBytes;
StatelessServerConnector(KeyValueSerializer<K, V> keyValueSerializer, ChronicleMap<K,
V> 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<K> ks = map.keySet();
out.writeStopBit(ks.size());
final Iterator<K> 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<V> 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<Map.Entry<K, V>> entries = map.entrySet();
out.writeStopBit(entries.size());
for (Map.Entry<K, V> 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<K, V> readEntries(Bytes in) {
long size = in.readStopBit();
final HashMap<K, V> result = new HashMap<K, V>();
for (long i = 0; i < size; i++) {
result.put(readKey(in), readValue(in));
}
return result;
}
}
class KeyValueSerializer<K, V> {
private final Serializer<V> valueSerializer;
private final Serializer<K> keySerializer;
KeyValueSerializer(SerializationBuilder<K> keySerializer,
SerializationBuilder<V> valueSerializer) {
this.keySerializer = new Serializer<K>(keySerializer);
this.valueSerializer = new Serializer<V>(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);
}
}
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+3 -2
View File
@@ -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<E>
}
@Override
public ChronicleSetBuilder<E> stateless(boolean isStateless) {
chronicleMapBuilder.stateless(isStateless);
public ChronicleSetBuilder<E> stateless(StatelessBuilder statelessBuilder) {
chronicleMapBuilder.stateless(statelessBuilder);
return this;
}