diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java index 49e2894bb..0706395e4 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMap.java @@ -28,6 +28,7 @@ import org.jetbrains.annotations.NotNull; import java.io.Closeable; import java.util.Map; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Future; /** * Extension of {@link ConcurrentMap} interface, stores the data off-heap. @@ -165,11 +166,11 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { * default value} or {@link AbstractChronicleMapBuilder#defaultValueProvider(DefaultValueProvider) * defaultValueProvider.} * - *

If the {@code ChronicleMap} is off-heap updatable, i. e. created via {@link - * ChronicleMapBuilder} builder (values are {@link Byteable}), there is one more option of - * what to do if the key is absent in the map, see {@link ChronicleMapBuilder#prepareValueBytesOnAcquire(PrepareValueBytes)}. - * By default, value bytes are just zeroed out, no default value, either provided for key or constant, is - * put for the absent key. + *

If the {@code ChronicleMap} is off-heap updatable, i. e. created via {@link ChronicleMapBuilder} + * builder (values are {@link Byteable}), there is one more option of what to do if the key is absent in + * the map, see {@link ChronicleMapBuilder#prepareValueBytesOnAcquire(PrepareValueBytes)}. By default, + * value bytes are just zeroed out, no default value, either provided for key or constant, is put for the + * absent key. * * @param key the key whose associated value is to be returned * @param usingValue the object to read value data in, if present. Can not be null @@ -183,8 +184,7 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue); /** - * Apply a mapping to the value returned by a key and return a result. - * A read lock is assumed. + * Apply a mapping to the value returned by a key and return a result. A read lock is assumed. * * @param key to apply the mapping to * @param function to calculate a result @@ -194,10 +194,8 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { R mapForKey(K key, @NotNull Function function); /** - * Apply a mutator to the value for a key and return a result. - * A write lock is assumed. - *

- * If there is no entry for this key it will be created and the value empty. + * Apply a mutator to the value for a key and return a result. A write lock is assumed.

If there is no + * entry for this key it will be created and the value empty. * * @param key to apply the mapping to * @param mutator to alter the value and calculate a result @@ -206,5 +204,17 @@ public interface ChronicleMap extends ConcurrentMap, ChronicleHash { */ R updateForKey(K key, @NotNull Mutator mutator); + + /** + * Returns a {@link java.util.concurrent.Future} containing the value to which the specified key is + * mapped, or {@code null} if this map contains no mapping for the key. This method behaves the same as + * {@link java.util.Map#get(java.lang.Object)} yet wraps result in a future a {@link + * java.util.concurrent.Future}. + * + * @see java.util.Map#get(java.lang.Object) + * @see java.util.concurrent.Future + */ + Future getLater(K key); + } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index a6eda3a45..54b9af5ab 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -37,6 +37,7 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.channels.SocketChannel; import java.util.*; +import java.util.concurrent.Future; import static java.nio.ByteBuffer.allocateDirect; import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; @@ -392,6 +393,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { return blockingFetch(writeEventAnSkip(LONG_SIZE)).readLong(); } + public synchronized Future getLater(K key) { + throw new UnsupportedOperationException(); + } + public synchronized V get(Object key) { final long sizeLocation = writeEventAnSkip(GET); @@ -951,8 +956,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private Bytes blockingFetch(long timeoutTime, long transactionId) throws IOException { - // bytes.buffer().position(0); - // the number of bytes in the response final int size = receive(SIZE_OF_SIZE, timeoutTime).readInt(); @@ -979,7 +982,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { ", does not match the expected transaction-id=" + transactionId); if (isException) { - Throwable throwable = (Throwable) bytes.readObject(); + Throwable throwable = (Throwable) bytes.readObject(); try { Field stackTrace = Throwable.class.getDeclaredField("stackTrace"); stackTrace.setAccessible(true); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index fb5bee879..88471c00c 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -40,6 +40,8 @@ import java.util.concurrent.atomic.AtomicBoolean; import static java.nio.channels.SelectionKey.*; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; +import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTIONID; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT; interface Work { @@ -1252,12 +1254,16 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable */ class StatelessServerConnector { - public static final int TRANSACTION_ID_OFFSET = AbstractChannelReplicator.SIZE_OF_SIZE + 1; + public static final int TRANSACTION_ID_OFFSET = SIZE_OF_SIZE + 1; public static final boolean MAP_SUPPORTS_BYTES = Boolean.valueOf(System.getProperty ("mapSupportsBytes")); private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class .getName()); public static final StatelessChronicleMap.EventId[] VALUES = StatelessChronicleMap.EventId.values(); + public static final int SIZE_OF_IS_EXCEPTION = 1; + public static final int HEADER_SIZE = SIZE_OF_SIZE + SIZE_OF_IS_EXCEPTION + SIZE_OF_TRANSACTIONID; + + private final KeyValueSerializer keyValueSerializer; private final VanillaChronicleMap map; private double maxEntrySizeBytes; @@ -1500,7 +1506,12 @@ class StatelessServerConnector { private Work sendException(Bytes writer, long sizeLocation, Throwable e) { + + // move the position to ignore any bytes written so far + writer.position(sizeLocation + HEADER_SIZE); + writeException(writer, e); + writeSizeAndFlags(sizeLocation, true, writer); return null; } @@ -1898,8 +1909,8 @@ class StatelessServerConnector { private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { long currentPosition = out.position(); final long size = out.position() - locationOfSize; - out.writeInt(locationOfSize, (int) size); - out.writeBoolean(locationOfSize + AbstractChannelReplicator.SIZE_OF_SIZE, isException); + out.writeInt(locationOfSize, (int) size); // size + out.writeBoolean(locationOfSize + SIZE_OF_SIZE, isException); // isException out.position(currentPosition); } diff --git a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java index d362d8369..31219660d 100755 --- a/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java @@ -42,7 +42,10 @@ import java.io.Serializable; import java.lang.reflect.Array; import java.nio.channels.FileChannel; import java.util.*; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import static java.lang.Long.numberOfTrailingZeros; import static java.lang.Math.max; @@ -80,7 +83,7 @@ class VanillaChronicleMap, final Alignment alignment; final int actualSegments; final long entriesPerSegment; - final MapEventListener> eventListener; + final transient MapEventListener> eventListener; // if set the ReturnsNull fields will cause some functions to return NULL // rather than as returning the Object can be expensive for something you probably don't use. final boolean putReturnsNull; @@ -314,7 +317,7 @@ class VanillaChronicleMap, void checkValue(Object value) { if (vClass != Void.class && !vClass.isInstance(value)) { throw new ClassCastException("Value must be a " + vClass.getName() + - " but was a "+ value.getClass()); + " but was a " + value.getClass()); } } @@ -381,6 +384,38 @@ class VanillaChronicleMap, } } + @Override + public Future getLater(K key) { + final V v = get(key); + return new Future() { + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + return false; + } + + @Override + public boolean isCancelled() { + return false; + } + + @Override + public boolean isDone() { + return true; + } + + @Override + public V get() throws InterruptedException, ExecutionException { + return v; + } + + @Override + public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + return v; + } + }; + } + @NotNull @Override public WriteContext acquireUsingLocked(@NotNull K key, @NotNull V usingValue) { diff --git a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java index 4740fc87b..4dee5e4d8 100644 --- a/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java +++ b/src/test/java/net/openhft/chronicle/map/ReplicationCheckingMap.java @@ -26,6 +26,7 @@ import java.util.Collection; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Future; /** * @author Rob Austin. @@ -254,6 +255,11 @@ public class ReplicationCheckingMap implements ChronicleMap { return map1.updateForKey(key, mutator); } + @Override + public Future getLater(K key) { + throw new UnsupportedOperationException(); + } + @Override public V acquireUsing(@NotNull K key, V usingValue) { return map1.acquireUsing(key, usingValue); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 50aba3975..16fd70559 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -233,7 +233,7 @@ public class StatelessClientTest { @Test - public void testStringKeyMap() throws IOException, InterruptedException { + public void testServerPutStringKeyMap() throws IOException, InterruptedException { try (ChronicleMap serverMap = ChronicleMapBuilder.of(String.class, Map .class) @@ -259,7 +259,7 @@ public class StatelessClientTest { @Test - public void testStringKeyMapPutIntoStatelesMap() throws IOException, InterruptedException { + public void testStringKeyMapPutIntoStatelessMap() throws IOException, InterruptedException { final Map data = new HashMap(); diff --git a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java index 88edb989b..4219bf767 100755 --- a/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/jrs166/map/StatelessChronicleMapTest.java @@ -31,6 +31,7 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.util.*; +import java.util.concurrent.Future; import static org.junit.Assert.*; @@ -176,6 +177,11 @@ public class StatelessChronicleMapTest extends JSR166TestCase { return d.updateForKey(key, mutator); } + @Override + public Future getLater(K key) { + return d.getLater(key); + } + @Override public V putIfAbsent(K key, V value) { return d.putIfAbsent(key, value);