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

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-06 18:12:10 +01:00
parent 65fe9415b9
commit a0c22f8795
3 changed files with 174 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+47 -27
View File
@@ -21,10 +21,7 @@ package net.openhft.chronicle.map;
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 net.openhft.chronicle.wire.*;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
@@ -409,6 +406,8 @@ class ClientWiredStatelessChronicleMap<K, V>
try {
final Wire wireIn = hub.proxyReply(timeoutTime, tid);
checkIsData(wireIn);
final Bytes<?> bytes = wireIn.bytes();
while (bytes.remaining() > 0) {
boolean bool = wireIn.read(Fields.hasNext).bool();
@@ -418,7 +417,7 @@ class ClientWiredStatelessChronicleMap<K, V>
break OUTER;
// todo process the exception
boolean isException = wireIn.read(Fields.isException).bool();
// boolean isException = wireIn.read(Fields.isException).bool();
}
} finally {
@@ -428,6 +427,15 @@ class ClientWiredStatelessChronicleMap<K, V>
return usingCollection;
}
private void checkIsData(Wire wireIn) {
int datalen = wireIn.bytes().readVolatileInt();
// if (Wires.isData(datalen))
// throw new IllegalStateException("expecting a data blob, from ->" + Bytes.toDebugString
// (wireIn.bytes(), 0, wireIn.bytes().limit()));
//
}
@NotNull
public Set<Map.Entry<K, V>> entrySet() {
return toMap().entrySet();
@@ -448,7 +456,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, tid);
checkIsData(wireIn);
final Bytes<?> bytes = wireIn.bytes();
while (bytes.remaining() > 0) {
boolean bool = wireIn.read(Fields.hasNext).bool();
@@ -460,10 +468,10 @@ class ClientWiredStatelessChronicleMap<K, V>
break OUTER;
if (bytes.remaining() > 0) {
/* if (bytes.remaining() > 0) {
// todo process the exception
boolean isException = wireIn.read(Fields.isException).bool();
}
// boolean isException = wireIn.read(Fields.isException).bool();
}*/
}
} finally {
@@ -496,7 +504,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, tid);
checkIsData(wireIn);
final Bytes<?> bytes = wireIn.bytes();
while (bytes.remaining() > 0) {
boolean bool = wireIn.read(Fields.hasNext).bool();
@@ -539,7 +547,9 @@ class ClientWiredStatelessChronicleMap<K, V>
// receive
hub.inBytesLock().lock();
try {
return hub.proxyReply(timeoutTime, tid).read(Fields.reply).int64();
final Wire wire = hub.proxyReply(timeoutTime, tid);
checkIsData(wire);
return wire.read(Fields.reply).int64();
} finally {
hub.inBytesLock().unlock();
}
@@ -586,9 +596,9 @@ class ClientWiredStatelessChronicleMap<K, V>
try {
final Wire wireIn = hub.proxyReply(timeoutTime, tid);
checkIsData(wireIn);
if (wireIn.read(Fields.resultIsNull).bool())
return null;
return readV(argName, wireIn, usingValue);
@@ -606,56 +616,64 @@ class ClientWiredStatelessChronicleMap<K, V>
}
private <E> E readObject(Fields argName, Wire wireIn, E usingValue, Class<E> clazz) {
final ValueIn valueIn = wireIn.read(argName);
if (valueIn.isNull())
return null;
if (StringBuilder.class.isAssignableFrom(clazz)) {
wireIn.read(argName).text((StringBuilder) usingValue);
valueIn.text((StringBuilder) usingValue);
return usingValue;
} else if (Marshallable.class.isAssignableFrom(clazz)) {
final E v;
if (usingValue == null)
try {
E v = clazz.newInstance();
wireIn.read(argName).marshallable((Marshallable) v);
return v;
v = clazz.newInstance();
} catch (Exception e) {
throw new RuntimeException(e);
}
else
v = usingValue;
wireIn.read(argName).marshallable((Marshallable) usingValue);
return usingValue;
valueIn.marshallable((Marshallable) v);
return v;
} else if (String.class.isAssignableFrom(clazz)) {
//noinspection unchecked
return (E) wireIn.read(argName).text();
return (E) valueIn.text();
} else if (Long.class.isAssignableFrom(clazz)) {
//noinspection unchecked
return (E) (Long) wireIn.read(argName).int64();
return (E) (Long) valueIn.int64();
} else if (Double.class.isAssignableFrom(clazz)) {
//noinspection unchecked
return (E) (Double) wireIn.read(argName).float64();
return (E) (Double) valueIn.float64();
} else if (Integer.class.isAssignableFrom(clazz)) {
//noinspection unchecked
return (E) (Integer) wireIn.read(argName).int32();
return (E) (Integer) valueIn.int32();
} else if (Float.class.isAssignableFrom(clazz)) {
//noinspection unchecked
return (E) (Float) wireIn.read(argName).float32();
return (E) (Float) valueIn.float32();
} else if (Short.class.isAssignableFrom(clazz)) {
//noinspection unchecked
return (E) (Short) wireIn.read(argName).int16();
return (E) (Short) valueIn.int16();
} else if (Character.class.isAssignableFrom(clazz)) {
//noinspection unchecked
final String text = wireIn.read(argName).text();
final String text = valueIn.text();
if (text == null || text.length() == 0)
return null;
return (E) (Character) text.charAt(0);
} else if (Byte.class.isAssignableFrom(clazz)) {
//noinspection unchecked
return (E) (Byte) wireIn.read(argName).int8();
return (E) (Byte) valueIn.int8();
} else {
@@ -673,6 +691,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, tid);
checkIsData(wireIn);
return wireIn.read(Fields.reply).bool();
} finally {
hub.inBytesLock().unlock();
@@ -688,6 +707,7 @@ class ClientWiredStatelessChronicleMap<K, V>
hub.inBytesLock().lock();
try {
final Wire wireIn = hub.proxyReply(timeoutTime, tid);
checkIsData(wireIn);
return wireIn.read(Fields.reply).int32();
} finally {
hub.inBytesLock().unlock();
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+49 -18
View File
@@ -42,7 +42,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import static net.openhft.chronicle.map.MapWireHandler.EventId.applicationVersion;
import static net.openhft.chronicle.map.MapWireHandler.EventId.getApplicationVersion;
import static net.openhft.chronicle.map.MapWireHandler.SIZE_OF_SIZE;
import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*;
@@ -51,7 +51,7 @@ import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*;
*/
public class ClientWiredStatelessTcpConnectionHub {
private static final Logger LOG = LoggerFactory.getLogger(StatelessChronicleMap.class);
private static final Logger LOG = LoggerFactory.getLogger(ClientWiredStatelessTcpConnectionHub.class);
protected final String name;
protected final InetSocketAddress remoteAddress;
@@ -242,7 +242,7 @@ public class ClientWiredStatelessTcpConnectionHub {
@NotNull
public String serverApplicationVersion(short channelID) {
TextWire wire = new TextWire(Bytes.elasticByteBuffer());
String result = proxyReturnString(applicationVersion, channelID, wire);
String result = proxyReturnString(getApplicationVersion, channelID, wire);
return (result == null) ? "" : result;
}
@@ -318,10 +318,10 @@ public class ClientWiredStatelessTcpConnectionHub {
final Wire wire = proxyReplyThrowable(timeoutTime, tid);
// 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);
}
// if (wire.read(() -> "isException").bool()) {
// final String text = wire.read(() -> "exception").text();
// throw new RuntimeException(text);
// }
return wire;
} catch (IOException e) {
close();
@@ -359,12 +359,17 @@ public class ClientWiredStatelessTcpConnectionHub {
// reads just the size
readSocket(SIZE_OF_SIZE, timeoutTime);
final int messageSize =
bytes.readUnsignedShort(bytes.position());
// System.out.println("in=" + Bytes.toHex(intWire.bytes()));
assert messageSize > 0 : "Invalid message size " + messageSize;
assert messageSize < 16 << 20 : "Invalid message size " + messageSize;
final int messageSize = bytes.readUnsignedShort(bytes.position());
System.out.println("messageSize=" + messageSize);
try {
assert messageSize > 0 : "Invalid message size " + messageSize;
assert messageSize < 1024 : "Invalid message size " + messageSize;
} catch (AssertionError e) {
assert messageSize < 1024 : "Invalid message size " + messageSize;
}
final int remainingBytes0 = messageSize;
readSocket(remainingBytes0, timeoutTime);
@@ -372,8 +377,19 @@ public class ClientWiredStatelessTcpConnectionHub {
bytes.skip(SIZE_OF_SIZE);
bytes.limit(bytes.position() + messageSize);
System.out.println("\n--------------------------------\n" +
"client read:\n\n" + Bytes.toDebugString(bytes));
try {
System.out.println("\n--------------------------------\nclient reads\n" +
Wires.fromSizePrefixedBlobs(bytes));
} catch (Exception e) {
System.out.println("\n--------------------------------\nclient reads\n" +
"client read:\n\n" + Bytes.toDebugString(bytes));
}
System.out.println("client read:\n\n" + Bytes.toDebugString(bytes));
int headerlen = bytes.readVolatileInt();
@@ -573,10 +589,18 @@ public class ClientWiredStatelessTcpConnectionHub {
bytes.limit(outBuffer.limit());
bytes.position(outBuffer.position());
System.out.println("\n--------------------------------------------\n" +
"client writes:\n\n" +
// Bytes.toDebugString(outBuffer));
Wires.fromSizePrefixedBlobs(bytes));
try {
System.out.println("\n--------------------------------------------\n" +
"client writes:\n\n" +
// Bytes.toDebugString(outBuffer));
Wires.fromSizePrefixedBlobs(bytes));
} catch (Exception e) {
System.out.println("\n--------------------------------------------\n" +
"client writes:\n\n" +
// Bytes.toDebugString(outBuffer));
Bytes.toDebugString(bytes));
}
} finally {
bytes.limit(limit);
bytes.position(position);
@@ -647,7 +671,14 @@ public class ClientWiredStatelessTcpConnectionHub {
// receive
inBytesLock().lock();
try {
return proxyReply(timeoutTime, tid).read(reply).text();
final Wire wire = proxyReply(timeoutTime, tid);
int datalen = wire.bytes().readVolatileInt();
assert Wires.isData(datalen);
return wire.read(reply).text();
} catch (Exception e) {
throw new RuntimeException(e);
} finally {
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+78 -55
View File
@@ -200,7 +200,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
// it is assumed by this point that the buffer has all the bytes in it for this message
long tid = inWire.read(Fields.tid).int64();
final long tid = inWire.read(Fields.tid).int64();
timestamp = inWire.read(timeStamp).int64();
channelId = inWire.read(Fields.channelId).int16();
@@ -210,11 +210,8 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
assert Wires.isData(body);
final StringBuilder eventName = Wires.acquireStringBuilder();
final ValueIn valueIn = inWire.readEventName(eventName);
if (!incompleteWork.isEmpty()) {
Runnable runnable = incompleteWork.get(tid);
if (runnable != null) {
@@ -228,7 +225,6 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
try {
if (putWithoutAcc.contentEquals(eventName)) {
valueIn.marshallable(wire -> {
@@ -271,12 +267,9 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
return;
}
outWire.writeDocument(true, wire -> outWire.write(Fields.tid).int64(tid));
// write the transaction id
if (createChannel.contentEquals(eventName)) {
writeVoid(() -> {
@@ -420,7 +413,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
if (applicationVersion.contentEquals(eventName)) {
if (getApplicationVersion.contentEquals(eventName)) {
write(b -> outWire.write(reply).text(applicationVersion()));
return;
}
@@ -449,10 +442,17 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
"server writes:\n\n<EMPTY>");
} else {
try {
System.out.println("--------------------------------------------\n" +
"server writes:\n\n" +
System.out.println("--------------------------------------------\n" +
"server writes:\n\n" +
Bytes.toDebugString(bytes, SIZE_OF_SIZE, len));
Wires.fromSizePrefixedBlobs(bytes, SIZE_OF_SIZE, len));
//Bytes.toDebugString(bytes, SIZE_OF_SIZE, len));
} catch (Exception e) {
System.out.println("--------------------------------------------\n" +
"server writes:\n\n" +
Bytes.toDebugString(bytes, SIZE_OF_SIZE, len));
}
}
}
}
@@ -570,13 +570,14 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
byte[] fromBytes = f.apply(b);
boolean isNull = fromBytes == null || fromBytes.length == 0;
outWire.write(resultIsNull).bool(isNull);
if (isNull)
return;
outWire.write(reply);
Bytes<?> bytes = outWire.bytes();
bytes.write(fromBytes);
if (isNull)
outWire.writeValue().text(null);
else {
Bytes<?> bytes = outWire.bytes();
bytes.write(fromBytes);
}
});
}
@@ -719,17 +720,15 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
write(b -> {
byte[] result = f.apply((ChronicleMap<byte[], byte[]>) b.delegate);
boolean isNull = result == null;
outWire.write(resultIsNull).bool(isNull);
if (isNull)
return;
isNull = result.length == 0;
boolean isNull = result == null || result.length == 0;
outWire.write(Fields.reply);
final Bytes<?> bytes = outWire.bytes();
bytes.write(result);
if (isNull)
outWire.writeValue().text(null);
else
bytes.write(result);
});
@@ -747,21 +746,31 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
bytesMap.output = null;
final Bytes<?> bytes = outWire.bytes();
bytes.mark();
outWire.write(isException).bool(false);
// final Bytes<?> bytes = outWire.bytes();
// bytes.mark();
// outWire.write(isException).bool(false);
outWire.writeDocument(false, out -> {
try {
outWire.bytes().mark();
c.accept(bytesMap);
} catch (Exception e) {
// bytes.reset();
// the idea of wire is that is platform independent,
// so we wil have to send the exception as a String
outWire.bytes().reset();
WireOut o = out.write(reply)
.type(e.getClass().getSimpleName());
if (e.getMessage() != null) {
o.writeValue().text(e.getMessage());
}
LOG.error("", e);
}
});
try {
c.accept(bytesMap);
} catch (Exception e) {
bytes.reset();
// the idea of wire is that is platform independent,
// so we wil have to send the exception as a String
outWire.write(isException).bool(true);
outWire.write(exception).text(toString(e));
LOG.error("", e);
return;
}
}
@@ -776,22 +785,25 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
bytesMap.output = null;
final Bytes<?> bytes = outWire.bytes();
bytes.mark();
try {
r.call();
outWire.write(isException).bool(false);
} catch (Exception e) {
bytes.reset();
// bytes.reset();
// the idea of wire is that is platform independent,
// so we wil have to send the exception as a String
outWire.write(isException).bool(true);
outWire.write(exception).text(toString(e));
outWire.writeDocument(false, out -> {
out.write(reply)
.type(e.getClass().getSimpleName())
.writeValue().text(e.getMessage());
});
LOG.error("", e);
return;
}
}
@SuppressWarnings("SameReturnValue")
@@ -805,15 +817,26 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
return;
}
bytesMap.output = null;
try {
process.accept(bytesMap);
outWire.write(isException).bool(false);
} catch (Exception e) {
outWire.write(isException).bool(true);
LOG.error("", e);
try {
outWire.bytes().mark();
process.accept(bytesMap);
} catch (Exception e) {
outWire.writeDocument(false, out -> {
// bytes.reset();
// the idea of wire is that is platform independent,
// so we wil have to send the exception as a String
outWire.bytes().reset();
out.write(reply)
.type(e.getClass().getSimpleName())
.writeValue().text(e.getMessage());
LOG.error("", e);
});
}
}
/**
@@ -870,7 +893,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
putIfAbsent(key, value),
removeWithValue(key, value),
toString,
applicationVersion,
getApplicationVersion,
persistedDataVersion,
putAll,
putAllWithoutAcc,