From af44d25788a308454aa83fd018308050e8124443 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 26 Dec 2014 17:59:52 +0000 Subject: [PATCH] added net.openhft.chronicle.map.StatelessChronicleMap.getUsing --- .../openhft/chronicle/map/ReaderWithSize.java | 9 ++-- .../chronicle/map/StatelessChronicleMap.java | 48 +++++++++++++++---- .../openhft/chronicle/map/TcpReplicator.java | 6 +-- .../openhft/chronicle/map/SerializerTest.java | 4 +- 4 files changed, 50 insertions(+), 17 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java b/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java index 13bf6f831..714efa832 100644 --- a/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java +++ b/src/main/java/net/openhft/chronicle/map/ReaderWithSize.java @@ -27,6 +27,7 @@ final class ReaderWithSize { private final SizeMarshaller sizeMarshaller; private final BytesReader originalReader; private final Provider> readerProvider; + ReaderWithSize(SerializationBuilder serializationBuilder) { sizeMarshaller = serializationBuilder.sizeMarshaller(); originalReader = serializationBuilder.reader(); @@ -37,17 +38,17 @@ final class ReaderWithSize { return readerProvider.getCopies(copies); } - public T read(Bytes in, @Nullable ThreadLocalCopies copies) { + public T read(Bytes in, @Nullable ThreadLocalCopies copies, @Nullable T using) { long size = sizeMarshaller.readSize(in); copies = readerProvider.getCopies(copies); BytesReader reader = readerProvider.get(copies, originalReader); - return reader.read(in, size); + return reader.read(in, size, using); } - public T readNullable(Bytes in, @Nullable ThreadLocalCopies copies) { + public T readNullable(Bytes in, @Nullable ThreadLocalCopies copies, T using) { if (in.readBoolean()) return null; - return read(in, copies); + return read(in, copies, using); } public BytesReader readerForLoop(@Nullable ThreadLocalCopies copies) { diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 98da4a27e..7c4ee9ea3 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -492,10 +492,31 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchObject(vClass, GET, (K) key); } - @NotNull + @Nullable public V getUsing(K key, V usingValue) { - throw new UnsupportedOperationException("getUsing() is not supported for stateless " + - "clients"); + + + final long startTime = System.currentTimeMillis(); + ThreadLocalCopies copies; + + long transactionId; + + outBytesLock.lock(); + try { + + final long sizeLocation = writeEventAnSkip(GET); + + copies = writeKey((K) key); + transactionId = send(sizeLocation, startTime); + } finally { + outBytesLock.unlock(); + } + + if (eventReturnsNull(GET)) + return null; + + return (V) readValue(transactionId, startTime, copies, usingValue); + } @NotNull @@ -1403,7 +1424,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private V readValue(long transactionId, long startTime, ThreadLocalCopies copies) { + private V readValue(long transactionId, long startTime, ThreadLocalCopies copies, V usingValue) { assert !outBytesLock.isHeldByCurrentThread(); long timeoutTime = startTime + this.config.timeoutMs(); @@ -1411,7 +1432,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // receive inBytesLock.lock(); try { - return readValue(copies, blockingFetchReadOnly(timeoutTime, transactionId)); + + if (usingValue != null) + return readValue(copies, blockingFetchReadOnly(timeoutTime, transactionId), usingValue); + else + + return readValue(copies, blockingFetchReadOnly(timeoutTime, transactionId)); } finally { inBytesLock.unlock(); } @@ -1481,7 +1507,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private V readValue(ThreadLocalCopies copies, Bytes in) { assert !outBytesLock.isHeldByCurrentThread(); assert inBytesLock.isHeldByCurrentThread(); - return valueReaderWithSize.readNullable(in, copies); + return valueReaderWithSize.readNullable(in, copies, null); + } + + private V readValue(ThreadLocalCopies copies, Bytes in, V using) { + assert !outBytesLock.isHeldByCurrentThread(); + assert inBytesLock.isHeldByCurrentThread(); + return valueReaderWithSize.readNullable(in, copies, using); } @SuppressWarnings("SameParameterValue") @@ -1703,7 +1735,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return null; if (rClass == vClass) - return (R) readValue(transactionId, startTime, copies); + return (R) readValue(transactionId, startTime, copies, null); else throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + "supported"); @@ -1732,7 +1764,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return null; if (rClass == vClass) - return (R) readValue(transactionId, startTime, copies); + return (R) readValue(transactionId, startTime, copies, null); else throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + "supported"); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index c7c870589..c78284b9a 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1565,7 +1565,7 @@ class StatelessServerConnector { @Nullable public Work mapForKey(@NotNull ByteBufferBytes reader, @NotNull Bytes writer, long sizeLocation) { - final K key = keyReaderWithSize.read(reader, null); + final K key = keyReaderWithSize.read(reader, null, null); final Function function = (Function) reader.readObject(); try { Object result = map.getMapped(key, function); @@ -1582,7 +1582,7 @@ class StatelessServerConnector { @Nullable public Work putMapped(@NotNull ByteBufferBytes reader, @NotNull Bytes writer, long sizeLocation) { - final K key = keyReaderWithSize.read(reader, null); + final K key = keyReaderWithSize.read(reader, null, null); final UnaryOperator unaryOperator = (UnaryOperator) reader.readObject(); try { Object result = map.putMapped(key, unaryOperator); @@ -1744,7 +1744,7 @@ class StatelessServerConnector { @Nullable private Work containsValue(Bytes reader, @NotNull Bytes writer, final long sizeLocation) { // todo optimize -- eliminate - final V v = valueReaderWithSize.read(reader, null); + final V v = valueReaderWithSize.read(reader, null, null); try { writer.writeBoolean(map.containsValue(v)); diff --git a/src/test/java/net/openhft/chronicle/map/SerializerTest.java b/src/test/java/net/openhft/chronicle/map/SerializerTest.java index 02dc96e62..f6dae510d 100644 --- a/src/test/java/net/openhft/chronicle/map/SerializerTest.java +++ b/src/test/java/net/openhft/chronicle/map/SerializerTest.java @@ -117,7 +117,7 @@ public class SerializerTest { long position = out.position(); in.limit(position); - Object actual = valueReaderWithSize.read(in, null); + Object actual = valueReaderWithSize.read(in, null, null); Assert.assertEquals(actual, value); } @@ -140,7 +140,7 @@ public class SerializerTest { long position = out.position(); in.limit(position); - Object actual = keyReaderWithSize.read(in, null); + Object actual = keyReaderWithSize.read(in, null, null); Assert.assertEquals(actual, key); } }