diff --git a/pom.xml b/pom.xml
index ff058a85d..65c156db6 100755
--- a/pom.xml
+++ b/pom.xml
@@ -79,7 +79,7 @@
net.openhft
chronicle-bytes
- 1.0.0-alpha-SNAPSHOT
+ 1.0.1-alpha-SNAPSHOT
diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
index e1256fe73..a0d115ff6 100644
--- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
@@ -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
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
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
hub.outWire().writeDocument(false, new Consumer() {
@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
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
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 getMapped(@Nullable K key, @NotNull SerializableFunction function) {
throw new UnsupportedOperationException();
@@ -366,8 +370,7 @@ class ClientWiredStatelessChronicleMap
proxyReturnVoid(() -> hub.outWire().writeDocument(false, new Consumer() {
@Override
public void accept(WireOut wireOut) {
- wireOut.write(Fields.methodName).text(clear.toString());
-
+ wireOut.writeEventName(clear);
}
}
));
@@ -410,7 +413,7 @@ class ClientWiredStatelessChronicleMap
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
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
// 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
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
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
try {
tid = hub.writeHeader(startTime, channelID, hub.outWire());
- hub.outWire().writeDocument(false, new Consumer() {
- @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
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() {
- @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() {
+ @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
hub.outWire().writeDocument(false, new Consumer() {
@Override
public void accept(WireOut wireOut) {
- wireOut.write(Fields.methodName).text(methodName.toString());
+ wireOut.writeEventName(() -> methodName);
}
}
);
@@ -802,7 +825,7 @@ class ClientWiredStatelessChronicleMap
hub.outWire().writeDocument(false, new Consumer() {
@Override
public void accept(WireOut wireOut) {
- wireOut.write(Fields.methodName).text(methodName.toString());
+ wireOut.writeEventName(() -> methodName);
}
}
);
@@ -847,7 +870,7 @@ class ClientWiredStatelessChronicleMap
hub.outWire().writeDocument(false, new Consumer() {
@Override
public void accept(WireOut wireOut) {
- wireOut.write(Fields.methodName).text(methodName.toString());
+ wireOut.writeEventName(() -> methodName);
}
}
);
@@ -862,8 +885,8 @@ class ClientWiredStatelessChronicleMap
@Nullable
- private R proxyReturnObject(
- @NotNull final EventId methodName, K key, V value, Class resultType) {
+ private R proxyReturnTypedObject(
+ @NotNull final EventId methodName, Class resultType, Object... args) {
final long startTime = System.currentTimeMillis();
long tid;
@@ -875,18 +898,7 @@ class ClientWiredStatelessChronicleMap
tid = hub.writeHeader(startTime, channelID, hub.outWire());
- hub.outWire().writeDocument(false, new Consumer() {
- @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
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
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() {
- @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
hub.outWire().writeDocument(false, new Consumer() {
@Override
public void accept(WireOut wireOut) {
- wireOut.write(Fields.methodName).text(methodName.toString());
+ wireOut.writeEventName(methodName);
}
}
);
@@ -976,13 +979,13 @@ class ClientWiredStatelessChronicleMap
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() {
@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
}
-/* 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 {
diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
index 656085ba2..3b8cf31ef 100644
--- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
@@ -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() {
@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 {
diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java
index 7f6deb2d8..72e91fbe3 100644
--- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java
@@ -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 implements WireHandler, Consumer {
@Override
public void accept(Iterator iterator) {
- outWire.write(result);
+ outWire.write(reply);
Bytes bytes = outWire.bytes();
bytes.write(iterator.next());
}
@@ -178,7 +176,8 @@ class MapWireHandler implements WireHandler, Consumer {
}
private void writeChunked(
- long tid, @NotNull final Function function,
+ long tid,
+ @NotNull final Function function,
@NotNull final Consumer c) throws StreamCorruptedException {
writeChunked(tid, function, c, Long.MAX_VALUE);
}
@@ -195,6 +194,7 @@ class MapWireHandler implements WireHandler, Consumer {
}
}
+
@SuppressWarnings("UnusedReturnValue")
void onEvent() throws StreamCorruptedException {
@@ -209,8 +209,11 @@ class MapWireHandler implements WireHandler, Consumer {
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 implements WireHandler, Consumer {
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) 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 implements WireHandler, Consumer {
// 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) 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) 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) 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 implements WireHandler, Consumer {
System.out.println("--------------------------------------------\n" +
"server writes:\n\n");
} else {
+
+
System.out.println("--------------------------------------------\n" +
"server writes:\n\n" +
Bytes.toDebugString(bytes, SIZE_OF_SIZE, len));
@@ -541,7 +574,7 @@ class MapWireHandler implements WireHandler, Consumer {
if (isNull)
return;
- outWire.write(result);
+ outWire.write(reply);
Bytes bytes = outWire.bytes();
bytes.write(fromBytes);
@@ -570,46 +603,34 @@ class MapWireHandler implements WireHandler, Consumer {
/**
* 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 implements WireHandler, Consumer {
*/
// 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 implements WireHandler, Consumer {
}
@SuppressWarnings("SameReturnValue")
- private void writeValue(final Function f) {
- writeValueFromBytes(b -> toByteArray(f.apply(b)));
+ private void writeValue(final Function, byte[]> f) {
+ writeValueFromBytes(b -> {
+ final Map delegate = (Map) b.delegate;
+ return f.apply(delegate);
+ });
}
@SuppressWarnings("SameReturnValue")
@@ -710,7 +718,7 @@ class MapWireHandler implements WireHandler, Consumer {
write(b -> {
- byte[] result = f.apply((ChronicleMap) b.delegate);
+ byte[] result = f.apply((ChronicleMap) b.delegate);
boolean isNull = result == null;
outWire.write(resultIsNull).bool(isNull);
@@ -719,7 +727,7 @@ class MapWireHandler implements WireHandler, Consumer {
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 implements WireHandler, Consumer {
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 implements WireHandler, Consumer {
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;
+
+ }
}
}
+
diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java
index 1c57d87db..ad43e581b 100644
--- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java
+++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java
@@ -43,10 +43,10 @@ public class MapWireHandlerBuilder {
hasNext,
timeStamp,
channelId,
- methodName,
+ eventName,
csp,
tid, // transactionId
- result,
+ reply,
resultKey,
resultValue,
arg1,