Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Merge branch 'master' of github.com:OpenHFT/Chronicle-Map

Browse Source 
Conflicts:
	src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
	src/main/java/net/openhft/chronicle/map/MapWireHandler.java
This commit is contained in:
leventov
2015-03-14 20:39:18 +07:00
parent b0b0bb87c1 de9443436f
commit 56b48e56b4
11 changed files with 399 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/hashing/LongHashFunction.java
+13 -1
View File
@@ -63,6 +63,7 @@ public abstract class LongHashFunction implements Serializable {
* CityHash64 algorithm, version 1.1</a> 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</a> 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();
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+1 -1
View File
@@ -256,7 +256,7 @@ public final class ChronicleMapBuilder<K, V> 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)
src/main/java/net/openhft/chronicle/map/ChronicleMapStatelessClientBuilder.java
+2 -1
View File
@@ -48,12 +48,13 @@ public final class ChronicleMapStatelessClientBuilder<K, V> implements
// -- the second is really shorter.
/**
* Equivalent of {@code ChronicleMapStatelessClientBuilder.<K, V>of(serverAddress).create()}.
* Equivalent of {@code ChronicleMapStatelessClientBuilder.<K,V>of(serverAddress).create()}.
*
* @param serverAddress address of the server map
* @param <K> key type of the map
* @param <V> 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 <K, V> ChronicleMap<K, V> createClientOf(InetSocketAddress serverAddress)
throws IOException {
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+111 -112
View File
@@ -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<K, V>
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<K, V>
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<K, V>
@Override
public int hashCode() {
return proxyReturnInt(HASH_CODE.toString());
return proxyReturnInt(hashCode.toString());
}
@NotNull
@@ -205,15 +204,15 @@ class ClientWiredStatelessChronicleMap<K, V>
@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<K, V>
}
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<K, V>
}
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<K, V>
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<K, V>
final long startTime = System.currentTimeMillis();
long transactionId = hub.nextUniqueTransaction(startTime);
Set<? extends Map.Entry<? extends K, ? extends V>> entries = map.entrySet();
Iterator<? extends Map.Entry<? extends K, ? extends V>> 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<K, V>
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<? extends K, ? extends V> 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<K, V>
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<K, V>
}
public void clear() {
proxyReturnVoid(() -> writeField(Fields.METHOD_NAME, CLEAR.toString()));
proxyReturnVoid(() -> writeField(Fields.methodName, clear.toString()));
}
@NotNull
public Set<K> keySet() {
final Set<K> usingCollection = new HashSet<K>();
readChunked(kClass, "KEY_SET", usingCollection);
readChunked(kClass, keySet, usingCollection);
return usingCollection;
}
@@ -359,11 +356,11 @@ class ClientWiredStatelessChronicleMap<K, V>
@NotNull
public Collection<V> values() {
final List<V> usingCollection = new ArrayList<>();
return readChunked(vClass, "VALUES", usingCollection);
return readChunked(vClass, values, usingCollection);
}
private <E, A extends Collection<E>> Collection<E> readChunked(
Class<E> vClass1, String values, A usingCollection) {
Class<E> vClass1, EventId values, A usingCollection) {
final long startTime = System.currentTimeMillis();
// send
@@ -380,14 +377,14 @@ class ClientWiredStatelessChronicleMap<K, V>
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<K, V>
final Map<K, V> result = new HashMap<K, V>();
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<K, V>
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<K, V>
// 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<K, V>
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<K, V>
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<K, V>
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<K, V>
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<K, V>
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<K, V>
}
@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<K, V>
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<K, V>
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<K, V>
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<K, V>
@Nullable
private <R> R proxyReturnObject(
@NotNull final String methodName, K key, V value, Class<V> resultType) {
@NotNull final EventId methodName, K key, V value, Class<V> resultType) {
final long startTime = System.currentTimeMillis();
long transactionId;
@@ -755,9 +752,9 @@ class ClientWiredStatelessChronicleMap<K, V>
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<K, V>
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<K, V>
}
@Nullable
private <R> R proxyReturnObject(Class<R> rClass, @NotNull final String methodName, Object key) {
private <R> R proxyReturnObject(
Class<R> rClass, @NotNull final EventId methodName, Object key) {
final long startTime = System.currentTimeMillis();
long transactionId;
@@ -787,19 +785,19 @@ class ClientWiredStatelessChronicleMap<K, V>
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<K, V>
}
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<K, V>
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<K, V>
}
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<K, V> {
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+16 -15
View File
@@ -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;
}
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+232 -190
View File
@@ -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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<Long, Runnable> incompleteWork = new HashMap<Long, Runnable>();
private final ArrayList<BytesChronicleMap> bytesChronicleMaps = new ArrayList<>();
@NotNull
private final StringBuilder methodName = new StringBuilder();
private final Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory;
private Wire inWire = null;
private Wire outWire = null;
@@ -72,19 +72,10 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
@Override
public void accept(Iterator<byte[]> 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<Iterator<Map.Entry<byte[], byte[]>>>() {
@Override
@@ -92,13 +83,14 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
final Map.Entry<byte[], byte[]> 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<K, V> implements WireHandler, Consumer<WireHandlers> {
private ReplicationHub hub;
private byte remoteIdentifier;
public MapWireHandler(
@NotNull Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
@NotNull ReplicationHub hub, byte localIdentifier, @NotNull List<Replica> channels) {
this(mapFactory, hub);
this.channels = channels;
this.localIdentifier = localIdentifier;
}
public MapWireHandler(Supplier<ChronicleHashInstanceBuilder<ChronicleMap<K, V>>> mapFactory,
@@ -123,19 +112,11 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
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<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator> c) throws StreamCorruptedException {
@@ -148,7 +129,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
@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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
}
/* private void writeEntryChunked(long transactionId,
@NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator<Map.Entry<byte[], byte[]>>> c) {
final BytesChronicleMap m = bytesMap(channelId);
final Iterator<Map.Entry<byte[], byte[]>> 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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<byte[], byte[]>) 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<K, V> implements WireHandler, Consumer<WireHandlers> {
// 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<EMPTY>");
} 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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
@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<BytesChronicleMap, byte[]> 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<K, V> implements WireHandler, Consumer<WireHandlers> {
@SuppressWarnings("SameReturnValue")
private void writeValue(final Function<BytesChronicleMap, net.openhft.lang.io.Bytes> 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<ChronicleMap<byte[], byte[]>, byte[]> f) {
@@ -667,13 +672,13 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
}
@SuppressWarnings("SameReturnValue")
private void write(@NotNull Consumer<BytesChronicleMap> c) {
@@ -694,7 +698,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<BytesChronicleMap> process) {
@@ -761,14 +756,23 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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);
}
}
}
src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java
+16 -16
View File
@@ -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
}
}
src/main/java/net/openhft/chronicle/map/Replica.java
+4 -5
View File
@@ -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() {
}
src/main/java/net/openhft/chronicle/set/ChronicleSetBuilder.java
+1 -1
View File
@@ -117,7 +117,7 @@ public final class ChronicleSetBuilder<K>
*
* @see #constantKeySizeBySample(Object)
* @see #actualChunkSize(int)
* @param averageKeySize
* @param averageKeySize the average size in bytes of the key
*/
@Override
public ChronicleSetBuilder<K> averageKeySize(double averageKeySize) {