From a0c22f8795078ffae7091f7c08cc939976e7ab93 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 6 Apr 2015 18:12:10 +0100 Subject: [PATCH] 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 --- .../map/ClientWiredStatelessChronicleMap.java | 74 ++++++---- .../ClientWiredStatelessTcpConnectionHub.java | 67 ++++++--- .../openhft/chronicle/map/MapWireHandler.java | 133 ++++++++++-------- 3 files changed, 174 insertions(+), 100 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index a0d115ff6..ff3879233 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -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 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 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 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> entrySet() { return toMap().entrySet(); @@ -448,7 +456,7 @@ class ClientWiredStatelessChronicleMap 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 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 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 // 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 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 } private E readObject(Fields argName, Wire wireIn, E usingValue, Class 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 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 hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, tid); + checkIsData(wireIn); return wireIn.read(Fields.reply).int32(); } finally { hub.inBytesLock().unlock(); diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index 3b8cf31ef..26b798af9 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -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 { diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 72e91fbe3..b4fd57014 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -200,7 +200,7 @@ class MapWireHandler implements WireHandler, Consumer { // 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 implements WireHandler, Consumer { 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 implements WireHandler, Consumer { try { - if (putWithoutAcc.contentEquals(eventName)) { valueIn.marshallable(wire -> { @@ -271,12 +267,9 @@ class MapWireHandler implements WireHandler, Consumer { 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 implements WireHandler, Consumer { } - if (applicationVersion.contentEquals(eventName)) { + if (getApplicationVersion.contentEquals(eventName)) { write(b -> outWire.write(reply).text(applicationVersion())); return; } @@ -449,10 +442,17 @@ class MapWireHandler implements WireHandler, Consumer { "server writes:\n\n"); } 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 implements WireHandler, Consumer { 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 implements WireHandler, Consumer { write(b -> { byte[] result = f.apply((ChronicleMap) 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 implements WireHandler, Consumer { } 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 implements WireHandler, Consumer { } 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 implements WireHandler, Consumer { 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 implements WireHandler, Consumer { putIfAbsent(key, value), removeWithValue(key, value), toString, - applicationVersion, + getApplicationVersion, persistedDataVersion, putAll, putAllWithoutAcc,