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

CE-20 migrating the code towards the https://github.com/OpenHFT/RFC/blob/master/Chronicle/Map/Remote/Chronicle-Map-0.1.md specification - still work in progress - DON'T REFACTOR

Browse Source
This commit is contained in:
Rob Austin
2015-04-02 17:31:39 +01:00
parent 92741c26fa
commit 65fe9415b9
5 changed files with 298 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+98 -128
View File
@@ -22,6 +22,7 @@ import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.hash.function.SerializableFunction;
import net.openhft.chronicle.map.MapWireHandlerBuilder.Fields;
import net.openhft.chronicle.wire.Marshallable;
import net.openhft.chronicle.wire.ValueOut;
import net.openhft.chronicle.wire.Wire;
import net.openhft.chronicle.wire.WireOut;
import org.jetbrains.annotations.NotNull;
@@ -143,7 +144,7 @@ class ClientWiredStatelessChronicleMap<K, V>
if (key == null || value == null)
throw new NullPointerException();
return proxyReturnObject(putIfAbsent, key, value, vClass);
return proxyReturnTypedObject(putIfAbsent, vClass, key, value);
}
@SuppressWarnings("NullableProblems")
@@ -151,21 +152,21 @@ class ClientWiredStatelessChronicleMap<K, V>
if (key == null)
throw new NullPointerException();
return value != null && proxyReturnBoolean(removeWithValue.toString(), (K) key, (V) value);
return value != null && proxyReturnBoolean(removeWithValue, (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(replaceWithOldAndNewValue.toString(), key, oldValue, newValue);
return proxyReturnBoolean(replaceWithOldAndNewValue, key, oldValue, newValue);
}
@SuppressWarnings("NullableProblems")
public V replace(K key, V value) {
if (key == null || value == null)
throw new NullPointerException();
return proxyReturnObject(replace, key, value, vClass);
return proxyReturnTypedObject(replace, vClass, key, value);
}
public int size() {
@@ -280,8 +281,8 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
hub.outWire().write(Fields.methodName).text(createChannel.toString());
hub.outWire().write(Fields.arg1).int16(channelID);
hub.outWire().writeEventName(createChannel);
hub.outWire().writeValue().int16(channelID);
}
});
@@ -313,7 +314,8 @@ class ClientWiredStatelessChronicleMap<K, V>
wire.write(Fields.tid).int64(tid);
wire.write(Fields.timeStamp).int64(startTime);
wire.write(Fields.channelId).int16(channelID);
hub.outWire().write(Fields.methodName).text(putAll.toString());
hub.outWire().writeEventName(putAll);
while (iterator.hasNext()) {
@@ -348,9 +350,11 @@ class ClientWiredStatelessChronicleMap<K, V>
public V put(K key, V value) {
if (key == null || value == null)
throw new NullPointerException();
return proxyReturnObject(putReturnsNull ? putWithoutAcc : put, key, value, vClass);
return proxyReturnTypedObject(putReturnsNull ? putWithoutAcc : put, vClass, key,
value);
}
@Nullable
public <R> R getMapped(@Nullable K key, @NotNull SerializableFunction<? super V, R> function) {
throw new UnsupportedOperationException();
@@ -366,8 +370,7 @@ class ClientWiredStatelessChronicleMap<K, V>
proxyReturnVoid(() -> hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(clear.toString());
wireOut.writeEventName(clear);
}
}
));
@@ -410,7 +413,7 @@ class ClientWiredStatelessChronicleMap<K, V>
while (bytes.remaining() > 0) {
boolean bool = wireIn.read(Fields.hasNext).bool();
if (bool) {
usingCollection.add(readObject(Fields.result, wireIn, null, vClass1));
usingCollection.add(readObject(Fields.reply, wireIn, null, vClass1));
} else
break OUTER;
@@ -508,7 +511,7 @@ class ClientWiredStatelessChronicleMap<K, V>
if (bytes.remaining() > 0) {
boolean isException = wireIn.read(Fields.isException).bool();
if (isException)
throw new RuntimeException(wireIn.read(Fields.result).text());
throw new RuntimeException(wireIn.read(Fields.reply).text());
}
}
@@ -536,37 +539,38 @@ class ClientWiredStatelessChronicleMap<K, V>
// receive
hub.inBytesLock().lock();
try {
return hub.proxyReply(timeoutTime, tid).read(Fields.result).int64();
return hub.proxyReply(timeoutTime, tid).read(Fields.reply).int64();
} finally {
hub.inBytesLock().unlock();
}
}
private void writeField(WireOut wireOut, Fields fieldName, Object value) {
writeField(fieldName, value, wireOut);
private void writeField(ValueOut wireOut, Object value) {
writeField(value, wireOut);
}
private void writeField(Fields fieldName, Object value, WireOut wire) {
private void writeField(Object value, ValueOut valueOut) {
assert hub.outBytesLock().isHeldByCurrentThread();
assert !hub.inBytesLock().isHeldByCurrentThread();
if (value instanceof Byte)
wire.write(fieldName).int8((Byte) value);
valueOut.int8((Byte) value);
else if (value instanceof Character)
wire.write(fieldName).text(value.toString());
valueOut.text(value.toString());
else if (value instanceof Short)
wire.write(fieldName).int16((Short) value);
valueOut.int16((Short) value);
else if (value instanceof Integer)
wire.write(fieldName).int32((Integer) value);
valueOut.int32((Integer) value);
else if (value instanceof Long)
wire.write(fieldName).int64((Long) value);
valueOut.int64((Long) value);
else if (value instanceof CharSequence) {
wire.write(fieldName).text((CharSequence) value);
valueOut.text((CharSequence) value);
} else if (value instanceof Marshallable) {
wire.write(fieldName).marshallable((Marshallable) value);
valueOut.marshallable((Marshallable) value);
} else {
throw new IllegalStateException("type=" + value.getClass() +
" is unsupported, it must either be of type Marshallable or CharSequence");
@@ -669,7 +673,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, tid);
return wireIn.read(Fields.result).bool();
return wireIn.read(Fields.reply).bool();
} finally {
hub.inBytesLock().unlock();
}
@@ -684,14 +688,14 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, tid);
return wireIn.read(Fields.result).int32();
return wireIn.read(Fields.reply).int32();
} finally {
hub.inBytesLock().unlock();
}
}
/*
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value) {
private boolean proxyReturnBoolean(@NotNull final EventId methodName, Object... parameters) {
final long startTime = System.currentTimeMillis();
long tid;
@@ -700,15 +704,7 @@ class ClientWiredStatelessChronicleMap<K, V>
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, key);
writeField(wireOut, Fields.arg2, value);
}
}
);
sendEvent(methodName, parameters);
hub.writeSocket(hub.outWire());
@@ -718,36 +714,63 @@ class ClientWiredStatelessChronicleMap<K, V>
return readBoolean(tid, startTime);
}*/
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(
@NotNull final EventId methodName, Object... args) {
final long startTime = System.currentTimeMillis();
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
sendEvent(methodName, args);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readBoolean(tid, startTime);
}
@SuppressWarnings("SameParameterValue")
private boolean proxyReturnBoolean(
@NotNull final String methodName, K key, V value1, V value2) {
final long startTime = System.currentTimeMillis();
private void sendEvent(@NotNull final EventId methodName, final Object[] args) {
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
assert args.length > 0;
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, key);
writeField(wireOut, Fields.arg2, value1);
writeField(wireOut, Fields.arg3, value2);
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
final ValueOut out = wireOut.writeEventName(methodName);
final MapWireHandler.Params[] paramNames = methodName.params();
if (paramNames.length == 1) {
writeField(out, args[0]);
return;
}
assert args.length == paramNames.length :
"methodName=" + methodName +
", args.length=" + args.length +
", paramNames.length=" + paramNames.length;
out.marshallable(m -> {
for (int i = 0; i < paramNames.length; i++) {
final ValueOut vo = m.write(paramNames[i]);
writeField(vo, args[i]);
}
});
}
);
hub.writeSocket(hub.outWire());
} finally {
hub.outBytesLock().unlock();
}
return readBoolean(tid, startTime);
}
);
}
@SuppressWarnings("SameParameterValue")
@@ -776,7 +799,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
wireOut.writeEventName(() -> methodName);
}
}
);
@@ -802,7 +825,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
wireOut.writeEventName(() -> methodName);
}
}
);
@@ -847,7 +870,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
wireOut.writeEventName(() -> methodName);
}
}
);
@@ -862,8 +885,8 @@ class ClientWiredStatelessChronicleMap<K, V>
@Nullable
private <R> R proxyReturnObject(
@NotNull final EventId methodName, K key, V value, Class<V> resultType) {
private <R> R proxyReturnTypedObject(
@NotNull final EventId methodName, Class<V> resultType, Object... args) {
final long startTime = System.currentTimeMillis();
long tid;
@@ -875,18 +898,7 @@ class ClientWiredStatelessChronicleMap<K, V>
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, key);
writeField(wireOut, Fields.arg2, value);
}
}
);
sendEvent(methodName, args);
hub.writeSocket(hub.outWire());
} finally {
@@ -897,7 +909,7 @@ class ClientWiredStatelessChronicleMap<K, V>
return null;
if (resultType == vClass)
return (R) readValue(Fields.result, tid, startTime, null);
return (R) readValue(Fields.reply, tid, startTime, null);
else
throw new UnsupportedOperationException("class of type class=" + resultType +
@@ -917,28 +929,19 @@ class ClientWiredStatelessChronicleMap<K, V>
return null;
if (rClass == vClass)
return (R) readValue(Fields.result, tid, startTime, null);
return (R) readValue(Fields.reply, tid, startTime, null);
else
throw new UnsupportedOperationException("class of type class=" + rClass + " is not " +
"supported");
}
private long proxySend(EventId methodName, long startTime, Object arg1) {
private long proxySend(EventId methodName, long startTime, Object... args) {
long tid;
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, arg1);
}
}
);
sendEvent(methodName, args);
hub.writeSocket(hub.outWire());
@@ -957,7 +960,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
wireOut.writeEventName(methodName);
}
}
);
@@ -976,13 +979,13 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.outBytesLock().lock();
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
hub.outWire().write(Fields.methodName).text(methodName.toString());
hub.outWire().write(Fields.eventName).text(methodName.toString());
hub.outWire().writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(Fields.methodName).text(methodName.toString());
writeField(wireOut, Fields.arg1, arg1);
final ValueOut out = wireOut.writeEventName(methodName);
writeField(out, arg1);
}
}
);
@@ -1022,39 +1025,6 @@ class ClientWiredStatelessChronicleMap<K, V>
}
/* static enum EventId {
createChannel,
HEARTBEAT,
STATEFUL_UPDATE,
longSize,
size,
isEmpty,
containsKey,
containsValue,
get,
put,
putWithoutAcc,
remove,
removeWithoutAcc,
clear,
KEY_SET,
VALUES,
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
+23 -12
View File
@@ -144,9 +144,11 @@ public class ClientWiredStatelessTcpConnectionHub {
}
}
public static boolean IS_DEBUG = java.lang.management.ManagementFactory.getRuntimeMXBean().
getInputArguments().toString().indexOf("jdwp") >= 0;
private void checkTimeout(long timeoutTime) {
if (timeoutTime < System.currentTimeMillis())
if (timeoutTime < System.currentTimeMillis() && !IS_DEBUG)
throw new RemoteCallTimeoutException();
}
@@ -380,13 +382,6 @@ public class ClientWiredStatelessTcpConnectionHub {
long tid0 = intWire.read(MapWireHandlerBuilder.Fields.tid).int64();
// check the transaction id is reasonable
/* assert tid0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR :
"TransactionId too small " + tid0 + " messageSize " + messageSize;
assert tid0 < 2100000000000L * TcpReplicator.TIMESTAMP_FACTOR :
"TransactionId too large " + tid0 + " messageSize " + messageSize;
*/
// if the transaction id is for this thread process it
if (tid0 == tid) {
clearParked();
@@ -533,8 +528,7 @@ public class ClientWiredStatelessTcpConnectionHub {
outBuffer.position(SIZE_OF_SIZE);
if (EventGroup.IS_DEBUG ) {
System.out.println("\n--------------------------------------------\n" +
"client writes:\n\n" + Bytes.toDebugString(outBuffer));
writeBytesToStandardOut(bytes, outBuffer);
}
outBuffer.position(0);
@@ -572,6 +566,23 @@ public class ClientWiredStatelessTcpConnectionHub {
}
private void writeBytesToStandardOut(Bytes<?> bytes, ByteBuffer outBuffer) {
final long position = bytes.position();
final long limit = bytes.limit();
try {
bytes.limit(outBuffer.limit());
bytes.position(outBuffer.position());
System.out.println("\n--------------------------------------------\n" +
"client writes:\n\n" +
// Bytes.toDebugString(outBuffer));
Wires.fromSizePrefixedBlobs(bytes));
} finally {
bytes.limit(limit);
bytes.position(position);
}
}
/**
* calculates the size of each chunk
*
@@ -601,7 +612,7 @@ public class ClientWiredStatelessTcpConnectionHub {
wire.writeDocument(false, new Consumer<WireOut>() {
@Override
public void accept(WireOut wireOut) {
wireOut.write(MapWireHandlerBuilder.Fields.methodName).text(methodName.toString());
wireOut.writeEventName(methodName);
}
});
@@ -636,7 +647,7 @@ public class ClientWiredStatelessTcpConnectionHub {
// receive
inBytesLock().lock();
try {
return proxyReply(timeoutTime, tid).read(() -> "result").text();
return proxyReply(timeoutTime, tid).read(reply).text();
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+174 -146
View File
@@ -30,10 +30,7 @@ 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.chronicle.wire.*;
import net.openhft.lang.io.DirectStore;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
@@ -51,6 +48,7 @@ import java.util.function.Function;
import java.util.function.Supplier;
import static net.openhft.chronicle.map.MapWireHandler.EventId.*;
import static net.openhft.chronicle.map.MapWireHandler.Params.*;
import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*;
/**
@@ -72,7 +70,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
@Override
public void accept(Iterator<byte[]> iterator) {
outWire.write(result);
outWire.write(reply);
Bytes<?> bytes = outWire.bytes();
bytes.write(iterator.next());
}
@@ -178,7 +176,8 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
private void writeChunked(
long tid, @NotNull final Function<BytesChronicleMap, Iterator> function,
long tid,
@NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator> c) throws StreamCorruptedException {
writeChunked(tid, function, c, Long.MAX_VALUE);
}
@@ -195,6 +194,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
}
@SuppressWarnings("UnusedReturnValue")
void onEvent() throws StreamCorruptedException {
@@ -209,8 +209,11 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
final int body = bytes1.readVolatileInt();
assert Wires.isData(body);
final StringBuilder methodName = Wires.acquireStringBuilder();
inWire.read(Fields.methodName).text(methodName);
final StringBuilder eventName = Wires.acquireStringBuilder();
final ValueIn valueIn = inWire.readEventName(eventName);
if (!incompleteWork.isEmpty()) {
Runnable runnable = incompleteWork.get(tid);
@@ -219,40 +222,51 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
return;
}
}
Bytes<?> bytes = outWire.bytes();
final Bytes<?> outBytes = outWire.bytes();
final Bytes<?> bytes = outBytes;
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);
if (putWithoutAcc.contentEquals(eventName)) {
valueIn.marshallable(wire -> {
final Params[] params = putIfAbsent.params;
final byte[] key = wire.read(params[0]).bytes();
final byte[] value = wire.read(params[1]).bytes();
writeVoid(b -> ((Map<byte[], byte[]>) b.delegate).putIfAbsent(key, value));
});
return;
}
if (keySet.contentEquals(methodName)) {
if (keySet.contentEquals(eventName)) {
writeChunked(tid, map -> map.keySet().iterator(), writeElement);
return;
}
if (values.contentEquals(methodName)) {
if (values.contentEquals(eventName)) {
writeChunked(tid, map -> map.delegate.values().iterator(), writeElement);
return;
}
if (entrySet.contentEquals(methodName)) {
if (entrySet.contentEquals(eventName)) {
writeChunked(tid, m -> m.delegate.entrySet().iterator(), writeEntry);
return;
}
if (entrySetRestricted.contentEquals(methodName)) {
if (entrySetRestricted.contentEquals(eventName)) {
long maxEntries = inWire.read(arg1).int64();
writeChunked(tid, m -> m.delegate.entrySet().iterator(), writeEntry,maxEntries);
return;
}
if (putAll.contentEquals(methodName)) {
if (putAll.contentEquals(eventName)) {
putAll(tid);
return;
}
@@ -263,153 +277,170 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
// write the transaction id
if (createChannel.contentEquals(methodName)) {
if (createChannel.contentEquals(eventName)) {
writeVoid(() -> {
short channelId1 = inWire.read(arg1).int16();
short channelId1 = valueIn.int16();
mapFactory.get().replicatedViaChannel(hub.createChannel(channelId1)).create();
return null;
});
return;
}
if (EventId.remoteIdentifier.contentEquals(methodName)) {
this.remoteIdentifier = inWire.read(result).int8();
if (EventId.remoteIdentifier.contentEquals(eventName)) {
this.remoteIdentifier = valueIn.int8();
return;
}
if (longSize.contentEquals(methodName)) {
write(b -> outWire.write(result).int64(b.longSize()));
if (longSize.contentEquals(eventName)) {
write(b -> outWire.write(reply).int64(b.longSize()));
return;
}
if (isEmpty.contentEquals(methodName)) {
write(b -> outWire.write(result).bool(b.isEmpty()));
if (isEmpty.contentEquals(eventName)) {
write(b -> outWire.write(reply).bool(b.isEmpty()));
return;
}
if (containsKey.contentEquals(methodName)) {
if (containsKey.contentEquals(eventName)) {
// todo remove the toByteArray(..)
write(b -> outWire.write(result)
.bool(b.delegate.containsKey(toByteArray(inWire, arg1))));
write(b -> outWire.write(reply)
.bool(b.delegate.containsKey(toByteArray(valueIn))));
return;
}
if (containsValue.contentEquals(methodName)) {
if (containsValue.contentEquals(eventName)) {
// todo remove the toByteArray(..)
write(b -> outWire.write(result)
.bool(b.delegate.containsValue(toByteArray(inWire, arg1))));
write(b -> outWire.write(reply)
.bool(b.delegate.containsValue(toByteArray(valueIn))));
return;
}
if (get.contentEquals(methodName)) {
if (get.contentEquals(eventName)) {
// todo remove the toByteArray(..)
writeValueUsingDelegate(map -> map.get(toByteArray(inWire, arg1)));
writeValueUsingDelegate(map -> {
final byte[] key = toByteArray(valueIn);
final byte[] value = map.get(key);
return value;
});
return;
}
if (put.contentEquals(methodName)) {
writeValue(b -> {
final net.openhft.lang.io.Bytes reader =
MapWireHandler.this.toReader(inWire, arg1, arg2);
if (put.contentEquals(eventName)) {
valueIn.marshallable(wire -> {
final Params[] params = put.params;
final byte[] key1 = wire.read(params[0]).bytes();
final byte[] value1 = wire.read(params[1]).bytes();
MapWireHandler.this.writeValue(b -> {
final byte[] result = b.put(key1, value1);
System.out.println("key='" + new String(b.keySet().iterator().next()) + "'");
return result;
});
// todo call return b.put(reader, reader, timestamp, identifier());
return b.put(reader, reader);
});
return;
}
if (remove.contentEquals(methodName)) {
writeValue(b -> b.remove(toReader(inWire, arg1)));
if (remove.contentEquals(eventName)) {
MapWireHandler.this.writeValue(b -> {
return b.remove(toByteArray(valueIn));
});
return;
}
if (clear.contentEquals(methodName)) {
if (clear.contentEquals(eventName)) {
writeVoid(BytesChronicleMap::clear);
return;
}
if (replace.contentEquals(methodName)) {
write(bytesMap -> {
if (replace.contentEquals(eventName)) {
// todo fix this this is a hack to get to work for now.
// todo may use something like :
// todo bytesMap.replace(reader, reader, timestamp, identifier());
final VanillaChronicleMap map = bytesMap.delegate;
byte[] result = (byte[]) map.replace(
toByteArray(inWire, arg1),
toByteArray(inWire, arg2));
valueIn.marshallable(wire -> {
boolean isNull = result == null || result.length == 0;
outWire.write(resultIsNull).bool(isNull);
if (!isNull) {
outWire.write(Fields.result);
bytes.write(result);
}
final Params[] params = replace.params;
final byte[] key = wire.read(params[0]).bytes();
final byte[] value = wire.read(params[1]).bytes();
writeValueFromBytes(b -> ((Map<byte[], byte[]>) b.delegate).replace(key, value));
});
return;
}
if (replaceWithOldAndNewValue.contentEquals(methodName)) {
if (replaceWithOldAndNewValue.contentEquals(eventName)) {
write(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, arg1, arg2, arg3);
final net.openhft.lang.io.Bytes reader = toReader(valueIn,
replaceWithOldAndNewValue.params);
boolean result = bytesMap.replace(reader, reader, reader);
outWire.write(Fields.result).bool(result);
outWire.write(Fields.reply).bool(result);
});
return;
}
if (putIfAbsent.contentEquals(methodName)) {
if (putIfAbsent.contentEquals(eventName)) {
valueIn.marshallable(wire -> {
final Params[] params = putIfAbsent.params;
final byte[] key = wire.read(params[0]).bytes();
final byte[] value = wire.read(params[1]).bytes();
writeValueFromBytes(b -> ((Map<byte[], byte[]>) b.delegate).putIfAbsent(key, value));
// 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 (removeWithValue.contentEquals(methodName)) {
if (removeWithValue.contentEquals(eventName)) {
write(bytesMap -> {
final net.openhft.lang.io.Bytes reader = toReader(inWire, arg1, arg2);
final net.openhft.lang.io.Bytes reader = toReader(valueIn, removeWithValue.params);
// todo call outWire.write(result)
// .bool(bytesMap.remove(reader, reader, timestamp, identifier()));
outWire.write(result).bool(bytesMap.remove(reader, reader));
outWire.write(reply).bool(bytesMap.remove(reader, reader));
});
return;
}
if (applicationVersion.contentEquals(methodName)) {
write(b -> outWire.write(result).text(applicationVersion()));
if (applicationVersion.contentEquals(eventName)) {
write(b -> outWire.write(reply).text(applicationVersion()));
return;
}
if (persistedDataVersion.contentEquals(methodName)) {
write(b -> outWire.write(result).text(persistedDataVersion()));
if (persistedDataVersion.contentEquals(eventName)) {
write(b -> outWire.write(reply).text(persistedDataVersion()));
return;
}
if (hashCode.contentEquals(methodName)) {
write(b -> outWire.write(result).int32(b.hashCode()));
if (hashCode.contentEquals(eventName)) {
write(b -> outWire.write(reply).int32(b.hashCode()));
return;
}
throw new IllegalStateException("unsupported event=" + methodName);
throw new IllegalStateException("unsupported event=" + eventName);
} finally {
// if (len > 4)
if (EventGroup.IS_DEBUG) {
long len = bytes.position() - SIZE_OF_SIZE;
@@ -417,6 +448,8 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
System.out.println("--------------------------------------------\n" +
"server writes:\n\n<EMPTY>");
} else {
System.out.println("--------------------------------------------\n" +
"server writes:\n\n" +
Bytes.toDebugString(bytes, SIZE_OF_SIZE, len));
@@ -541,7 +574,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
if (isNull)
return;
outWire.write(result);
outWire.write(reply);
Bytes<?> bytes = outWire.bytes();
bytes.write(fromBytes);
@@ -570,46 +603,34 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
/**
* creates a lang buffer that holds just the payload of the args
*
* @param wire the inbound wire
* @param args the key names of the {@code wire} args
* @return a new lang buffer containing the bytes of the args
*/
private net.openhft.lang.io.Bytes toReader(@NotNull Wire wire, @NotNull WireKey... args) {
private net.openhft.lang.io.Bytes toReader(ValueIn valueIn, Params... params) {
Bytes<?> bytes1 = wire.bytes();
final long inSize = bytes1.limit();
final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize).bytes();
// todo this is a bit of a hack
final Bytes<?> bytes1 = valueIn.wireIn().bytes();
final long inSize = bytes1.remaining();
// copy the bytes to the reader
for (final WireKey field : args) {
final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize + 1024).bytes();
final ValueIn read = wire.read(field);
final long fieldLength = read.readLength();
final long endPos = bytes1.position() + fieldLength;
final long limit = bytes1.limit();
valueIn.marshallable(new ReadMarshallable() {
@Override
public void readMarshallable(WireIn wire) throws IllegalStateException {
for (Params p : params) {
try {
final Bytes source = bytes1;
source.limit(endPos);
// write the size
bytes.writeStopBit(source.remaining());
while (source.remaining() > 0) {
if (source.remaining() >= 8)
bytes.writeLong(source.readLong());
else
bytes.writeByte(source.readByte());
final ValueIn read = wire.read(p);
final Bytes<?> bytes2 = read.wireIn().bytes();
final byte[] data = read.bytes();
System.out.println("out='" + new String(data) + "'");
bytes.writeStopBit(data.length);
bytes.write(data);
}
} finally {
bytes1.position(endPos);
bytes1.limit(limit);
}
});
}
return bytes.flip();
}
@@ -622,35 +643,19 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
*/
// todo remove this method - just added to get it to work for now
private byte[] toByteArray(@NotNull Wire wire, @NotNull WireKey field) {
private byte[] toByteArray(@NotNull Wire wire, @NotNull Params field) {
final ValueIn read = wire.read(field);
final long fieldLength = read.readLength();
return toByteArray(read);
}
Bytes<?> bytes = wire.bytes();
final long endPos = bytes.position() + fieldLength;
final long limit = bytes.limit();
byte[] result = new byte[]{};
try {
private byte[] toByteArray(ValueIn valueIn) {
final Bytes source = bytes;
source.limit(endPos);
if (source.remaining() > Integer.MAX_VALUE)
throw new BufferOverflowException();
// write the size
result = new byte[(int) source.remaining()];
source.read(result);
final String text = valueIn.text();
return text.getBytes();
} finally {
bytes.position(endPos);
bytes.limit(limit);
}
return result;
}
/**
@@ -701,8 +706,11 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
@SuppressWarnings("SameReturnValue")
private void writeValue(final Function<BytesChronicleMap, net.openhft.lang.io.Bytes> f) {
writeValueFromBytes(b -> toByteArray(f.apply(b)));
private void writeValue(final Function<Map<byte[], byte[]>, byte[]> f) {
writeValueFromBytes(b -> {
final Map<byte[], byte[]> delegate = (Map) b.delegate;
return f.apply(delegate);
});
}
@SuppressWarnings("SameReturnValue")
@@ -710,7 +718,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
write(b -> {
byte[] result = f.apply((ChronicleMap) b.delegate);
byte[] result = f.apply((ChronicleMap<byte[], byte[]>) b.delegate);
boolean isNull = result == null;
outWire.write(resultIsNull).bool(isNull);
@@ -719,7 +727,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
isNull = result.length == 0;
outWire.write(Fields.result);
outWire.write(Fields.reply);
final Bytes<?> bytes = outWire.bytes();
bytes.write(result);
@@ -832,27 +840,35 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
this.localIdentifier = localIdentifier;
}
static enum EventId {
static enum Params implements WireKey {
key,
value,
oldValue,
newValue
}
static enum EventId implements WireKey {
longSize,
size,
isEmpty,
containsKey,
containsValue,
get,
put,
putWithoutAcc,
remove,
removeWithoutAcc,
containsKey(key),
containsValue(value),
get(key),
put(key, value),
putWithoutAcc(key, value),
remove(key),
removeWithoutAcc(key),
clear,
keySet,
values,
entrySet,
entrySetRestricted,
replace,
replaceWithOldAndNewValue,
putIfAbsent,
removeWithValue,
replace(key, value),
replaceWithOldAndNewValue(key, oldValue, newValue),
putIfAbsent(key, value),
removeWithValue(key, value),
toString,
applicationVersion,
persistedDataVersion,
@@ -866,10 +882,22 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
createChannel,
remoteIdentifier;
public boolean contentEquals(CharSequence c) {
private final Params[] params;
EventId(Params... params) {
this.params = params;
}
public boolean contentEquals(CharSequence c) {
return this.toString().contentEquals(c);
}
public Params[] params() {
return this.params;
}
}
}
src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java
+2 -2
View File
@@ -43,10 +43,10 @@ public class MapWireHandlerBuilder {
hasNext,
timeStamp,
channelId,
methodName,
eventName,
csp,
tid, // transactionId
result,
reply,
resultKey,
resultValue,
arg1,