From 6ffcdd45014e55cdb097d1e0a545b2e546f2d178 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 7 Apr 2015 17:25:41 +0100 Subject: [PATCH] CE-20 refactors --- ...ientWiredChronicleMapStatelessBuilder.java | 2 +- .../map/ClientWiredStatelessChronicleMap.java | 29 ++++++++----- .../ClientWiredStatelessTcpConnectionHub.java | 41 +++++++++---------- .../openhft/chronicle/map/MapWireHandler.java | 14 ++++++- 4 files changed, 53 insertions(+), 33 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java index e1bd407b4..cc89cdec1 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredChronicleMapStatelessBuilder.java @@ -129,7 +129,7 @@ public final class ClientWiredChronicleMapStatelessBuilder implements // todo clean this up if (hub == null) - hub = new ClientWiredStatelessTcpConnectionHub(this, localIdentifier, doHandShaking); + hub = new ClientWiredStatelessTcpConnectionHub(this, localIdentifier, doHandShaking, "MAP"); if (!used.getAndSet(true)) { return new ClientWiredStatelessChronicleMap( diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 3c31007ef..a5eee0f03 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -64,12 +64,14 @@ class ClientWiredStatelessChronicleMap // used with toString() private static final int MAX_NUM_ENTRIES = 20; + private String csp; public ClientWiredStatelessChronicleMap( @NotNull final ClientWiredChronicleMapStatelessBuilder config, @NotNull final Class kClass, @NotNull final Class vClass, short channelID) { this.channelID = channelID; - hub = config.hub; + this.csp = "//server/path/" + channelID + "#MAP"; + this.hub = config.hub; this.putReturnsNull = config.putReturnsNull(); this.removeReturnsNull = config.removeReturnsNull(); this.kClass = kClass; @@ -121,7 +123,7 @@ class ClientWiredStatelessChronicleMap public String serverApplicationVersion() { - return hub.serverApplicationVersion(channelID); + return hub.serverApplicationVersion(csp); } @Override @@ -219,7 +221,8 @@ class ClientWiredStatelessChronicleMap @NotNull public String serverPersistedDataVersion() { - return hub.proxyReturnString(persistedDataVersion, channelID); + + return hub.proxyReturnString(persistedDataVersion, csp); } public boolean isEmpty() { @@ -305,7 +308,7 @@ class ClientWiredStatelessChronicleMap wire.write(Fields.csp).text("MAP"); wire.write(Fields.tid).int64(tid); - wire.write(Fields.channelId).int16(channelID); + // wire.write(Fields.channelId).int16(channelID); hub.outWire().writeEventName(putAll); @@ -715,11 +718,12 @@ class ClientWiredStatelessChronicleMap @SuppressWarnings("SameParameterValue") private boolean proxyReturnBooleanV(@NotNull final EventId eventId, V value) { final long startTime = System.currentTimeMillis(); - return readBoolean(sendEvent(startTime, eventId, out -> writeField(out, value)), startTime); + return readBoolean(sendEvent(startTime, eventId, out -> writeField(out, value)), + startTime); } @SuppressWarnings("SameParameterValue") - private boolean proxyReturnBooleanK(@NotNull final EventId eventId, K key) { + private boolean proxyReturnBooleanK(@NotNull final EventId eventId, K key ) { final long startTime = System.currentTimeMillis(); return readBoolean(sendEvent(startTime, eventId, out -> writeField(out, key)), startTime); } @@ -733,7 +737,8 @@ class ClientWiredStatelessChronicleMap @SuppressWarnings("SameParameterValue") - private void proxyReturnVoid(@NotNull final EventId eventId, Consumer consumer) { + private void proxyReturnVoid(@NotNull final EventId eventId, + @Nullable final Consumer consumer) { final long startTime = System.currentTimeMillis(); long tid = sendEvent(startTime, eventId, consumer); readVoid(tid, startTime); @@ -753,7 +758,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().lock(); try { - tid = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = writeHeader(startTime); hub.outWire().writeDocument(false, wireOut -> { final ValueOut valueOut = wireOut.writeEventName(eventId); @@ -773,6 +778,10 @@ class ClientWiredStatelessChronicleMap return tid; } + private long writeHeader(long startTime) { + return hub.writeHeader(startTime, hub.outWire(), csp); + } + @SuppressWarnings("SameParameterValue") private boolean proxyReturnBoolean(@NotNull final EventId eventId) { final long startTime = System.currentTimeMillis(); @@ -855,7 +864,7 @@ class ClientWiredStatelessChronicleMap long tid; hub.outBytesLock().lock(); try { - tid = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = writeHeader(startTime); hub.outWire().writeDocument(false, wireOut -> wireOut.writeEventName(methodName)); hub.writeSocket(hub.outWire()); } finally { @@ -868,7 +877,7 @@ class ClientWiredStatelessChronicleMap long tid; hub.outBytesLock().lock(); try { - tid = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = writeHeader(startTime); hub.outWire().write(Fields.eventName).text(methodName.toString()); hub.outWire().writeDocument(false, wireOut -> { diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index 991c5f96a..60e0cd5d0 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.locks.ReentrantLock; 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.*; +import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.reply; /** * Created by Rob Austin @@ -57,7 +57,7 @@ public class ClientWiredStatelessTcpConnectionHub { protected final int tcpBufferSize; private final ReentrantLock inBytesLock = new ReentrantLock(true); private final ReentrantLock outBytesLock = new ReentrantLock(); - + private String type; @NotNull private final AtomicLong transactionID = new AtomicLong(0); @@ -88,7 +88,7 @@ public class ClientWiredStatelessTcpConnectionHub { public ClientWiredStatelessTcpConnectionHub( ClientWiredChronicleMapStatelessBuilder config, byte localIdentifier, - boolean doHandShaking) { + boolean doHandShaking, String cspType) { this.localIdentifier = localIdentifier; this.doHandShaking = doHandShaking; this.remoteAddress = config.remoteAddress(); @@ -96,8 +96,9 @@ public class ClientWiredStatelessTcpConnectionHub { this.config = config; this.name = config.name(); this.timeoutMs = config.timeoutMs(); + this.type = cspType; attemptConnect(remoteAddress); - checkVersion(config.channelID()); + } private synchronized void attemptConnect(final InetSocketAddress remoteAddress) { @@ -127,9 +128,9 @@ public class ClientWiredStatelessTcpConnectionHub { return outBytesLock; } - protected void checkVersion(short channelID) { + protected void checkVersion(@NotNull String csp) { - final String serverVersion = serverApplicationVersion(channelID); + final String serverVersion = serverApplicationVersion(csp); final String clientVersion = clientVersion(); if (!serverVersion.equals(clientVersion)) { @@ -137,7 +138,7 @@ public class ClientWiredStatelessTcpConnectionHub { "Stateless-Client are on different " + "versions, " + " we suggest that you use the same version, server=" + - serverApplicationVersion(channelID) + ", " + + serverApplicationVersion(csp) + ", " + "client=" + clientVersion); } } @@ -192,7 +193,7 @@ public class ClientWiredStatelessTcpConnectionHub { } clientChannel = result; - checkVersion(config.channelID()); + checkVersion("//#" + type); } @@ -238,9 +239,9 @@ public class ClientWiredStatelessTcpConnectionHub { } @NotNull - public String serverApplicationVersion(short channelID) { + public String serverApplicationVersion(@NotNull String csp) { TextWire wire = new TextWire(Bytes.elasticByteBuffer()); - String result = proxyReturnString(getApplicationVersion, channelID, wire); + String result = proxyReturnString(getApplicationVersion, wire, csp); return (result == null) ? "" : result; } @@ -614,8 +615,8 @@ public class ClientWiredStatelessTcpConnectionHub { private long proxySend(@NotNull final EventId methodName, final long startTime, - short channelID, - @NotNull final Wire wire) { + @NotNull final Wire wire, + @NotNull final String csp) { assert outBytesLock().isHeldByCurrentThread(); assert !inBytesLock().isHeldByCurrentThread(); @@ -623,7 +624,7 @@ public class ClientWiredStatelessTcpConnectionHub { // send outBytesLock().lock(); try { - long tid = writeHeader(startTime, channelID, wire); + long tid = writeHeader(startTime, wire, csp); wire.writeDocument(false, wireOut -> { wireOut.writeEventName(methodName); wireOut.writeValue().marshallable(w -> { @@ -639,19 +640,19 @@ public class ClientWiredStatelessTcpConnectionHub { @SuppressWarnings("SameParameterValue") @Nullable - String proxyReturnString(@NotNull final EventId messageId, short channelID) { - return proxyReturnString(messageId, channelID, outWire); + String proxyReturnString(@NotNull final EventId messageId, String csp) { + return proxyReturnString(messageId, outWire, csp); } @SuppressWarnings("SameParameterValue") @Nullable - String proxyReturnString(@NotNull final EventId eventId, short channelID, Wire outWire) { + String proxyReturnString(@NotNull final EventId eventId, Wire outWire, String csp) { final long startTime = System.currentTimeMillis(); long tid; outBytesLock().lock(); try { - tid = proxySend(eventId, startTime, channelID, outWire); + tid = proxySend(eventId, startTime, outWire, csp); } finally { outBytesLock().unlock(); } @@ -682,7 +683,7 @@ public class ClientWiredStatelessTcpConnectionHub { return outWire; } - long writeHeader(long startTime, short channelID, Wire wire) { + long writeHeader(long startTime, Wire wire, String csp) { assert outBytesLock().isHeldByCurrentThread(); markSize(wire); @@ -690,10 +691,8 @@ public class ClientWiredStatelessTcpConnectionHub { long tid = nextUniqueTransaction(startTime); wire.writeDocument(true, wireOut -> { - wireOut.write(csp).text("MAP"); + wireOut.write(MapWireHandlerBuilder.Fields.csp).text(csp); wireOut.write(MapWireHandlerBuilder.Fields.tid).int64(tid); - // wireOut.write(timeStamp).int64(startTime); - wireOut.write(channelId).int16(channelID); }); 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 eda2e89aa..4a8b399a7 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -196,15 +196,27 @@ class MapWireHandler implements WireHandler, Consumer { } } + StringBuilder cspText = new StringBuilder(); @SuppressWarnings("UnusedReturnValue") void onEvent() throws StreamCorruptedException { // it is assumed by this point that the buffer has all the bytes in it for this message + + inWire.read(csp).text(cspText); + + final int i = cspText.lastIndexOf("/"); + + if (i != -1 && i < (cspText.length() - 1)) { + final String channelStr = cspText.substring(i + 1, cspText.length() - "#MAP".length()); + channelId = Short.parseShort(channelStr); + } else + channelId = 0; + + final long tid = inWire.read(Fields.tid).int64(); - channelId = inWire.read(Fields.channelId).int16(); final Bytes bytes1 = inWire.bytes();