diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 18074c7cb..9935f713f 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -27,6 +27,7 @@ import net.openhft.lang.io.NativeBytes; import net.openhft.lang.thread.NamedThreadFactory; import net.openhft.lang.threadlocal.ThreadLocalCopies; import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -60,6 +61,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private final byte[] connectionByte = new byte[1]; private final ByteBuffer connectionOutBuffer = ByteBuffer.wrap(connectionByte); private final String name; + @NotNull private final AbstractChronicleMapBuilder chronicleMapBuilder; private volatile ByteBuffer outBuffer; @@ -68,13 +70,19 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon private volatile ByteBuffer inBuffer; private volatile ByteBufferBytes inBytes; + @NotNull private final ReaderWithSize keyReaderWithSize; + @NotNull private final WriterWithSize keyWriterWithSize; + @NotNull private final ReaderWithSize valueReaderWithSize; + @NotNull private final WriterWithSize valueWriterWithSize; private volatile SocketChannel clientChannel; + @Nullable private CloseablesManager closeables; + @NotNull private final StatelessMapConfig config; private int maxEntrySize; private final Class kClass; @@ -117,10 +125,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon UPDATE_FOR_KEY } + @NotNull private AtomicLong transactionID = new AtomicLong(0); - StatelessChronicleMap(final StatelessMapConfig config, - final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { + StatelessChronicleMap(@NotNull final StatelessMapConfig config, + @NotNull final AbstractChronicleMapBuilder chronicleMapBuilder) throws IOException { this.chronicleMapBuilder = chronicleMapBuilder; this.config = config; keyReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.keyBuilder); @@ -148,9 +157,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } + @NotNull @Override public Future getLater(@NotNull final K key) { return lazyExecutorService().submit(new Callable() { + @Nullable @Override public V call() throws Exception { return StatelessChronicleMap.this.get(key); @@ -158,9 +169,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon }); } + @NotNull @Override public Future putLater(@NotNull final K key, @NotNull final V value) { return lazyExecutorService().submit(new Callable() { + @Nullable @Override public V call() throws Exception { return StatelessChronicleMap.this.put(key, value); @@ -312,6 +325,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon LOG.debug("Attached to a map with a remote identifier=" + remoteIdentifier); } + @NotNull public File file() { throw new UnsupportedOperationException(); } @@ -400,7 +414,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon * @return true if the contain the same data */ @Override - public boolean equals(Object object) { + public boolean equals(@Nullable Object object) { if (this == object) return true; if (object == null || object.getClass().isAssignableFrom(Map.class)) return false; @@ -421,6 +435,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchInt(HASH_CODE); } + @NotNull public String toString() { return fetchObject(String.class, TO_STRING); } @@ -434,6 +449,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchBooleanK(CONTAINS_KEY, (K) key); } + @NotNull private NullPointerException keyNotNullNPE() { return new NullPointerException("key can not be null"); } @@ -450,11 +466,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchObject(vClass, GET, (K) key); } + @NotNull public V getUsing(K key, V usingValue) { throw new UnsupportedOperationException("getUsing() is not supported for stateless " + "clients"); } + @NotNull public V acquireUsing(@NotNull K key, V usingValue) { throw new UnsupportedOperationException("acquireUsing() is not supported for stateless " + "clients"); @@ -485,22 +503,24 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return fetchObject(vClass, putReturnsNull ? PUT_WITHOUT_ACC : PUT, key, value); } - public R mapForKey(K key, @NotNull Function function) { + @Nullable + public R mapForKey(@Nullable K key, @NotNull Function function) { if (key == null) throw keyNotNullNPE(); return fetchObject(MAP_FOR_KEY, key, function); } + @Nullable @Override - public R updateForKey(K key, @NotNull Mutator mutator) { + public R updateForKey(@Nullable K key, @NotNull Mutator mutator) { if (key == null) throw keyNotNullNPE(); return fetchObject(UPDATE_FOR_KEY, key, mutator); } - private synchronized void writeEntriesForPutAll(Map map) { + private synchronized void writeEntriesForPutAll(@NotNull Map map) { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; @@ -850,7 +870,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private long writeEvent(StatelessChronicleMap.EventId event) { + private long writeEvent(@NotNull StatelessChronicleMap.EventId event) { outBuffer.clear(); outBytes.clear(); @@ -862,7 +882,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon /** * skips for the transactionid */ - private long writeEventAnSkip(EventId event) { + private long writeEventAnSkip(@NotNull EventId event) { final long sizeLocation = writeEvent(event); // skips for the transaction id @@ -917,6 +937,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon (value == null ? 0 : value.hashCode()); } + @NotNull public final String toString() { return getKey() + "=" + getValue(); } @@ -952,7 +973,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon outBytes.buffer().clear(); break; - } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { + } catch (@NotNull java.nio.channels.ClosedChannelException | ClosedConnectionException e) { checkTimeout(timeoutTime); clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress()); } @@ -1148,7 +1169,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private void send(final Bytes out, long timeoutTime) throws IOException { + private void send(@NotNull final Bytes out, long timeoutTime) throws IOException { outBuffer.limit((int) out.position()); outBuffer.position(0); @@ -1270,7 +1291,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - private void resizeToMessage(long start, IllegalArgumentException e) { + private void resizeToMessage(long start, @NotNull IllegalArgumentException e) { String message = e.getMessage(); if (message.startsWith(START_OF)) { String substring = message.substring("Attempt to write ".length(), message.length()); @@ -1298,6 +1319,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } + @Nullable private O readObject(long transactionId, long startTime) { long timeoutTime = startTime + this.config.timeoutMs(); @@ -1335,7 +1357,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } } - private long send(final EventId eventId, final long startTime) { + private long send(@NotNull final EventId eventId, final long startTime) { // send outBytesLock.lock(); try { @@ -1350,7 +1372,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return valueReaderWithSize.readNullable(in, copies); } - private boolean fetchBoolean(final EventId eventId, K key, V value) { + private boolean fetchBoolean(@NotNull final EventId eventId, K key, V value) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1373,7 +1395,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private boolean fetchBoolean(final EventId eventId, K key, V value1, V value2) { + private boolean fetchBoolean(@NotNull final EventId eventId, K key, V value1, V value2) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1397,7 +1419,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private boolean fetchBooleanV(final EventId eventId, V value) { + private boolean fetchBooleanV(@NotNull final EventId eventId, V value) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1415,7 +1437,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private boolean fetchBooleanK(final EventId eventId, K key) { + private boolean fetchBooleanK(@NotNull final EventId eventId, K key) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1433,7 +1455,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private long fetchLong(final EventId eventId) { + private long fetchLong(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1449,7 +1471,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return readLong(transactionId, startTime); } - private boolean fetchBoolean(final EventId eventId) { + private boolean fetchBoolean(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1465,7 +1487,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon return readBoolean(transactionId, startTime); } - private void fetchVoid(final EventId eventId) { + private void fetchVoid(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); @@ -1495,7 +1517,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private O fetchObject(final Class tClass, final EventId eventId) { + @Nullable + private O fetchObject(final Class tClass, @NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); long transactionId = send(eventId, startTime); @@ -1511,7 +1534,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private int fetchInt(final EventId eventId) { + private int fetchInt(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -1528,7 +1551,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private R fetchObject(Class rClass, final EventId eventId, K key, V value) { + @Nullable + private R fetchObject(Class rClass, @NotNull final EventId eventId, K key, V value) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1559,7 +1583,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon "supported"); } - private R fetchObject(Class rClass, final EventId eventId, K key, V value1, V value2) { + @Nullable + private R fetchObject(Class rClass, @NotNull final EventId eventId, K key, V value1, V value2) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1592,7 +1617,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon "supported"); } - private R fetchObject(Class rClass, final EventId eventId, K key) { + @Nullable + private R fetchObject(Class rClass, @NotNull final EventId eventId, K key) { final long startTime = System.currentTimeMillis(); ThreadLocalCopies copies; @@ -1622,7 +1648,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private boolean eventReturnsNull(EventId eventId) { + private boolean eventReturnsNull(@NotNull EventId eventId) { switch (eventId) { case PUT_ALL_WITHOUT_ACC: @@ -1654,7 +1680,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon } - private R fetchObject(final EventId eventId, K key, Object object) { + @Nullable + private R fetchObject(@NotNull final EventId eventId, K key, @NotNull Object object) { final long startTime = System.currentTimeMillis(); long transactionId;