From 92741c26fa8cea7261469612105b2e5d88295f01 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 31 Mar 2015 18:19: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 | 149 +++++++++++++++--- .../ClientWiredStatelessTcpConnectionHub.java | 20 ++- .../openhft/chronicle/map/MapWireHandler.java | 9 +- .../chronicle/map/MapWireHandlerBuilder.java | 2 +- 4 files changed, 149 insertions(+), 31 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 788c63c57..e1256fe73 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -23,6 +23,7 @@ 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.Wire; +import net.openhft.chronicle.wire.WireOut; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; @@ -275,8 +276,15 @@ class ClientWiredStatelessChronicleMap public void createChannel(short channelID) { proxyReturnVoid( () -> { - hub.outWire().write(Fields.methodName).text(createChannel.toString()); - hub.outWire().write(Fields.arg1).int16(channelID); + + 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); + } + }); + }); } @@ -301,7 +309,7 @@ class ClientWiredStatelessChronicleMap assert hub.outBytesLock().isHeldByCurrentThread(); hub.markSize(wire); hub.startTime(startTime); - wire.write(Fields.type).text("MAP"); + wire.write(Fields.csp).text("MAP"); wire.write(Fields.tid).int64(tid); wire.write(Fields.timeStamp).int64(startTime); wire.write(Fields.channelId).int16(channelID); @@ -313,8 +321,10 @@ class ClientWiredStatelessChronicleMap hub.outWire().write(Fields.hasNext).bool(iterator.hasNext()); - writeField(Fields.arg1, e.getKey()); - writeField(Fields.arg2, e.getValue()); + + // todo + // writeField(wireOut, Fields.arg1, e.getKey()); + // writeField(wireOut, Fields.arg2, e.getValue()); final Bytes bytes = wire.bytes(); if (bytes.remaining() < bytes.capacity() * 0.5) @@ -353,7 +363,15 @@ class ClientWiredStatelessChronicleMap } public void clear() { - proxyReturnVoid(() -> writeField(Fields.methodName, clear.toString())); + proxyReturnVoid(() -> hub.outWire().writeDocument(false, new Consumer() { + @Override + public void accept(WireOut wireOut) { + wireOut.write(Fields.methodName).text(clear.toString()); + + } + } + )); + } @@ -525,12 +543,12 @@ class ClientWiredStatelessChronicleMap } - private void writeField(Fields fieldName, Object value) { - writeField(fieldName, value, hub.outWire()); + private void writeField(WireOut wireOut, Fields fieldName, Object value) { + writeField(fieldName, value, wireOut); } - private void writeField(Fields fieldName, Object value, Wire wire) { + private void writeField(Fields fieldName, Object value, WireOut wire) { assert hub.outBytesLock().isHeldByCurrentThread(); assert !hub.inBytesLock().isHeldByCurrentThread(); @@ -681,9 +699,17 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().write(Fields.methodName).text(methodName.toString()); - writeField(Fields.arg1, key); - writeField(Fields.arg2, value); + + 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); + } + } + ); + hub.writeSocket(hub.outWire()); } finally { @@ -703,10 +729,18 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().write(Fields.methodName).text(methodName.toString()); - writeField(Fields.arg1, key); - writeField(Fields.arg2, value1); - writeField(Fields.arg3, value2); + + 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.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -738,7 +772,16 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().write(Fields.methodName).text(methodName.toString()); + + hub.outWire().writeDocument(false, new Consumer() { + @Override + public void accept(WireOut wireOut) { + wireOut.write(Fields.methodName).text(methodName.toString()); + } + } + ); + + hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -756,7 +799,13 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().write(Fields.methodName).text(methodName.toString()); + hub.outWire().writeDocument(false, new Consumer() { + @Override + public void accept(WireOut wireOut) { + wireOut.write(Fields.methodName).text(methodName.toString()); + } + } + ); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -794,7 +843,15 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().write(Fields.methodName).text(methodName.toString()); + + hub.outWire().writeDocument(false, new Consumer() { + @Override + public void accept(WireOut wireOut) { + wireOut.write(Fields.methodName).text(methodName.toString()); + } + } + ); + hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -817,9 +874,19 @@ class ClientWiredStatelessChronicleMap assert !hub.inBytesLock().isHeldByCurrentThread(); tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().write(Fields.methodName).text(methodName.toString()); - writeField(Fields.arg1, key); - writeField(Fields.arg2, value); + + 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); + + } + } + ); + + hub.writeSocket(hub.outWire()); } finally { @@ -861,8 +928,19 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().write(Fields.methodName).text(methodName.toString()); - writeField(Fields.arg1, arg1); + + hub.outWire().writeDocument(false, new Consumer() { + @Override + public void accept(WireOut wireOut) { + wireOut.write(Fields.methodName).text(methodName.toString()); + writeField(wireOut, Fields.arg1, arg1); + + + } + } + ); + + hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -876,8 +954,17 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); - hub.outWire().write(Fields.methodName).text(methodName.toString()); + hub.outWire().writeDocument(false, new Consumer() { + @Override + public void accept(WireOut wireOut) { + wireOut.write(Fields.methodName).text(methodName.toString()); + } + } + ); + hub.writeSocket(hub.outWire()); + + } finally { hub.outBytesLock().unlock(); } @@ -890,7 +977,17 @@ class ClientWiredStatelessChronicleMap try { tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); - hub.outWire().write(Fields.arg1).int64(arg1); + + hub.outWire().writeDocument(false, new Consumer() { + @Override + public void accept(WireOut wireOut) { + wireOut.write(Fields.methodName).text(methodName.toString()); + writeField(wireOut, Fields.arg1, arg1); + } + } + ); + + hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index 629569235..656085ba2 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -27,8 +27,8 @@ import net.openhft.chronicle.map.MapWireHandler.EventId; import net.openhft.chronicle.network2.event.EventGroup; import net.openhft.chronicle.wire.TextWire; import net.openhft.chronicle.wire.Wire; -import net.openhft.chronicle.wire.WireIn; import net.openhft.chronicle.wire.WireOut; +import net.openhft.chronicle.wire.Wires; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import org.slf4j.Logger; @@ -373,9 +373,14 @@ public class ClientWiredStatelessTcpConnectionHub { System.out.println("\n--------------------------------\n" + "client read:\n\n" + Bytes.toDebugString(bytes)); + int headerlen = bytes.readVolatileInt(); + + assert !Wires.isData(headerlen); + 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; @@ -593,7 +598,13 @@ public class ClientWiredStatelessTcpConnectionHub { outBytesLock().lock(); try { long tid = writeHeader(startTime, channelID, wire); - wire.write(MapWireHandlerBuilder.Fields.methodName).text(methodName.toString()); + wire.writeDocument(false, new Consumer() { + @Override + public void accept(WireOut wireOut) { + wireOut.write(MapWireHandlerBuilder.Fields.methodName).text(methodName.toString()); + } + }); + writeSocket(wire); return tid; } finally { @@ -649,13 +660,16 @@ public class ClientWiredStatelessTcpConnectionHub { wire.writeDocument(true, new Consumer() { @Override public void accept(WireOut wireOut) { - wireOut.write(type).text("MAP"); + 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; } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index b9f575f1e..7f6deb2d8 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -204,6 +204,11 @@ class MapWireHandler implements WireHandler, Consumer { timestamp = inWire.read(timeStamp).int64(); channelId = inWire.read(Fields.channelId).int16(); + final Bytes bytes1 = inWire.bytes(); + + final int body = bytes1.readVolatileInt(); + assert Wires.isData(body); + final StringBuilder methodName = Wires.acquireStringBuilder(); inWire.read(Fields.methodName).text(methodName); @@ -252,8 +257,10 @@ class MapWireHandler implements WireHandler, Consumer { return; } + + outWire.writeDocument(true, wire -> outWire.write(Fields.tid).int64(tid)); + // write the transaction id - outWire.write(Fields.tid).int64(tid); if (createChannel.contentEquals(methodName)) { diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java index dae60acba..1c57d87db 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java @@ -44,7 +44,7 @@ public class MapWireHandlerBuilder { timeStamp, channelId, methodName, - type, + csp, tid, // transactionId result, resultKey,