diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index ff3879233..41d302480 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -275,12 +275,9 @@ class ClientWiredStatelessChronicleMap proxyReturnVoid( () -> { - hub.outWire().writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { - hub.outWire().writeEventName(createChannel); - hub.outWire().writeValue().int16(channelID); - } + hub.outWire().writeDocument(false, wireOut -> { + hub.outWire().writeEventName(createChannel); + hub.outWire().writeValue().int16(channelID); }); }); @@ -364,12 +361,7 @@ class ClientWiredStatelessChronicleMap } public void clear() { - proxyReturnVoid(() -> hub.outWire().writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { - wireOut.writeEventName(clear); - } - } + proxyReturnVoid(() -> hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(clear) )); } @@ -761,35 +753,32 @@ class ClientWiredStatelessChronicleMap assert args.length > 0; - hub.outWire().writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { + hub.outWire().writeDocument(false, wireOut -> { - final ValueOut out = wireOut.writeEventName(methodName); + final ValueOut out = wireOut.writeEventName(methodName); - final MapWireHandler.Params[] paramNames = methodName.params(); + final MapWireHandler.Params[] paramNames = methodName.params(); - if (paramNames.length == 1) { - writeField(out, args[0]); - return; - } + if (paramNames.length == 1) { + writeField(out, args[0]); + return; + } - assert args.length == paramNames.length : - "methodName=" + methodName + - ", args.length=" + args.length + - ", paramNames.length=" + paramNames.length; + assert args.length == paramNames.length : + "methodName=" + methodName + + ", args.length=" + args.length + + ", paramNames.length=" + paramNames.length; - out.marshallable(m -> { + out.marshallable(m -> { - for (int i = 0; i < paramNames.length; i++) { - final ValueOut vo = m.write(paramNames[i]); - writeField(vo, args[i]); - } - - }); - - } + for (int i = 0; i < paramNames.length; i++) { + final ValueOut vo = m.write(paramNames[i]); + writeField(vo, args[i]); } + + }); + + } ); } @@ -816,12 +805,7 @@ class ClientWiredStatelessChronicleMap try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { - wireOut.writeEventName(() -> methodName); - } - } + hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName) ); @@ -842,12 +826,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { - wireOut.writeEventName(() -> methodName); - } - } + hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName) ); hub.writeSocket(hub.outWire()); } finally { @@ -887,12 +866,7 @@ class ClientWiredStatelessChronicleMap try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { - wireOut.writeEventName(() -> methodName); - } - } + hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(() -> methodName) ); hub.writeSocket(hub.outWire()); @@ -977,12 +951,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { - wireOut.writeEventName(methodName); - } - } + hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(methodName) ); hub.writeSocket(hub.outWire()); @@ -1001,13 +970,10 @@ class ClientWiredStatelessChronicleMap tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.eventName).text(methodName.toString()); - hub.outWire().writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { - final ValueOut out = wireOut.writeEventName(methodName); - writeField(out, arg1); - } - } + hub.outWire().writeDocument(false, wireOut -> { + final ValueOut out = wireOut.writeEventName(methodName); + writeField(out, arg1); + } ); diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index 26b798af9..80f90c974 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -359,11 +359,9 @@ public class ClientWiredStatelessTcpConnectionHub { // reads just the size readSocket(SIZE_OF_SIZE, timeoutTime); -// System.out.println("in=" + Bytes.toHex(intWire.bytes())); - 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; @@ -378,18 +376,13 @@ public class ClientWiredStatelessTcpConnectionHub { bytes.limit(bytes.position() + messageSize); - 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(); @@ -633,12 +626,7 @@ public class ClientWiredStatelessTcpConnectionHub { outBytesLock().lock(); try { long tid = writeHeader(startTime, channelID, wire); - wire.writeDocument(false, new Consumer() { - @Override - public void accept(WireOut wireOut) { - wireOut.writeEventName(methodName); - } - }); + wire.writeDocument(false, wireOut -> wireOut.writeEventName(methodName)); writeSocket(wire); return tid; @@ -699,20 +687,14 @@ public class ClientWiredStatelessTcpConnectionHub { startTime(startTime); long tid = nextUniqueTransaction(startTime); - wire.writeDocument(true, new Consumer() { - @Override - public void accept(WireOut wireOut) { - wireOut.write(csp).text("MAP"); - wireOut.write(MapWireHandlerBuilder.Fields.tid).int64(tid); - wireOut.write(timeStamp).int64(startTime); - wireOut.write(channelId).int16(channelID); - } + wire.writeDocument(true, wireOut -> { + wireOut.write(csp).text("MAP"); + wireOut.write(MapWireHandlerBuilder.Fields.tid).int64(tid); + wireOut.write(timeStamp).int64(startTime); + wireOut.write(channelId).int16(channelID); }); - final Bytes bytes = wire.bytes(); - System.out.println(Bytes.toDebugString(wire.bytes(), 0, bytes.position())); - - return tid; + return tid; } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index b4fd57014..5676e2144 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -66,6 +66,7 @@ class MapWireHandler implements WireHandler, Consumer { private final Supplier>> mapFactory; private Wire inWire = null; private Wire outWire = null; + private final Consumer writeElement = new Consumer>() { @Override @@ -75,6 +76,7 @@ class MapWireHandler implements WireHandler, Consumer { bytes.write(iterator.next()); } }; + private final Consumer writeEntry = new Consumer>>() { @Override @@ -93,7 +95,7 @@ class MapWireHandler implements WireHandler, Consumer { }; private WireHandlers publishLater; private byte localIdentifier; - private long timestamp; + private short channelId; private List channels; private ReplicationHub hub; @@ -201,7 +203,9 @@ class MapWireHandler implements WireHandler, Consumer { // it is assumed by this point that the buffer has all the bytes in it for this message final long tid = inWire.read(Fields.tid).int64(); - timestamp = inWire.read(timeStamp).int64(); + + // timestamp + inWire.read(timeStamp).int64(); channelId = inWire.read(Fields.channelId).int16(); final Bytes bytes1 = inWire.bytes(); @@ -330,8 +334,6 @@ class MapWireHandler implements WireHandler, Consumer { MapWireHandler.this.writeValue(b -> { final byte[] result = b.put(key1, value1); - - System.out.println("key='" + new String(b.keySet().iterator().next()) + "'"); return result; }); @@ -442,17 +444,11 @@ class MapWireHandler implements WireHandler, Consumer { "server writes:\n\n"); } else { - try { - System.out.println("--------------------------------------------\n" + - "server writes:\n\n" + - Wires.fromSizePrefixedBlobs(bytes, SIZE_OF_SIZE, len)); - //Bytes.toDebugString(bytes, SIZE_OF_SIZE, len)); - } catch (Exception e) { - System.out.println("--------------------------------------------\n" + + System.out.println("--------------------------------------------\n" + "server writes:\n\n" + - Bytes.toDebugString(bytes, SIZE_OF_SIZE, len)); - } + Wires.fromSizePrefixedBlobs(bytes, SIZE_OF_SIZE, len)); + } } } @@ -460,18 +456,6 @@ class MapWireHandler implements WireHandler, Consumer { } - // todo remove - private byte[] toByteArray(net.openhft.lang.io.Bytes bytes) { - if (bytes == null || bytes.remaining() == 0) - return new byte[]{}; - - if (bytes.remaining() > Integer.MAX_VALUE) - throw new BufferOverflowException(); - - byte[] result = new byte[(int) bytes.remaining()]; - bytes.write(result); - return result; - } // todo remove private byte[] toBytes(WireKey fieldName) { @@ -616,39 +600,25 @@ class MapWireHandler implements WireHandler, Consumer { final net.openhft.lang.io.Bytes bytes = DirectStore.allocate(inSize + 1024).bytes(); - valueIn.marshallable(new ReadMarshallable() { - @Override - public void readMarshallable(WireIn wire) throws IllegalStateException { - for (Params p : params) { + valueIn.marshallable(wire -> { - 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); - } + for (Params p : params) { + final ValueIn read = wire.read(p); + + final byte[] data = read.bytes(); + + bytes.writeStopBit(data.length); + bytes.write(data); } + }); return bytes.flip(); } - /** - * creates a lang buffer that holds just the payload of the args - * - * @param wire the inbound wire - * @return a new lang buffer containing the bytes of the args - */ - // todo remove this method - just added to get it to work for now - private byte[] toByteArray(@NotNull Wire wire, @NotNull Params field) { - - final ValueIn read = wire.read(field); - return toByteArray(read); - } private byte[] toByteArray(ValueIn valueIn) { @@ -839,15 +809,6 @@ class MapWireHandler implements WireHandler, Consumer { } - /** - * only used for debugging - */ - @SuppressWarnings("UnusedDeclaration") - private void showOutWire() { - Bytes bytes = outWire.bytes(); - System.out.println("pos=" + bytes.position() + ",bytes=" + - Bytes.toDebugString(bytes, 0, bytes.position())); - } /** * converts the exception into a String, so that it can be sent to c# clients @@ -863,7 +824,7 @@ class MapWireHandler implements WireHandler, Consumer { this.localIdentifier = localIdentifier; } - static enum Params implements WireKey { + enum Params implements WireKey { key, value, @@ -871,7 +832,7 @@ class MapWireHandler implements WireHandler, Consumer { newValue } - static enum EventId implements WireKey { + enum EventId implements WireKey { longSize, size,