diff --git a/src/main/java/net/openhft/chronicle/map/Serializer.java b/src/main/java/net/openhft/chronicle/map/Serializer.java index 52643c9f2..d91eec3da 100644 --- a/src/main/java/net/openhft/chronicle/map/Serializer.java +++ b/src/main/java/net/openhft/chronicle/map/Serializer.java @@ -11,7 +11,7 @@ import net.openhft.lang.model.Byteable; import net.openhft.lang.model.constraints.NotNull; /** - * uses a {@code SerializationBuilder} to readMarshallable and writeMarshallable objects + * uses a {@code SerializationBuilder} to read and write object(s) as bytes * * @author Rob Austin. */ @@ -27,23 +27,21 @@ public class Serializer> { public Serializer(final SerializationBuilder serializationBuilder) { - sizeMarshaller = serializationBuilder.sizeMarshaller(); - originalReader = serializationBuilder.reader(); - - originalWriter = (VW) serializationBuilder.interop(); originalMetaValueWriter = (MVW) serializationBuilder.metaInterop(); metaValueWriterProvider = (MetaProvider) serializationBuilder.metaInteropProvider(); + originalReader = serializationBuilder.reader(); + originalWriter = (VW) serializationBuilder.interop(); + readerProvider = Provider.of((Class) originalReader.getClass()); writerProvider = Provider.of((Class) originalWriter.getClass()); } public O readMarshallable(@NotNull Bytes in) throws IllegalStateException { - ThreadLocalCopies copies; - copies = readerProvider.getCopies(null); - BytesReader valueReader = readerProvider.get(copies, originalReader); + final ThreadLocalCopies copies = readerProvider.getCopies(null); + final BytesReader valueReader = readerProvider.get(copies, originalReader); try { long valueSize = sizeMarshaller.readSize(in); @@ -55,35 +53,32 @@ public class Serializer> { public void writeMarshallable(@NotNull Bytes out, O value) { - boolean byteableValue = value instanceof Byteable; - ThreadLocalCopies copies; - copies = readerProvider.getCopies(null); + ThreadLocalCopies copies = readerProvider.getCopies(null); VW valueWriter = writerProvider.get(copies, originalWriter); copies = writerProvider.getCopies(copies); - long valueSize; - MetaBytesWriter metaValueWriter = null; - Byteable valueAsByteable = null; - if (!byteableValue) { + final long valueSize; + + MetaBytesWriter metaValueWriter = null; + + if ((value instanceof Byteable)) { + valueSize = ((Byteable) value).maxSize(); + } else { copies = writerProvider.getCopies(copies); valueWriter = writerProvider.get(copies, originalWriter); copies = metaValueWriterProvider.getCopies(copies); metaValueWriter = metaValueWriterProvider.get( copies, originalMetaValueWriter, valueWriter, value); valueSize = metaValueWriter.size(valueWriter, value); - } else { - valueAsByteable = (Byteable) value; - valueSize = valueAsByteable.maxSize(); } sizeMarshaller.writeSize(out, valueSize); - - if (metaValueWriter != null) { + if (metaValueWriter != null) metaValueWriter.write(valueWriter, out, value); - } else { + else throw new UnsupportedOperationException(""); - } + } } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java index 3297e5b20..5693cc691 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessMapClient.java @@ -1,15 +1,11 @@ package net.openhft.chronicle.map; -import net.openhft.chronicle.map.serialization.MetaBytesInterop; -import net.openhft.chronicle.map.serialization.MetaBytesWriter; -import net.openhft.chronicle.map.threadlocal.ThreadLocalCopies; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.Bytes; import org.jetbrains.annotations.NotNull; import java.io.File; import java.io.IOException; -import java.io.Serializable; import java.util.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -22,34 +18,7 @@ import static net.openhft.chronicle.map.StatelessMapClient.EventId.*; * * @author Rob Austin. */ -public class StatelessMapClient, - V, VW, MVW extends MetaBytesWriter> extends AbstractMap - implements ChronicleMap, Serializable { - - - private ThreadLocal sourceBuffer = new ThreadLocal(); - - - Map transactionIDs = new HashMap(); - - - AtomicLong transactionID = new AtomicLong(); - - private long timeoutMs = TimeUnit.SECONDS.toMillis(20); - private final Serializer valueSerializer; - private final Serializer keySerializer; - - - @Override - public File file() { - return null; - } - - @Override - public void close() throws IOException { - throw new UnsupportedOperationException("This is not supported in the " + this.getClass() - .getSimpleName()); - } +public class StatelessMapClient implements ChronicleMap { public static enum EventId { LONG_SIZE, SIZE, @@ -69,6 +38,27 @@ public class StatelessMapClient, REMOVE_WITH_VALUE } + private final ThreadLocal sourceBuffer = new ThreadLocal(); + private final Map transactionIDs = new HashMap(); + + private final AtomicLong transactionID = new AtomicLong(); + + private long timeoutMs = TimeUnit.SECONDS.toMillis(20); + private final Serializer valueSerializer; + private final Serializer keySerializer; + + + @Override + public File file() { + return null; + } + + @Override + public void close() throws IOException { + throw new UnsupportedOperationException("This is not supported in the " + this.getClass() + .getSimpleName()); + } + long nextUniqueTransaction() { long time = System.currentTimeMillis(); @@ -86,165 +76,127 @@ public class StatelessMapClient, private final Bytes out; - public StatelessMapClient(Bytes out, ChronicleMapBuilder builder) { + public StatelessMapClient(Bytes out, Class kClass, Class vClass) { - valueSerializer = new Serializer(builder.valueBuilder); - keySerializer = new Serializer(builder.keyBuilder); + 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) { - out.writeByte(PUT_IF_ABSENT.ordinal()); - + writeEvent(PUT_IF_ABSENT); writeKey(key); writeValue(value); - return null; + return readValue(blockingFetch()); + } @Override public boolean remove(Object key, Object value) { - out.writeByte(REMOVE_WITH_VALUE.ordinal()); - - + writeEvent(REMOVE_WITH_VALUE); writeKey((K) key); writeValue((V) value); - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch().readBoolean(); } + @Override public boolean replace(K key, V oldValue, V newValue) { - out.writeByte(REPLACE_WITH_OLD_AND_NEW_VALUE.ordinal()); - + writeEvent(REPLACE_WITH_OLD_AND_NEW_VALUE); writeKey(key); writeValue(oldValue); writeValue(newValue); - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch().readBoolean(); + } + private void writeValue(V newValue) { + valueSerializer.writeMarshallable(out, newValue); } @Override public V replace(K key, V value) { - out.writeByte(REPLACE.ordinal()); + writeEvent(REPLACE); writeKey(key); writeValue(value); - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - // get the data back from the server - return readValue(blockingFetch(transactionId)); + return readValue(blockingFetch()); } @Override public int size() { - out.writeByte(SIZE.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); + writeEvent(SIZE); // get the data back from the server - return blockingFetch(transactionId).readInt(); + return blockingFetch().readInt(); } @Override public boolean isEmpty() { - out.writeByte(IS_EMPTY.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); + writeEvent(IS_EMPTY); // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch().readBoolean(); } @Override public boolean containsKey(Object key) { - out.writeByte(CONTAINS_KEY.ordinal()); - + writeEvent(CONTAINS_KEY); writeKey((K) key); - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch().readBoolean(); } @Override public boolean containsValue(Object value) { - out.writeByte(CONTAINS_VALUE.ordinal()); + writeEvent(CONTAINS_VALUE); writeValue((V) value); - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - // get the data back from the server - return blockingFetch(transactionId).readBoolean(); + return blockingFetch().readBoolean(); } @Override public long longSize() { - out.writeByte(LONG_SIZE.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); + writeEvent(LONG_SIZE); // get the data back from the server - return blockingFetch(transactionId).readLong(); + return blockingFetch().readLong(); } @Override public V get(Object key) { - out.writeByte(GET.ordinal()); - + writeEvent(GET); writeKey((K) key); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - // get the data back from the server - return readValue(blockingFetch(transactionId)); + return readValue(blockingFetch()); } - V readValue(ByteBufferBytes source) { - - + V readValue(Bytes source) { return valueSerializer.readMarshallable(source); - - } @Override @@ -259,61 +211,43 @@ public class StatelessMapClient, @Override public V put(K key, V value) { - out.writeByte(PUT.ordinal()); + writeEvent(PUT); + writeKey(key); + writeValue(value); - writeKey((K) key); - writeValue((V) value); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - // get the data back from the server - return readValue(blockingFetch(transactionId)); + return readValue(blockingFetch()); } @Override public V remove(Object key) { - out.writeByte(REMOVE.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - + writeEvent(REMOVE); + writeKey((K) key); // get the data back from the server - return readValue(blockingFetch(transactionId)); + return readValue(blockingFetch()); } @Override public void putAll(Map map) { - throw new UnsupportedOperationException("todo"); - /* out.writeByte(PUT_ALL.ordinal()); - - - HashMap safeCopy = new HashMap(map); - - writeEntries(copies, safeCopy); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); + writeEvent(PUT_ALL); + writeEntries(map); // get the data back from the server - blockingFetch(transactionId); -*/ - //return; - + blockingFetch(); } - private void writeEntries(ThreadLocalCopies copies, HashMap safeCopy) { + private void writeEntries(Map map) { + + final HashMap safeCopy = new HashMap(map); + out.writeLong(safeCopy.size()); - Set entries = (Set) safeCopy.entrySet(); + final Set entries = (Set) safeCopy.entrySet(); for (Entry e : entries) { writeKey(e.getKey()); @@ -323,68 +257,50 @@ public class StatelessMapClient, @Override public void clear() { - out.writeByte(CLEAR.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); + writeEvent(CLEAR); // get the data back from the server - blockingFetch(transactionId); + blockingFetch(); - return; } @NotNull @Override public Set keySet() { - out.writeByte(KEY_SET.ordinal()); - - // send the transaction id - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); + writeEvent(KEY_SET); // get the data back from the server - return readKeySet(blockingFetch(transactionId)); + return readKeySet(blockingFetch()); } - private Set readKeySet(ByteBufferBytes bufferBytes) { - int size = bufferBytes.readInt(); - - HashSet result = new HashSet<>(); + private Set readKeySet(Bytes in) { + int size = in.readInt(); + final HashSet result = new HashSet<>(); for (int i = 0; i < size; i++) { - result.add(readKey(bufferBytes)); + result.add(readKey(out)); } return result; } - private K readKey(ByteBufferBytes bufferBytes) { - long transactionId = nextUniqueTransaction(); - out.writeLong(transactionId); - - // get the data back from the server - return readKey(blockingFetch(transactionId)); - + private K readKey(Bytes bufferBytes) { + return keySerializer.readMarshallable(bufferBytes); } @NotNull @Override public Collection values() { - out.writeByte(VALUES.ordinal()); - long transactionId = nextUniqueTransaction(); - - out.writeLong(transactionId); + writeEvent(VALUES); // get the data back from the server - ByteBufferBytes bufferBytes = blockingFetch(transactionId); + Bytes in = blockingFetch(); - int size = bufferBytes.readInt(); + int size = in.readInt(); ArrayList result = new ArrayList(); for (int i = 0; i < size; i++) { - result.add(readValue(bufferBytes)); + result.add(readValue(in)); } return result; } @@ -392,33 +308,27 @@ public class StatelessMapClient, @NotNull @Override public Set> entrySet() { - out.writeByte(ENTRY_SET.ordinal()); + writeEvent(ENTRY_SET); - long transactionId = nextUniqueTransaction(); // get the data back from the server - ByteBufferBytes bufferBytes = blockingFetch(transactionId); + Bytes bytes = blockingFetch(); - int size = bufferBytes.readInt(); + int size = bytes.readInt(); Set> result = new HashSet>(); for (int i = 0; i < size; i++) { - K k = readKey(bufferBytes); - V v = readValue(bufferBytes); + K k = keySerializer.readMarshallable(bytes); + V v = valueSerializer.readMarshallable(bytes); result.add(new Entry(k, v)); } - return result; } - void writeValue(V value) { - valueSerializer.writeMarshallable(out, value); - } - /** * write the keysize and the key to the the {@code target} buffer * @@ -428,6 +338,9 @@ public class StatelessMapClient, keySerializer.writeMarshallable(out, key); } + private void writeEvent(EventId event) { + out.write((byte) event.ordinal()); + } public Buffer buffer() { Buffer result = sourceBuffer.get(); @@ -463,7 +376,7 @@ public class StatelessMapClient, * @param transactionId * @return */ - private ByteBufferBytes blockingFetch(long transactionId) { + private Bytes blockingFetch(long transactionId) { final Buffer buffer = buffer(); buffer.set(null); @@ -538,6 +451,13 @@ public class StatelessMapClient, } + + + private Bytes blockingFetch() { + long transactionId = nextUniqueTransaction(); + out.writeLong(transactionId); + return blockingFetch(transactionId); + } }