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 improved StatelessServerConnector, but its still in draft.

Browse Source
This commit is contained in:
Rob Austin
2014-10-16 10:35:55 +01:00
parent 56a3046344
commit 6030b6338c
4 changed files with 104 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/Serializer.java
+12 -15
View File
@@ -1,9 +1,6 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.map.serialization.BytesReader;
import net.openhft.chronicle.map.serialization.MetaBytesWriter;
import net.openhft.chronicle.map.serialization.MetaProvider;
import net.openhft.chronicle.map.serialization.SizeMarshaller;
import net.openhft.chronicle.map.serialization.*;
import net.openhft.chronicle.map.threadlocal.Provider;
import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies;
import net.openhft.lang.io.Bytes;
@@ -15,24 +12,24 @@ import net.openhft.lang.model.constraints.NotNull;
*
* @author Rob Austin.
*/
public class Serializer<O, VW, MVW extends MetaBytesWriter<O, VW>> {
class Serializer<O> {
final SizeMarshaller sizeMarshaller;
final BytesReader<O> originalReader;
transient Provider<BytesReader<O>> readerProvider;
final VW originalWriter;
transient Provider<VW> writerProvider;
final MVW originalMetaValueWriter;
final MetaProvider<O, VW, MVW> metaValueWriterProvider;
final Object originalWriter;
transient Provider writerProvider;
final MetaBytesInterop originalMetaValueWriter;
final MetaProvider metaValueWriterProvider;
public Serializer(final SerializationBuilder<O> serializationBuilder) {
Serializer(final SerializationBuilder serializationBuilder) {
sizeMarshaller = serializationBuilder.sizeMarshaller();
originalMetaValueWriter = (MVW) serializationBuilder.metaInterop();
metaValueWriterProvider = (MetaProvider) serializationBuilder.metaInteropProvider();
originalMetaValueWriter = serializationBuilder.metaInterop();
metaValueWriterProvider = serializationBuilder.metaInteropProvider();
originalReader = serializationBuilder.reader();
originalWriter = (VW) serializationBuilder.interop();
originalWriter = serializationBuilder.interop();
readerProvider = Provider.of((Class) originalReader.getClass());
writerProvider = Provider.of((Class) originalWriter.getClass());
@@ -54,12 +51,12 @@ public class Serializer<O, VW, MVW extends MetaBytesWriter<O, VW>> {
public void writeMarshallable(O value, @NotNull Bytes out) {
ThreadLocalCopies copies = readerProvider.getCopies(null);
VW valueWriter = writerProvider.get(copies, originalWriter);
Object valueWriter = writerProvider.get(copies, originalWriter);
copies = writerProvider.getCopies(copies);
final long valueSize;
MetaBytesWriter<O, VW> metaValueWriter = null;
MetaBytesWriter metaValueWriter = null;
if ((value instanceof Byteable)) {
valueSize = ((Byteable) value).maxSize();
src/main/java/net/openhft/chronicle/map/StatelessMapClient.java
+30 -48
View File
@@ -18,7 +18,7 @@ import static net.openhft.chronicle.map.StatelessMapClient.EventId.*;
*
* @author Rob Austin.
*/
public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
public static enum EventId {
LONG_SIZE, SIZE,
@@ -44,20 +44,11 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
private final AtomicLong transactionID = new AtomicLong();
private long timeoutMs = TimeUnit.SECONDS.toMillis(20);
private final Serializer<V, ?, ?> valueSerializer;
private final Serializer<K, ?, ?> keySerializer;
public StatelessMapClient(Bytes out, Class<K> kClass, Class<V> vClass) {
final SerializationBuilder<K> keyBuilder = new SerializationBuilder<K>(kClass,
SerializationBuilder.Role.KEY);
final SerializationBuilder<V> valueBuilder = new SerializationBuilder<V>(vClass,
SerializationBuilder.Role.VALUE);
keySerializer = new Serializer(keyBuilder);
valueSerializer = new Serializer(valueBuilder);
public StatelessMapClient(Bytes out, KeyValueSerializer<K, V> keyValueSerializer) {
this.out = out;
this.keyValueSerializer = keyValueSerializer;
}
@Override
@@ -86,6 +77,7 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
private final Bytes out;
private final KeyValueSerializer<K, V> keyValueSerializer;
@Override
@@ -94,8 +86,7 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
writeKey(key);
writeValue(value);
return readValue(blockingFetch());
return readKey();
}
@@ -110,7 +101,6 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
}
@Override
public boolean replace(K key, V oldValue, V newValue) {
writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE);
@@ -122,20 +112,19 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
return blockingFetch().readBoolean();
}
private void writeValue(V newValue) {
valueSerializer.writeMarshallable(newValue, out);
private void writeKey(K key) {
keyValueSerializer.writeKey(key, out);
}
@Override
public V replace(K key, V value) {
writeEvent(REPLACE);
writeKey(key);
writeValue(value);
// get the data back from the server
return readValue(blockingFetch());
return readKey();
}
@@ -189,13 +178,10 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
writeEvent(GET);
writeKey((K) key);
// get the data back from the server
return readValue(blockingFetch());
return readKey();
}
V readValue(Bytes source) {
return valueSerializer.readMarshallable(source);
}
@Override
public V getUsing(K key, V value) {
@@ -209,22 +195,30 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
@Override
public V put(K key, V value) {
writeEvent(PUT);
writeKey(key);
writeValue(value);
// get the data back from the server
return readValue(blockingFetch());
return readKey();
}
private V readKey() {
return keyValueSerializer.readValue(blockingFetch());
}
private void writeValue(V value) {
keyValueSerializer.writeValue(value, out);
}
@Override
public V remove(Object key) {
writeEvent(REMOVE);
writeKey((K) key);
// get the data back from the server
return readValue(blockingFetch());
return readKey();
}
@@ -276,14 +270,11 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
final HashSet<K> result = new HashSet<>();
for (long i = 0; i < size; i++) {
result.add(readKey(out));
result.add(keyValueSerializer.readKey(in));
}
return result;
}
private K readKey(Bytes bufferBytes) {
return keySerializer.readMarshallable(bufferBytes);
}
@NotNull
@Override
@@ -291,21 +282,22 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
writeEvent(VALUES);
// get the data back from the server
final Bytes in = blockingFetch();
final Bytes in = blockingFetch();
final long size = in.readStopBit();
final long size = in.readStopBit();
if (size > Integer.MAX_VALUE)
throw new IllegalStateException("size=" + size + " is too large.");
final ArrayList<V> result = new ArrayList<V>((int) size);
final ArrayList<V> result = new ArrayList<V>((int) size);
for (int i = 0; i < size; i++) {
result.add(readValue(in));
result.add(keyValueSerializer.readValue(in));
}
return result;
}
@NotNull
@Override
public Set<Map.Entry<K, V>> entrySet() {
@@ -313,15 +305,15 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
// get the data back from the server
Bytes bytes = blockingFetch();
Bytes in = blockingFetch();
long size = bytes.readStopBit();
long size = in.readStopBit();
Set<Map.Entry<K, V>> result = new HashSet<Map.Entry<K, V>>();
for (int i = 0; i < size; i++) {
K k = keySerializer.readMarshallable(bytes);
V v = valueSerializer.readMarshallable(bytes);
K k = keyValueSerializer.readKey(in);
V v = keyValueSerializer.readValue(in);
result.add(new Entry(k, v));
}
@@ -329,15 +321,6 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
}
/**
* write the keysize and the key to the the {@code target} buffer
*
* @param key the key of the map
*/
private void writeKey(K key) {
keySerializer.writeMarshallable(key, out);
}
private void writeEvent(EventId event) {
out.write((byte) event.ordinal());
}
@@ -449,7 +432,6 @@ public class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
return getKey() + "=" + getValue();
}
}
src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java
+51 -28
View File
@@ -11,27 +11,16 @@ import java.util.Set;
/**
* @author Rob Austin.
*/
public class StatelessServerConnector<K, V> {
class StatelessServerConnector<K, V> {
private final KeyValueSerializer<K, V> keyValueSerializer;
private final ChronicleMap<K, V> map;
private final Serializer<V, ?, ?> valueSerializer;
private final Serializer<K, ?, ?> keySerializer;
public StatelessServerConnector(Class<K> kClass, Class<V> vClass, ChronicleMap<K, V> map) {
public StatelessServerConnector(KeyValueSerializer<K, V> keyValueSerializer, ChronicleMap<K, V> map) {
this.keyValueSerializer = keyValueSerializer;
this.map = map;
final SerializationBuilder<K> keyBuilder = new SerializationBuilder<K>(kClass,
SerializationBuilder.Role.KEY);
final SerializationBuilder<V> valueBuilder = new SerializationBuilder<V>(vClass,
SerializationBuilder.Role.VALUE);
keySerializer = new Serializer(keyBuilder);
valueSerializer = new Serializer(valueBuilder);
}
@@ -112,46 +101,39 @@ public class StatelessServerConnector<K, V> {
map.replace(key, oldValue, newValue);
}
public void longSize(Bytes in, Bytes out) {
reflectTransactionId(in, out);
out.writeLong(map.longSize());
}
public void size(Bytes in, Bytes out) {
reflectTransactionId(in, out);
out.writeInt(map.size());
}
public void isEmpty(Bytes in, Bytes out) {
reflectTransactionId(in, out);
out.writeBoolean(map.isEmpty());
}
public void containsKey(Bytes in, Bytes out) {
K k = readKey(in);
reflectTransactionId(in, out);
out.writeBoolean(map.containsKey(k));
}
public void containsValue(Bytes in, Bytes out) {
V v = readValue(in);
reflectTransactionId(in, out);
out.writeBoolean(map.containsValue(v));
}
public void get(Bytes in, Bytes out) {
K k = readKey(in);
reflectTransactionId(in, out);
writeValue(map.get(k), out);
}
public void put(Bytes in, Bytes out) {
K k = readKey(in);
V v = readValue(in);
@@ -159,14 +141,12 @@ public class StatelessServerConnector<K, V> {
writeValue(map.put(k, v), out);
}
public void remove(Bytes in, Bytes out) {
final V value = map.remove(readKey(in));
reflectTransactionId(in, out);
writeValue(value, out);
}
public void putAll(Bytes in, Bytes out) {
map.putAll(readEntries(in));
reflectTransactionId(in, out);
@@ -236,14 +216,13 @@ public class StatelessServerConnector<K, V> {
map.replace(k, v);
}
/**
* 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) {
keySerializer.writeMarshallable(key, out);
keyValueSerializer.writeKey(key, out);
}
private void reflectTransactionId(Bytes in, Bytes out) {
@@ -251,15 +230,59 @@ public class StatelessServerConnector<K, V> {
}
private void writeValue(V value, final Bytes out) {
valueSerializer.writeMarshallable(value, out);
keyValueSerializer.writeValue(value, out);
}
private K readKey(Bytes in) {
return keySerializer.readMarshallable(in);
return keyValueSerializer.readKey(in);
}
private V readValue(Bytes in) {
return keyValueSerializer.readValue(in);
}
}
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);
}
}