diff --git a/src/main/java/net/openhft/chronicle/map/Serializer.java b/src/main/java/net/openhft/chronicle/map/Serializer.java index d91eec3da..00bcdacb6 100644 --- a/src/main/java/net/openhft/chronicle/map/Serializer.java +++ b/src/main/java/net/openhft/chronicle/map/Serializer.java @@ -51,7 +51,7 @@ public class Serializer> { } } - public void writeMarshallable(@NotNull Bytes out, O value) { + public void writeMarshallable(O value, @NotNull Bytes out) { ThreadLocalCopies copies = readerProvider.getCopies(null); VW valueWriter = writerProvider.get(copies, originalWriter); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java index 5693cc691..967a000a8 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java @@ -47,6 +47,18 @@ public class StatelessMapClient implements ChronicleMap { private final Serializer valueSerializer; private final Serializer keySerializer; + public StatelessMapClient(Bytes out, Class kClass, Class vClass) { + + final SerializationBuilder keyBuilder = new SerializationBuilder(kClass, + SerializationBuilder.Role.KEY); + final SerializationBuilder valueBuilder = new SerializationBuilder(vClass, + SerializationBuilder.Role.VALUE); + + keySerializer = new Serializer(keyBuilder); + valueSerializer = new Serializer(valueBuilder); + + this.out = out; + } @Override public File file() { @@ -76,20 +88,6 @@ public class StatelessMapClient implements ChronicleMap { private final Bytes out; - public StatelessMapClient(Bytes out, Class kClass, Class vClass) { - - final SerializationBuilder keyBuilder = new SerializationBuilder(kClass, - SerializationBuilder.Role.KEY); - final SerializationBuilder valueBuilder = new SerializationBuilder(vClass, - SerializationBuilder.Role.VALUE); - - keySerializer = new Serializer(keyBuilder); - valueSerializer = new Serializer(valueBuilder); - - this.out = out; - } - - @Override public V putIfAbsent(K key, V value) { writeEvent(PUT_IF_ABSENT); @@ -125,7 +123,7 @@ public class StatelessMapClient implements ChronicleMap { } private void writeValue(V newValue) { - valueSerializer.writeMarshallable(out, newValue); + valueSerializer.writeMarshallable(newValue, out); } @@ -245,7 +243,7 @@ public class StatelessMapClient implements ChronicleMap { final HashMap safeCopy = new HashMap(map); - out.writeLong(safeCopy.size()); + out.writeStopBit(safeCopy.size()); final Set entries = (Set) safeCopy.entrySet(); @@ -274,10 +272,10 @@ public class StatelessMapClient implements ChronicleMap { } private Set readKeySet(Bytes in) { - int size = in.readInt(); + long size = in.readStopBit(); final HashSet result = new HashSet<>(); - for (int i = 0; i < size; i++) { + for (long i = 0; i < size; i++) { result.add(readKey(out)); } return result; @@ -293,11 +291,14 @@ public class StatelessMapClient implements ChronicleMap { writeEvent(VALUES); // get the data back from the server - Bytes in = blockingFetch(); + final Bytes in = blockingFetch(); - int size = in.readInt(); + final long size = in.readStopBit(); - ArrayList result = new ArrayList(); + if (size > Integer.MAX_VALUE) + throw new IllegalStateException("size=" + size + " is too large."); + + final ArrayList result = new ArrayList((int) size); for (int i = 0; i < size; i++) { result.add(readValue(in)); @@ -314,7 +315,7 @@ public class StatelessMapClient implements ChronicleMap { // get the data back from the server Bytes bytes = blockingFetch(); - int size = bytes.readInt(); + long size = bytes.readStopBit(); Set> result = new HashSet>(); @@ -322,7 +323,6 @@ public class StatelessMapClient implements ChronicleMap { K k = keySerializer.readMarshallable(bytes); V v = valueSerializer.readMarshallable(bytes); result.add(new Entry(k, v)); - } return result; @@ -335,7 +335,7 @@ public class StatelessMapClient implements ChronicleMap { * @param key the key of the map */ private void writeKey(K key) { - keySerializer.writeMarshallable(out, key); + keySerializer.writeMarshallable(key, out); } private void writeEvent(EventId event) { diff --git a/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java new file mode 100644 index 000000000..f40669bbf --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/StatelessServerConnector.java @@ -0,0 +1,265 @@ +package net.openhft.chronicle.map; + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.model.constraints.NotNull; + +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + +/** + * @author Rob Austin. + */ +public class StatelessServerConnector { + + + private final ChronicleMap map; + + + private final Serializer valueSerializer; + private final Serializer keySerializer; + + public StatelessServerConnector(Class kClass, Class vClass, ChronicleMap map) { + + this.map = map; + + final SerializationBuilder keyBuilder = new SerializationBuilder(kClass, + SerializationBuilder.Role.KEY); + final SerializationBuilder valueBuilder = new SerializationBuilder(vClass, + SerializationBuilder.Role.VALUE); + + keySerializer = new Serializer(keyBuilder); + valueSerializer = new Serializer(valueBuilder); + + } + + + public void marshall(@NotNull Bytes in, @NotNull Bytes out) { + + byte b = in.readByte(); + + StatelessMapClient.EventId eventId = StatelessMapClient.EventId.values()[b]; + + switch (eventId) { + + case LONG_SIZE: + longSize(in, out); + break; + + case IS_EMPTY: + isEmpty(in, out); + break; + + case CONTAINS_KEY: + containsKey(in, out); + break; + case CONTAINS_VALUE: + containsValue(in, out); + break; + case GET: + get(in, out); + break; + case PUT: + put(in, out); + break; + case REMOVE: + remove(in, out); + break; + case PUT_ALL: + putAll(in, out); + break; + case CLEAR: + clear(in, out); + break; + case KEY_SET: + keySet(in, out); + break; + case VALUES: + values(in, out); + break; + case ENTRY_SET: + entrySet(in, out); + break; + case REPLACE: + replace(in, out); + break; + case REPLACE_WITH_OLD_AND_NEW_VALUE: + replaceWithOldAndNew(in, out); + break; + case PUT_IF_ABSENT: + putIfAbsent(in, out); + break; + case REMOVE_WITH_VALUE: + removeWithValue(in, out); + break; + + } + + } + + private void removeWithValue(Bytes in, Bytes out) { + boolean result = map.remove(readKey(in), readValue(in)); + reflectTransactionId(in, out); + out.writeBoolean(result); + } + + private void replaceWithOldAndNew(Bytes in, Bytes out) { + final K key = readKey(in); + V oldValue = readValue(in); + V newValue = readValue(in); + reflectTransactionId(in, out); + 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); + reflectTransactionId(in, out); + 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); + } + + private Map readEntries(Bytes in) { + + long size = in.readStopBit(); + final HashMap result = new HashMap(); + + for (long i = 0; i < size; i++) { + result.put(readKey(in), readValue(in)); + } + return result; + + } + + public void clear(Bytes in, Bytes out) { + map.clear(); + reflectTransactionId(in, out); + } + + + public void keySet(Bytes in, Bytes out) { + reflectTransactionId(in, out); + + final Set ks = map.keySet(); + out.writeStopBit(ks.size()); + for (K key : ks) { + writeKey(key, out); + } + } + + public void values(Bytes in, Bytes out) { + reflectTransactionId(in, out); + final Collection values = map.values(); + out.writeStopBit(values.size()); + for (final V value : values) { + writeValue(value, out); + } + } + + public void entrySet(Bytes in, Bytes out) { + reflectTransactionId(in, out); + + final Set> entries = map.entrySet(); + out.writeStopBit(entries.size()); + for (Map.Entry e : entries) { + writeKey(e.getKey(), out); + writeValue(e.getValue(), out); + } + } + + public void putIfAbsent(Bytes in, Bytes out) { + K key = readKey(in); + V v = readValue(in); + reflectTransactionId(in, out); + writeValue(map.putIfAbsent(key, v), out); + } + + + public void replace(Bytes in, Bytes out) { + K k = readKey(in); + V v = readValue(in); + + reflectTransactionId(in, out); + 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); + } + + private void reflectTransactionId(Bytes in, Bytes out) { + out.writeLong(in.readLong()); + } + + private void writeValue(V value, final Bytes out) { + valueSerializer.writeMarshallable(value, out); + } + + private K readKey(Bytes in) { + return keySerializer.readMarshallable(in); + } + + private V readValue(Bytes in) { + return valueSerializer.readMarshallable(in); + } + +} diff --git a/src/test/java/net/openhft/chronicle/map/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java index eeea8b527..c5d52aa3e 100644 --- a/src/test/java/net/openhft/chronicle/map/SerializerTest.java +++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java @@ -59,7 +59,7 @@ public class SerializerTest { Serializer v = new Serializer(builder.valueBuilder); - v.writeMarshallable(out, value); + v.writeMarshallable(value, out); long position = out.position(); in.limit(position); @@ -83,7 +83,7 @@ public class SerializerTest { Serializer v = new Serializer(builder.keyBuilder); - v.writeMarshallable(out, key); + v.writeMarshallable(key, out); long position = out.position(); in.limit(position);