diff --git a/src/main/java/net/openhft/chronicle/hash/hashing/LongHashFunction.java b/src/main/java/net/openhft/chronicle/hash/hashing/LongHashFunction.java index 864145ba6..4f57a6b72 100644 --- a/src/main/java/net/openhft/chronicle/hash/hashing/LongHashFunction.java +++ b/src/main/java/net/openhft/chronicle/hash/hashing/LongHashFunction.java @@ -63,6 +63,7 @@ public abstract class LongHashFunction implements Serializable { * CityHash64 algorithm, version 1.1 without seed values. This implementation produce * equal results for equal input on platforms with different {@link ByteOrder}, but is slower * on big-endian platforms than on little-endian. + * @return an instance of LongHashFunction * * @see #city_1_1(long) * @see #city_1_1(long, long) @@ -91,7 +92,7 @@ public abstract class LongHashFunction implements Serializable { * CityHash64 algorithm, version 1.1 using the two given seed values. This implementation * produce equal results for equal input on platforms with different {@link ByteOrder}, but * is slower on big-endian platforms than on little-endian. - * + * @return an instance of LongHashFunction * @see #city_1_1() * @see #city_1_1(long) */ @@ -136,6 +137,8 @@ public abstract class LongHashFunction implements Serializable { * value is interpreted in {@linkplain ByteOrder#nativeOrder() native} byte order. For example, * the result of {@code hashLong(v)} call is identical to the result of * {@code hashLongs(new long[] {v})} call for any {@code long} value. + * @param input the input + * @return a hashLong */ public abstract long hashLong(long input); @@ -145,6 +148,8 @@ public abstract class LongHashFunction implements Serializable { * value is interpreted in {@linkplain ByteOrder#nativeOrder() native} byte order. For example, * the result of {@code hashInt(v)} call is identical to the result of * {@code hashInts(new int[] {v})} call for any {@code int} value. + * @param input the input + * @return a hashLong */ public abstract long hashInt(int input); @@ -156,6 +161,9 @@ public abstract class LongHashFunction implements Serializable { * {@code hashShorts(new short[] {v})} call for any {@code short} value. * As a consequence, {@code hashShort(v)} call produce always the same result as {@code * hashChar((char) v)}. + * + * @param input the input + * @return a hashShort */ public abstract long hashShort(short input); @@ -175,12 +183,16 @@ public abstract class LongHashFunction implements Serializable { * {@code LongHashFunction} methods that accept sequences of bytes. For example, the result of * {@code hashByte(v)} call is identical to the result of * {@code hashBytes(new byte[] {v})} call for any {@code byte} value. + * + * @param input the input + * @return a hashByte */ public abstract long hashByte(byte input); /** * Returns the hash code for the empty (zero-length) bytes sequence, * for example {@code hashBytes(new byte[0])}. + * @return a hash */ public abstract long hashVoid(); diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index b5a1b8816..2798b6000 100755 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -256,7 +256,7 @@ public final class ChronicleMapBuilder implements * encoded (and each character takes 2 bytes on-heap), because default off-heap {@link String} * encoding is UTF-8 in {@code ChronicleMap}.) * - * @param averageKeySize + * @param averageKeySize the average size of the key * @throws IllegalStateException {@inheritDoc} * @throws IllegalArgumentException {@inheritDoc} * @see #constantKeySizeBySample(Object) diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapStatelessClientBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapStatelessClientBuilder.java index 376ce0a5b..9c08e0c29 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapStatelessClientBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapStatelessClientBuilder.java @@ -48,12 +48,13 @@ public final class ChronicleMapStatelessClientBuilder implements // -- the second is really shorter. /** - * Equivalent of {@code ChronicleMapStatelessClientBuilder.of(serverAddress).create()}. + * Equivalent of {@code ChronicleMapStatelessClientBuilder.of(serverAddress).create()}. * * @param serverAddress address of the server map * @param key type of the map * @param value type of the map * @return stateless client of the ChronicleMap of server + * @throws IOException if it not possible to connect to the {@code serverAddress} */ public static ChronicleMap createClientOf(InetSocketAddress serverAddress) throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 4309156a6..ac6873685 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -34,7 +34,8 @@ import java.util.function.Consumer; import java.util.function.Predicate; import static java.util.Collections.emptyList; -import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId.*; +import static net.openhft.chronicle.map.MapWireHandler.EventId; +import static net.openhft.chronicle.map.MapWireHandler.EventId.*; /** @@ -136,7 +137,7 @@ class ClientWiredStatelessChronicleMap if (key == null || value == null) throw new NullPointerException(); - return proxyReturnObject(PUT_IF_ABSENT.toString(), key, value, vClass); + return proxyReturnObject(putIfAbsent, key, value, vClass); } @SuppressWarnings("NullableProblems") @@ -144,23 +145,21 @@ class ClientWiredStatelessChronicleMap if (key == null) throw new NullPointerException(); - return value != null && - proxyReturnBoolean(REMOVE_WITH_VALUE.toString(), (K) key, (V) value); + return value != null && proxyReturnBoolean(removeWithValue.toString(), (K) key, (V) value); } @SuppressWarnings("NullableProblems") public boolean replace(K key, V oldValue, V newValue) { if (key == null || oldValue == null || newValue == null) throw new NullPointerException(); - return proxyReturnBoolean( - REPLACE_WITH_OLD_AND_NEW_VALUE.toString(), key, oldValue, newValue); + return proxyReturnBoolean(replaceWithOldAndNewValue.toString(), key, oldValue, newValue); } @SuppressWarnings("NullableProblems") public V replace(K key, V value) { if (key == null || value == null) throw new NullPointerException(); - return proxyReturnObject(REPLACE.toString(), key, value, vClass); + return proxyReturnObject(replace, key, value, vClass); } public int size() { @@ -195,7 +194,7 @@ class ClientWiredStatelessChronicleMap @Override public int hashCode() { - return proxyReturnInt(HASH_CODE.toString()); + return proxyReturnInt(hashCode.toString()); } @NotNull @@ -205,15 +204,15 @@ class ClientWiredStatelessChronicleMap @NotNull public String serverPersistedDataVersion() { - return hub.proxyReturnString(PERSISTED_DATA_VERSION.toString(), channelID); + return hub.proxyReturnString(persistedDataVersion, channelID); } public boolean isEmpty() { - return proxyReturnBoolean(IS_EMPTY.toString()); + return proxyReturnBoolean(isEmpty.toString()); } public boolean containsKey(Object key) { - return proxyReturnBooleanK(CONTAINS_KEY.toString(), (K) key); + return proxyReturnBooleanK(containsKey, (K) key); } @NotNull @@ -222,11 +221,11 @@ class ClientWiredStatelessChronicleMap } public boolean containsValue(Object value) { - return proxyReturnBooleanV(CONTAINS_VALUE.toString(), (V) value); + return proxyReturnBooleanV(containsValue, (V) value); } public long longSize() { - return proxyReturnLong(LONG_SIZE.toString()); + return proxyReturnLong(longSize.toString()); } @Override @@ -235,7 +234,7 @@ class ClientWiredStatelessChronicleMap } public V get(Object key) { - return proxyReturnObject(vClass, GET.toString(), (K) key); + return proxyReturnObject(vClass, get, (K) key); } @Nullable @@ -258,16 +257,15 @@ class ClientWiredStatelessChronicleMap public V remove(Object key) { if (key == null) throw keyNotNullNPE(); - return proxyReturnObject(vClass, - removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); + return proxyReturnObject(vClass, removeReturnsNull ? removeWithoutAcc : remove, (K) key); } @Override public void createChannel(short channelID) { proxyReturnVoid( () -> { - writeField(Fields.METHOD_NAME, "CREATE_CHANNEL"); - hub.outWire().write(Fields.ARG_1).int16(channelID); + hub.outWire().write(Fields.methodName).text(createChannel.toString()); + hub.outWire().write(Fields.arg1).int16(channelID); }); } @@ -276,18 +274,14 @@ class ClientWiredStatelessChronicleMap final long startTime = System.currentTimeMillis(); long transactionId = hub.nextUniqueTransaction(startTime); - Set> entries = map.entrySet(); Iterator> iterator = entries.iterator(); - boolean hasNext; - - do { - - hasNext = iterator.hasNext(); - + OUTER: + while (iterator.hasNext()) { hub.outBytesLock().lock(); try { + assert hub.outBytesLock().isHeldByCurrentThread(); assert !hub.inBytesLock().isHeldByCurrentThread(); @@ -296,28 +290,32 @@ class ClientWiredStatelessChronicleMap assert hub.outBytesLock().isHeldByCurrentThread(); hub.markSize(wire); hub.startTime(startTime); + wire.write(Fields.type).text("MAP"); + wire.write(Fields.transactionId).int64(transactionId); + wire.write(Fields.timeStamp).int64(startTime); + wire.write(Fields.channelId).int16(channelID); + hub.outWire().write(Fields.methodName).text(putAll.toString()); - wire.write(Fields.TRANSACTION_ID).int64(transactionId); - wire.write(Fields.TIME_STAMP).int64(startTime); - wire.write(Fields.CHANNEL_ID).int16(channelID); + while (iterator.hasNext()) { - writeField(Fields.METHOD_NAME, "PUT_ALL"); - - hub.outWire().write(Fields.HAS_NEXT).bool(hasNext); - - if (hasNext) { Map.Entry e = iterator.next(); - writeField(Fields.ARG_1, e.getKey()); - writeField(Fields.ARG_2, e.getValue()); + + + hub.outWire().write(Fields.hasNext).bool(iterator.hasNext()); + writeField(Fields.arg1, e.getKey()); + writeField(Fields.arg2, e.getValue()); + + if (wire.bytes().remaining() < wire.bytes().capacity() * 0.5) + break OUTER; } - hub.writeSocket(hub.outWire()); - } finally { + hub.writeSocket(hub.outWire()); hub.outBytesLock().unlock(); } - } while (hasNext); + } + // wait for the transaction id to be received, this will only be received once the // last chunk has been processed @@ -328,8 +326,7 @@ class ClientWiredStatelessChronicleMap public V put(K key, V value) { if (key == null || value == null) throw new NullPointerException(); - return proxyReturnObject(putReturnsNull ? PUT_WITHOUT_ACC.toString() : PUT.toString(), - key, value, vClass); + return proxyReturnObject(putReturnsNull ? putWithoutAcc : put, key, value, vClass); } @Nullable @@ -344,14 +341,14 @@ class ClientWiredStatelessChronicleMap } public void clear() { - proxyReturnVoid(() -> writeField(Fields.METHOD_NAME, CLEAR.toString())); + proxyReturnVoid(() -> writeField(Fields.methodName, clear.toString())); } @NotNull public Set keySet() { final Set usingCollection = new HashSet(); - readChunked(kClass, "KEY_SET", usingCollection); + readChunked(kClass, keySet, usingCollection); return usingCollection; } @@ -359,11 +356,11 @@ class ClientWiredStatelessChronicleMap @NotNull public Collection values() { final List usingCollection = new ArrayList<>(); - return readChunked(vClass, "VALUES", usingCollection); + return readChunked(vClass, values, usingCollection); } private > Collection readChunked( - Class vClass1, String values, A usingCollection) { + Class vClass1, EventId values, A usingCollection) { final long startTime = System.currentTimeMillis(); // send @@ -380,14 +377,14 @@ class ClientWiredStatelessChronicleMap final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); while (wireIn.bytes().remaining() > 0) { - boolean bool = wireIn.read(Fields.HAS_NEXT).bool(); + boolean bool = wireIn.read(Fields.hasNext).bool(); if (bool) { - usingCollection.add(readObject(Fields.RESULT, wireIn, null, vClass1)); + usingCollection.add(readObject(Fields.result, wireIn, null, vClass1)); } else break OUTER; // todo process the exception - boolean isException = wireIn.read(Fields.IS_EXCEPTION).bool(); + boolean isException = wireIn.read(Fields.isException).bool(); } } finally { @@ -406,7 +403,7 @@ class ClientWiredStatelessChronicleMap final Map result = new HashMap(); final long startTime = System.currentTimeMillis(); // send - final long transactionId = proxySend("ENTRY_SET", startTime); + final long transactionId = proxySend(entrySet, startTime); assert !hub.outBytesLock().isHeldByCurrentThread(); final long timeoutTime = startTime + hub.timeoutMs; @@ -419,18 +416,18 @@ class ClientWiredStatelessChronicleMap final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); while (wireIn.bytes().remaining() > 0) { - boolean bool = wireIn.read(Fields.HAS_NEXT).bool(); + boolean bool = wireIn.read(Fields.hasNext).bool(); if (bool) { result.put( - readK(Fields.RESULT_KEY, wireIn, null), - readV(Fields.RESULT_VALUE, wireIn, null)); + readK(Fields.resultKey, wireIn, null), + readV(Fields.resultValue, wireIn, null)); } else break OUTER; if (wireIn.bytes().remaining() > 0) { // todo process the exception - boolean isException = wireIn.read(Fields.IS_EXCEPTION).bool(); + boolean isException = wireIn.read(Fields.isException).bool(); } } @@ -456,7 +453,7 @@ class ClientWiredStatelessChronicleMap // receive hub.inBytesLock().lock(); try { - return hub.proxyReply(timeoutTime, transactionId).read(Fields.RESULT).int64(); + return hub.proxyReply(timeoutTime, transactionId).read(Fields.result).int64(); } finally { hub.inBytesLock().unlock(); } @@ -503,7 +500,7 @@ class ClientWiredStatelessChronicleMap final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(Fields.RESULT_IS_NULL).bool()) + if (wireIn.read(Fields.resultIsNull).bool()) return null; return readV(argName, wireIn, usingValue); @@ -589,7 +586,7 @@ class ClientWiredStatelessChronicleMap hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - return wireIn.read(Fields.RESULT).bool(); + return wireIn.read(Fields.result).bool(); } finally { hub.inBytesLock().unlock(); } @@ -604,7 +601,7 @@ class ClientWiredStatelessChronicleMap hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - return wireIn.read(Fields.RESULT).int32(); + return wireIn.read(Fields.result).int32(); } finally { hub.inBytesLock().unlock(); } @@ -619,9 +616,9 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); - writeField(Fields.ARG_1, key); - writeField(Fields.ARG_2, value); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + writeField(Fields.arg1, key); + writeField(Fields.arg2, value); hub.writeSocket(hub.outWire()); } finally { @@ -641,10 +638,10 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); - writeField(Fields.ARG_1, key); - writeField(Fields.ARG_2, value1); - writeField(Fields.ARG_3, value2); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + writeField(Fields.arg1, key); + writeField(Fields.arg2, value1); + writeField(Fields.arg3, value2); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -655,13 +652,13 @@ class ClientWiredStatelessChronicleMap } @SuppressWarnings("SameParameterValue") - private boolean proxyReturnBooleanV(@NotNull final String methodName, V value) { + private boolean proxyReturnBooleanV(@NotNull final EventId methodName, V value) { final long startTime = System.currentTimeMillis(); return readBoolean(proxySend(methodName, startTime, value), startTime); } @SuppressWarnings("SameParameterValue") - private boolean proxyReturnBooleanK(@NotNull final String methodName, K key) { + private boolean proxyReturnBooleanK(@NotNull final EventId methodName, K key) { final long startTime = System.currentTimeMillis(); return readBoolean(proxySend(methodName, startTime, key), startTime); @@ -676,7 +673,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); + hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -694,7 +691,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); + hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -732,7 +729,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); + hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -744,7 +741,7 @@ class ClientWiredStatelessChronicleMap @Nullable private R proxyReturnObject( - @NotNull final String methodName, K key, V value, Class resultType) { + @NotNull final EventId methodName, K key, V value, Class resultType) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -755,9 +752,9 @@ class ClientWiredStatelessChronicleMap assert !hub.inBytesLock().isHeldByCurrentThread(); transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); - writeField(Fields.ARG_1, key); - writeField(Fields.ARG_2, value); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + writeField(Fields.arg1, key); + writeField(Fields.arg2, value); hub.writeSocket(hub.outWire()); } finally { @@ -768,7 +765,7 @@ class ClientWiredStatelessChronicleMap return null; if (resultType == vClass) - return (R) readValue(Fields.RESULT, transactionId, startTime, null); + return (R) readValue(Fields.result, transactionId, startTime, null); else throw new UnsupportedOperationException("class of type class=" + resultType + @@ -776,7 +773,8 @@ class ClientWiredStatelessChronicleMap } @Nullable - private R proxyReturnObject(Class rClass, @NotNull final String methodName, Object key) { + private R proxyReturnObject( + Class rClass, @NotNull final EventId methodName, Object key) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -787,19 +785,19 @@ class ClientWiredStatelessChronicleMap return null; if (rClass == vClass) - return (R) readValue(Fields.RESULT, transactionId, startTime, null); + return (R) readValue(Fields.result, transactionId, startTime, null); else throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + "supported"); } - private long proxySend(String methodName, long startTime, Object arg1) { + private long proxySend(EventId methodName, long startTime, Object arg1) { long transactionId; hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); - writeField(Fields.ARG_1, arg1); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + writeField(Fields.arg1, arg1); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -808,12 +806,12 @@ class ClientWiredStatelessChronicleMap } - private long proxySend(String methodName, long startTime) { + private long proxySend(EventId methodName, long startTime) { long transactionId; hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); + hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -821,12 +819,12 @@ class ClientWiredStatelessChronicleMap return transactionId; } - private boolean eventReturnsNull(@NotNull String methodName) { + private boolean eventReturnsNull(@NotNull EventId methodName) { switch (methodName) { - case "PUT_ALL_WITHOUT_ACC": - case "PUT_WITHOUT_ACC": - case "REMOVE_WITHOUT_ACC": + case putAllWithoutAcc: + case putWithoutAcc: + case removeWithoutAcc: return true; default: return false; @@ -847,38 +845,39 @@ class ClientWiredStatelessChronicleMap } - static enum EventId { +/* static enum EventId { + createChannel, HEARTBEAT, STATEFUL_UPDATE, - LONG_SIZE, - SIZE, - IS_EMPTY, - CONTAINS_KEY, - CONTAINS_VALUE, - GET, - PUT, - PUT_WITHOUT_ACC, - REMOVE, - REMOVE_WITHOUT_ACC, - CLEAR, + longSize, + size, + isEmpty, + containsKey, + containsValue, + get, + put, + putWithoutAcc, + remove, + removeWithoutAcc, + clear, KEY_SET, VALUES, - ENTRY_SET, - REPLACE, - REPLACE_WITH_OLD_AND_NEW_VALUE, - PUT_IF_ABSENT, - REMOVE_WITH_VALUE, - TO_STRING, - APPLICATION_VERSION, - PERSISTED_DATA_VERSION, - PUT_ALL, - PUT_ALL_WITHOUT_ACC, - HASH_CODE, - MAP_FOR_KEY, - PUT_MAPPED, - KEY_BUILDER, - VALUE_BUILDER - } + entrySet, + replace, + replaceWithOldAndNewValue, + putIfAbsent, + removeWithValue, + toString, + applicationVersion, + persistedDataVersion, + putAll, + putAllWithoutAcc, + hashCode, + mapForKey, + putMapped, + keyBuilder, + valueBuilder + }*/ class Entry implements Map.Entry { diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index e8d849204..0fa15984f 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -23,6 +23,7 @@ import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.RemoteCallTimeoutException; import net.openhft.chronicle.hash.impl.util.BuildVersion; import net.openhft.chronicle.hash.impl.util.CloseablesManager; +import net.openhft.chronicle.map.MapWireHandler.EventId; import net.openhft.chronicle.network2.event.EventGroup; import net.openhft.chronicle.wire.TextWire; import net.openhft.chronicle.wire.Wire; @@ -38,7 +39,7 @@ import java.nio.channels.SocketChannel; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; -import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId.APPLICATION_VERSION; +import static net.openhft.chronicle.map.MapWireHandler.EventId.applicationVersion; import static net.openhft.chronicle.map.MapWireHandler.SIZE_OF_SIZE; import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; @@ -236,7 +237,7 @@ public class ClientWiredStatelessTcpConnectionHub { @NotNull public String serverApplicationVersion(short channelID) { TextWire wire = new TextWire(Bytes.elasticByteBuffer()); - String result = proxyReturnString(APPLICATION_VERSION.toString(), channelID, wire); + String result = proxyReturnString(applicationVersion, channelID, wire); return (result == null) ? "" : result; } @@ -310,9 +311,9 @@ public class ClientWiredStatelessTcpConnectionHub { final Wire wire = proxyReplyThrowable(timeoutTime, transactionId); - // handle an exception if the message contains the IS_EXCEPTION field - if (wire.read(() -> "IS_EXCEPTION").bool()) { - final String text = wire.read(() -> "EXCEPTION").text(); + // handle an exception if the message contains the isException field + if (wire.read(() -> "isException").bool()) { + final String text = wire.read(() -> "exception").text(); throw new RuntimeException(text); } return wire; @@ -367,7 +368,7 @@ public class ClientWiredStatelessTcpConnectionHub { System.out.println("\n--------------------------------\n" + "client read:\n\n" + Bytes.toDebugString(intWire.bytes())); - long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64(); + long transactionId0 = intWire.read(MapWireHandlerBuilder.Fields.transactionId).int64(); // check the transaction id is reasonable @@ -556,7 +557,7 @@ public class ClientWiredStatelessTcpConnectionHub { } - private long proxySend(@NotNull final String methodName, + private long proxySend(@NotNull final EventId methodName, final long startTime, short channelID, @NotNull final Wire wire) { @@ -568,7 +569,7 @@ public class ClientWiredStatelessTcpConnectionHub { outBytesLock().lock(); try { long transactionId = writeHeader(startTime, channelID, wire); - wire.write(() -> "METHOD_NAME").text(methodName); + wire.write(MapWireHandlerBuilder.Fields.methodName).text(methodName.toString()); writeSocket(wire); return transactionId; } finally { @@ -578,13 +579,13 @@ public class ClientWiredStatelessTcpConnectionHub { @SuppressWarnings("SameParameterValue") @Nullable - String proxyReturnString(@NotNull final String messageId, short channelID) { + String proxyReturnString(@NotNull final EventId messageId, short channelID) { return proxyReturnString(messageId, channelID, outWire); } @SuppressWarnings("SameParameterValue") @Nullable - String proxyReturnString(@NotNull final String messageId, short channelID, Wire outWire) { + String proxyReturnString(@NotNull final EventId messageId, short channelID, Wire outWire) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -600,7 +601,7 @@ public class ClientWiredStatelessTcpConnectionHub { // receive inBytesLock().lock(); try { - return proxyReply(timeoutTime, transactionId).read(() -> "RESULT").text(); + return proxyReply(timeoutTime, transactionId).read(() -> "result").text(); } catch (Exception e) { throw new RuntimeException(e); } finally { @@ -621,10 +622,10 @@ public class ClientWiredStatelessTcpConnectionHub { startTime(startTime); long transactionId = nextUniqueTransaction(startTime); - wire.write(TYPE).text("MAP"); - wire.write(TRANSACTION_ID).int64(transactionId); - wire.write(TIME_STAMP).int64(startTime); - wire.write(CHANNEL_ID).int16(channelID); + wire.write(type).text("MAP"); + wire.write(MapWireHandlerBuilder.Fields.transactionId).int64(transactionId); + wire.write(timeStamp).int64(startTime); + wire.write(channelId).int16(channelID); return transactionId; } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 09cbd15e2..456fdc2d1 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -26,12 +26,14 @@ import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; import net.openhft.chronicle.hash.impl.util.BuildVersion; import net.openhft.chronicle.hash.replication.ReplicationHub; +import net.openhft.chronicle.map.MapWireHandlerBuilder.Fields; import net.openhft.chronicle.network2.WireHandler; import net.openhft.chronicle.network2.event.EventGroup; import net.openhft.chronicle.network2.event.WireHandlers; import net.openhft.chronicle.wire.ValueIn; import net.openhft.chronicle.wire.Wire; import net.openhft.chronicle.wire.WireKey; +import net.openhft.chronicle.wire.Wires; import net.openhft.lang.io.DirectStore; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -42,15 +44,13 @@ import java.io.PrintWriter; import java.io.StreamCorruptedException; import java.io.StringWriter; import java.nio.BufferOverflowException; -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.concurrent.Callable; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import static net.openhft.chronicle.map.MapWireHandler.EventId.*; import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; /** @@ -58,13 +58,13 @@ import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; */ class MapWireHandler implements WireHandler, Consumer { - private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); - public static final int SIZE_OF_SIZE = 2; + public static final int SIZE_OF_SIZE = 2; + private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); + final Map incompleteWork = new HashMap(); private final ArrayList bytesChronicleMaps = new ArrayList<>(); @NotNull - private final StringBuilder methodName = new StringBuilder(); private final Supplier>> mapFactory; private Wire inWire = null; private Wire outWire = null; @@ -72,19 +72,10 @@ class MapWireHandler implements WireHandler, Consumer { @Override public void accept(Iterator iterator) { - outWire.write(RESULT); + outWire.write(result); outWire.bytes().write(iterator.next()); } }; - - private WireHandlers publishLater; - - @Override - public void accept(WireHandlers wireHandlers) { - this.publishLater = wireHandlers; - } - - private final Consumer writeEntry = new Consumer>>() { @Override @@ -92,13 +83,14 @@ class MapWireHandler implements WireHandler, Consumer { final Map.Entry entry = iterator.next(); - outWire.write(RESULT_KEY); + outWire.write(resultKey); outWire.bytes().write(entry.getKey()); - outWire.write(RESULT_VALUE); + outWire.write(resultValue); outWire.bytes().write(entry.getValue()); } }; + private WireHandlers publishLater; private byte localIdentifier; private long timestamp; private short channelId; @@ -106,15 +98,12 @@ class MapWireHandler implements WireHandler, Consumer { private ReplicationHub hub; private byte remoteIdentifier; - public MapWireHandler( @NotNull Supplier>> mapFactory, @NotNull ReplicationHub hub, byte localIdentifier, @NotNull List channels) { this(mapFactory, hub); this.channels = channels; this.localIdentifier = localIdentifier; - - } public MapWireHandler(Supplier>> mapFactory, @@ -123,19 +112,11 @@ class MapWireHandler implements WireHandler, Consumer { this.hub = hub; } - @Override - public void process(Wire in, Wire out) throws StreamCorruptedException { - try { - this.inWire = in; - this.outWire = out; - onEvent(); - } catch (Exception e) { - LOG.error("", e); - } + public void accept(WireHandlers wireHandlers) { + this.publishLater = wireHandlers; } - private void writeChunked( long transactionId, @NotNull final Function function, @NotNull final Consumer c) throws StreamCorruptedException { @@ -148,7 +129,7 @@ class MapWireHandler implements WireHandler, Consumer { @Override public void process(Wire in, Wire out) throws StreamCorruptedException { - outWire.write(TRANSACTION_ID).int64(transactionId); + outWire.write(Fields.transactionId).int64(transactionId); // this allows us to write more data than the buffer will allow for (; ; ) { @@ -157,11 +138,10 @@ class MapWireHandler implements WireHandler, Consumer { write(map -> { - outWire.write(HAS_NEXT).bool(hasNext); + outWire.write(Fields.hasNext).bool(hasNext); if (hasNext) c.accept(iterator); - }); if (!hasNext) @@ -182,126 +162,119 @@ class MapWireHandler implements WireHandler, Consumer { } - /* private void writeEntryChunked(long transactionId, - @NotNull final Function function, - @NotNull final Consumer>> c) { - - final BytesChronicleMap m = bytesMap(channelId); - final Iterator> iterator = function.apply(m); - - // this allows us to write more data than the buffer will allow - for (; ; ) { - - // each chunk has its own transaction-id - outWire.write(TRANSACTION_ID).int64(transactionId); - - write(map -> { - - boolean hasNext = iterator.hasNext(); - outWire.write(HAS_NEXT).bool(hasNext); - - if (!hasNext) - return; - - c.accept(iterator); - - }); - + @Override + public void process(Wire in, Wire out) throws StreamCorruptedException { + try { + this.inWire = in; + this.outWire = out; + onEvent(); + } catch (Exception e) { + LOG.error("", e); } - - }*/ + } @SuppressWarnings("UnusedReturnValue") void onEvent() throws StreamCorruptedException { // it is assumed by this point that the buffer has all the bytes in it for this message - long transactionId = inWire.read(TRANSACTION_ID).int64(); - timestamp = inWire.read(TIME_STAMP).int64(); - channelId = inWire.read(CHANNEL_ID).int16(); - inWire.read(METHOD_NAME).text(methodName); + long transactionId = inWire.read(Fields.transactionId).int64(); + timestamp = inWire.read(timeStamp).int64(); + channelId = inWire.read(Fields.channelId).int16(); - if ("PUT_WITHOUT_ACC".contentEquals(methodName)) { - writeVoid(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); - // todo bytesMap.put(reader, reader, timestamp, identifier()); - bytesMap.put(reader, reader); - }); - return; + final StringBuilder methodName = Wires.acquireStringBuilder(); + inWire.read(Fields.methodName).text(methodName); + + if (!incompleteWork.isEmpty()) { + Runnable runnable = incompleteWork.get(transactionId); + if (runnable != null) { + runnable.run(); + return; + } } - - if ("KEY_SET".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.keySet().iterator(), writeElement); - return; - } - - if ("VALUES".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.delegate.values().iterator(), writeElement); - return; - } - - if ("ENTRY_SET".contentEquals(methodName)) { - writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry); - return; - } - - // write the transaction id - outWire.write(() -> "TRANSACTION_ID").int64(transactionId); - try { + if (putWithoutAcc.contentEquals(methodName)) { + writeVoid(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, arg1, arg2); + // todo bytesMap.put(reader, reader, timestamp, identifier()); + bytesMap.put(reader, reader); + }); + return; + } - if ("PUT_ALL".contentEquals(methodName)) { + if (keySet.contentEquals(methodName)) { + writeChunked(transactionId, map -> map.keySet().iterator(), writeElement); + return; + } + + if (values.contentEquals(methodName)) { + writeChunked(transactionId, map -> map.delegate.values().iterator(), writeElement); + return; + } + + if (entrySet.contentEquals(methodName)) { + writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry); + return; + } + + if (putAll.contentEquals(methodName)) { putAll(transactionId); return; } - if ("CREATE_CHANNEL".contentEquals(methodName)) { + // write the transaction id + outWire.write(Fields.transactionId).int64(transactionId); + + + if (createChannel.contentEquals(methodName)) { writeVoid(() -> { - short channelId1 = inWire.read(ARG_1).int16(); + short channelId1 = inWire.read(arg1).int16(); mapFactory.get().replicatedViaChannel(hub.createChannel(channelId1)).create(); return null; }); return; } - - if ("REMOTE_IDENTIFIER".contentEquals(methodName)) { - this.remoteIdentifier = inWire.read(RESULT).int8(); + if (EventId.remoteIdentifier.contentEquals(methodName)) { + this.remoteIdentifier = inWire.read(result).int8(); return; } - if ("LONG_SIZE".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).int64(b.longSize())); + if (longSize.contentEquals(methodName)) { + write(b -> outWire.write(result).int64(b.longSize())); return; } - if ("IS_EMPTY".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).bool(b.isEmpty())); + if (isEmpty.contentEquals(methodName)) { + write(b -> outWire.write(result).bool(b.isEmpty())); return; } - if ("CONTAINS_KEY".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).bool( - b.delegate.containsKey(toByteArray(inWire, ARG_1)))); + if (containsKey.contentEquals(methodName)) { + // todo remove the toByteArray(..) + write(b -> outWire.write(result) + .bool(b.delegate.containsKey(toByteArray(inWire, arg1)))); return; } - if ("CONTAINS_VALUE".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).bool( - b.delegate.containsValue(toByteArray(inWire, ARG_1)))); + if (containsValue.contentEquals(methodName)) { + // todo remove the toByteArray(..) + write(b -> outWire.write(result) + .bool(b.delegate.containsValue(toByteArray(inWire, arg1)))); return; } - if ("GET".contentEquals(methodName)) { - writeValueUsingDelegate(map -> map.get(toByteArray(inWire, ARG_1))); + if (get.contentEquals(methodName)) { + // todo remove the toByteArray(..) + writeValueUsingDelegate(map -> map.get(toByteArray(inWire, arg1))); return; } - if ("PUT".contentEquals(methodName)) { + if (put.contentEquals(methodName)) { writeValue(b -> { final net.openhft.lang.io.Bytes reader = - MapWireHandler.this.toReader(inWire, ARG_1, ARG_2); + MapWireHandler.this.toReader(inWire, arg1, arg2); // todo call return b.put(reader, reader, timestamp, identifier()); return b.put(reader, reader); @@ -310,34 +283,33 @@ class MapWireHandler implements WireHandler, Consumer { return; } - if ("REMOVE".contentEquals(methodName)) { - writeValue(b -> b.remove(toReader(inWire, ARG_1))); + if (remove.contentEquals(methodName)) { + writeValue(b -> b.remove(toReader(inWire, arg1))); return; } - if ("CLEAR".contentEquals(methodName)) { + if (clear.contentEquals(methodName)) { writeVoid(BytesChronicleMap::clear); return; } - if ("REPLACE".contentEquals(methodName)) { + if (replace.contentEquals(methodName)) { write(bytesMap -> { // todo fix this this is a hack to get to work for now. - // todo may use somehting like : + // todo may use something like : // todo bytesMap.replace(reader, reader, timestamp, identifier()); - - VanillaChronicleMap map = bytesMap.delegate; + final VanillaChronicleMap map = bytesMap.delegate; byte[] result = (byte[]) map.replace( - toByteArray(inWire, ARG_1), - toByteArray(inWire, ARG_2)); + toByteArray(inWire, arg1), + toByteArray(inWire, arg2)); boolean isNull = result == null || result.length == 0; - outWire.write(() -> "RESULT_IS_NULL").bool(isNull); + outWire.write(resultIsNull).bool(isNull); if (!isNull) { - outWire.write(() -> "RESULT"); + outWire.write(Fields.result); outWire.bytes().write(result); } @@ -345,50 +317,56 @@ class MapWireHandler implements WireHandler, Consumer { return; } - if ("REPLACE_WITH_OLD_AND_NEW_VALUE".contentEquals(methodName)) { + if (replaceWithOldAndNewValue.contentEquals(methodName)) { write(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2, ARG_3); + final net.openhft.lang.io.Bytes reader = toReader(inWire, arg1, arg2, arg3); boolean result = bytesMap.replace(reader, reader, reader); - outWire.write(RESULT).bool(result); + outWire.write(Fields.result).bool(result); }); return; } - if ("PUT_IF_ABSENT".contentEquals(methodName)) { - writeValue(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); - // todo call bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); - return bytesMap.putIfAbsent(reader, reader); + if (putIfAbsent.contentEquals(methodName)) { + + + // todo call bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); + writeValueFromBytes(b -> { + // todo remove the toByteArray(..) + byte[] key = MapWireHandler.this.toByteArray(inWire, arg1); + byte[] value = MapWireHandler.this.toByteArray(inWire, arg2); + return ((Map) b.delegate).putIfAbsent(key, value); }); + + return; } - if ("REMOVE_WITH_VALUE".contentEquals(methodName)) { + if (removeWithValue.contentEquals(methodName)) { write(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); - // todo call outWire.write(RESULT) + final net.openhft.lang.io.Bytes reader = toReader(inWire, arg1, arg2); + // todo call outWire.write(result) // .bool(bytesMap.remove(reader, reader, timestamp, identifier())); - outWire.write(RESULT).bool(bytesMap.remove(reader, reader)); + outWire.write(result).bool(bytesMap.remove(reader, reader)); }); return; } - if ("APPLICATION_VERSION".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).text(applicationVersion())); + if (applicationVersion.contentEquals(methodName)) { + write(b -> outWire.write(result).text(applicationVersion())); return; } - if ("PERSISTED_DATA_VERSION".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).text(persistedDataVersion())); + if (persistedDataVersion.contentEquals(methodName)) { + write(b -> outWire.write(result).text(persistedDataVersion())); return; } - if ("HASH_CODE".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).int32(b.hashCode())); + if (hashCode.contentEquals(methodName)) { + write(b -> outWire.write(result).int32(b.hashCode())); return; } @@ -399,14 +377,21 @@ class MapWireHandler implements WireHandler, Consumer { // if (len > 4) if (EventGroup.IS_DEBUG) { long len = outWire.bytes().position() - SIZE_OF_SIZE; - System.out.println("--------------------------------------------\nserver wrote:\n\n" - + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len)); + if (len == 0) { + System.out.println("--------------------------------------------\n" + + "server wrote:\n\n"); + } else { + System.out.println("--------------------------------------------\n" + + "server wrote:\n\n" + + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len)); + } } } } + // todo remove private byte[] toByteArray(net.openhft.lang.io.Bytes bytes) { if (bytes == null || bytes.remaining() == 0) return new byte[]{}; @@ -419,10 +404,10 @@ class MapWireHandler implements WireHandler, Consumer { return result; } + // todo remove private byte[] toBytes(WireKey fieldName) { final Wire wire = inWire; - System.out.println(Bytes.toDebugString(inWire.bytes())); final ValueIn read = wire.read(fieldName); final long l = read.readLength(); @@ -448,7 +433,7 @@ class MapWireHandler implements WireHandler, Consumer { private void putAll(long transactionId) { - /*final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId); + final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId); if (bytesMap == null) return; @@ -461,6 +446,11 @@ class MapWireHandler implements WireHandler, Consumer { return; } + // Note : you can not assume that all the entries in a putAll will be continuous, + // they maybe other transactions from other threads. + // we it should be possible for a single entry to fill the Tcp buffer, so each entry + // should have the ability to be processed separately + // and then only applied to the map once all the entries are received. runnable = new Runnable() { // we should try and collect the data and then apply it atomically as quickly possible @@ -468,29 +458,57 @@ class MapWireHandler implements WireHandler, Consumer { @Override public void run() { - if (inWire.read(HAS_NEXT).bool()) { - collectData.put(toBytes(ARG_1), toBytes(ARG_2)); - } else { - // the old code assumed that all the data would fit into a single buffer - // this assumption is invalid - if (!collectData.isEmpty()) { - bytesMap.delegate.putAll((Map) collectData); + + + // the old code assumed that ALL of the entries would fit into a single buffer + // this assumption is invalid + boolean hasNext; + for (; ; ) { + + hasNext = inWire.read(Fields.hasNext).bool(); + // todo remove toBytes() + collectData.put(toBytes(arg1), toBytes(arg2)); + + if (!hasNext) { + incompleteWork.remove(transactionId); - outWire.write(TRANSACTION_ID).int64(transactionId); - // todo handle the case where there is an exception - outWire.write(IS_EXCEPTION).bool(false); + outWire.write(Fields.transactionId).int64(transactionId); + writeVoid(() -> { + bytesMap.delegate.putAll((Map) collectData); + return null; + }); + return; } + if (inWire.bytes().remaining() == 0) + return; } } }; incompleteWork.put(transactionId, runnable); - runnable.run();*/ + runnable.run(); } + @SuppressWarnings("SameReturnValue") + private void writeValueFromBytes(final Function f) { + + write(b -> { + + byte[] fromBytes = f.apply(b); + boolean isNull = fromBytes == null || fromBytes.length == 0; + outWire.write(resultIsNull).bool(isNull); + if (isNull) + return; + + outWire.write(result); + outWire.bytes().write(fromBytes); + + }); + } + private byte identifier() { // if the client provides a remote identifier then we will use that otherwise we will use // the local identifier @@ -643,22 +661,9 @@ class MapWireHandler implements WireHandler, Consumer { @SuppressWarnings("SameReturnValue") private void writeValue(final Function f) { - - write(b -> { - - byte[] fromBytes = toByteArray(f.apply(b)); - boolean isNull = fromBytes.length == 0; - outWire.write(RESULT_IS_NULL).bool(isNull); - if (isNull) - return; - - outWire.write(RESULT); - outWire.bytes().write(fromBytes); - - }); + writeValueFromBytes(b -> toByteArray(f.apply(b))); } - @SuppressWarnings("SameReturnValue") private void writeValueUsingDelegate(final Function, byte[]> f) { @@ -667,13 +672,13 @@ class MapWireHandler implements WireHandler, Consumer { byte[] result = f.apply((ChronicleMap) b.delegate); boolean isNull = result == null; - outWire.write(RESULT_IS_NULL).bool(isNull); + outWire.write(resultIsNull).bool(isNull); if (isNull) return; isNull = result.length == 0; - outWire.write(RESULT); + outWire.write(Fields.result); outWire.bytes().write(result); }); @@ -681,7 +686,6 @@ class MapWireHandler implements WireHandler, Consumer { } - @SuppressWarnings("SameReturnValue") private void write(@NotNull Consumer c) { @@ -694,7 +698,7 @@ class MapWireHandler implements WireHandler, Consumer { bytesMap.output = null; outWire.bytes().mark(); - outWire.write(IS_EXCEPTION).bool(false); + outWire.write(isException).bool(false); try { c.accept(bytesMap); @@ -702,8 +706,8 @@ class MapWireHandler implements WireHandler, Consumer { outWire.bytes().reset(); // the idea of wire is that is platform independent, // so we wil have to send the exception as a String - outWire.write(IS_EXCEPTION).bool(true); - outWire.write(EXCEPTION).text(toString(e)); + outWire.write(isException).bool(true); + outWire.write(exception).text(toString(e)); LOG.error("", e); return; } @@ -722,31 +726,22 @@ class MapWireHandler implements WireHandler, Consumer { bytesMap.output = null; outWire.bytes().mark(); - outWire.write(IS_EXCEPTION).bool(false); try { r.call(); + outWire.write(isException).bool(false); } catch (Exception e) { outWire.bytes().reset(); // the idea of wire is that is platform independent, // so we wil have to send the exception as a String - outWire.write(IS_EXCEPTION).bool(true); - outWire.write(EXCEPTION).text(toString(e)); + outWire.write(isException).bool(true); + outWire.write(exception).text(toString(e)); LOG.error("", e); return; } } - /** - * only used for debugging - */ - @SuppressWarnings("UnusedDeclaration") - private void showOutWire() { - System.out.println("pos=" + outWire.bytes().position() + ",bytes=" + - Bytes.toDebugString(outWire.bytes(), 0, outWire.bytes().position())); - } - @SuppressWarnings("SameReturnValue") private void writeVoid(@NotNull Consumer process) { @@ -761,14 +756,23 @@ class MapWireHandler implements WireHandler, Consumer { bytesMap.output = null; try { process.accept(bytesMap); - outWire.write(IS_EXCEPTION).bool(false); + outWire.write(isException).bool(false); } catch (Exception e) { - outWire.write(IS_EXCEPTION).bool(true); + outWire.write(isException).bool(true); LOG.error("", e); } } + /** + * only used for debugging + */ + @SuppressWarnings("UnusedDeclaration") + private void showOutWire() { + System.out.println("pos=" + outWire.bytes().position() + ",bytes=" + + Bytes.toDebugString(outWire.bytes(), 0, outWire.bytes().position())); + } + /** * converts the exception into a String, so that it can be sent to c# clients */ @@ -783,5 +787,43 @@ class MapWireHandler implements WireHandler, Consumer { this.localIdentifier = localIdentifier; } + static enum EventId { + + longSize, + size, + isEmpty, + containsKey, + containsValue, + get, + put, + putWithoutAcc, + remove, + removeWithoutAcc, + clear, + keySet, + values, + entrySet, + replace, + replaceWithOldAndNewValue, + putIfAbsent, + removeWithValue, + toString, + applicationVersion, + persistedDataVersion, + putAll, + putAllWithoutAcc, + hashCode, + mapForKey, + putMapped, + keyBuilder, + valueBuilder, + createChannel, + remoteIdentifier; + + public boolean contentEquals(CharSequence c) { + return this.toString().contentEquals(c); + } + } + } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java index f4de30927..91eb3d7f0 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java @@ -21,7 +21,6 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; import net.openhft.chronicle.hash.replication.ReplicationHub; import net.openhft.chronicle.network2.WireHandler; - import net.openhft.chronicle.wire.WireKey; import org.jetbrains.annotations.NotNull; @@ -39,22 +38,23 @@ public class MapWireHandlerBuilder { return new MapWireHandler<>(mapFactory, hub, localIdentifier, channels); } + // note : peter has asked for these to be in camel case public static enum Fields implements WireKey { - HAS_NEXT, - TIME_STAMP, - CHANNEL_ID, - METHOD_NAME, - TYPE, - TRANSACTION_ID, - RESULT, - RESULT_KEY, - RESULT_VALUE, - ARG_1, - ARG_2, - ARG_3, - IS_EXCEPTION, - EXCEPTION, - RESULT_IS_NULL + hasNext, + timeStamp, + channelId, + methodName, + type, + transactionId, + result, + resultKey, + resultValue, + arg1, + arg2, + arg3, + isException, + exception, + resultIsNull } } diff --git a/src/main/java/net/openhft/chronicle/map/Replica.java b/src/main/java/net/openhft/chronicle/map/Replica.java index c0b973ec9..4ba937eba 100644 --- a/src/main/java/net/openhft/chronicle/map/Replica.java +++ b/src/main/java/net/openhft/chronicle/map/Replica.java @@ -164,8 +164,7 @@ public interface Replica extends Closeable { * writeExternalEntry()}. This method is typically called when we receive a remote * replication event, this event could originate from either a remote {@code put(K key, V *value)} or {@code remove(Object key)} - * @param copies - * @param context + * @param context the Bytes Replicated Context * @param source bytes to read an entry from */ void readExternalEntry( @@ -210,7 +209,7 @@ public interface Replica extends Closeable { /** * Called whenever a put() or remove() has occurred to a replicating map. * - * @param context the entry you will receive, this does not have to be locked, as + * @param entry the entry you will receive, this does not have to be locked, as * locking is already provided from the caller. * @param chronicleId only assigned when clustering * @return {@code false} if this entry should be ignored because the identifier of the @@ -220,13 +219,13 @@ public interface Replica extends Closeable { public abstract boolean onEntry(final Bytes entry, final int chronicleId); /** - * Called just after {@link #onEntry(VanillaContext, int)}. No-op by default. + * Called just after {@link #onEntry(net.openhft.lang.io.Bytes, int)}. No-op by default. */ public void onAfterEntry() { } /** - * Called just before {@link #onEntry(VanillaContext, int)}. No-op by default. + * Called just before {@link #onEntry(net.openhft.lang.io.Bytes, int)}. No-op by default. */ public void onBeforeEntry() { } diff --git a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java index 3e2befac3..787513507 100755 --- a/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java +++ b/src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java @@ -117,7 +117,7 @@ public final class ChronicleSetBuilder * * @see #constantKeySizeBySample(Object) * @see #actualChunkSize(int) - * @param averageKeySize + * @param averageKeySize the average size in bytes of the key */ @Override public ChronicleSetBuilder averageKeySize(double averageKeySize) { diff --git a/src/test/java/net/openhft/chronicle/map/ConcurrentMap8Test.java b/src/test/java/net/openhft/chronicle/map/ConcurrentMap8Test.java index fb7a3f2bb..2a9071919 100644 --- a/src/test/java/net/openhft/chronicle/map/ConcurrentMap8Test.java +++ b/src/test/java/net/openhft/chronicle/map/ConcurrentMap8Test.java @@ -134,7 +134,7 @@ public class ConcurrentMap8Test extends JSR166TestCase { static ConcurrentMap longMap() { if (longMap == null) { longMap = newLongMap(); - for (int i = 0; i < SIZE; ++i) + for (int i = 0; i < size; ++i) longMap.put(Long.valueOf(i), Long.valueOf(2 * i)); } diff --git a/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java b/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java index 738710137..208972d8a 100644 --- a/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java +++ b/src/test/java/net/openhft/chronicle/map/TcpTimeBasedReplicationSoakTest.java @@ -6,10 +6,7 @@ import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.model.Byteable; import net.openhft.lang.model.DataValueClasses; import net.openhft.lang.values.IntValue; -import org.junit.After; -import org.junit.Assert; -import org.junit.Before; -import org.junit.Test; +import org.junit.*; import org.mockito.Mockito; import java.io.Closeable; @@ -104,6 +101,7 @@ public class TcpTimeBasedReplicationSoakTest { } + @Ignore @Test public void testSoakTestWithRandomData() throws IOException, InterruptedException {