From 08d01fce6f5d12bd917bb177ddb7d7e8245114d1 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 10 Mar 2015 12:50:45 +0000 Subject: [PATCH 1/3] moved wire testing into chronicle.map --- .../map/StatelessWiredConnector.java | 17 +--- ...redChronicleMapStatelessClientBuilder.java | 12 ++- .../map/WiredStatelessChronicleMap.java | 21 ++--- .../chronicle/{wire => map}/TestWire.java | 85 +++++++++++++++++-- 4 files changed, 99 insertions(+), 36 deletions(-) rename src/test/java/net/openhft/chronicle/{wire => map}/TestWire.java (85%) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index d27045ff9..4b1b097a1 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -64,12 +64,13 @@ class StatelessWiredConnector bytesChronicleMaps = new ArrayList<>(); private final TextWire inWire = new TextWire(Bytes.elasticByteBuffer()); private final TextWire outWire = new TextWire(Bytes.elasticByteBuffer()); - //private final ByteBuffer resultBuffer = ByteBuffer.allocateDirect(64); + @NotNull private final ByteBuffer inLanByteBuffer = ByteBuffer.allocateDirect(64); private final StringBuilder methodName = new StringBuilder(); @@ -80,7 +81,6 @@ class StatelessWiredConnector channelList; private ReplicationHub hub; @@ -108,8 +108,6 @@ class StatelessWiredConnector channelList) { - if (channelList == null) - throw new IllegalStateException("The StatelessWiredConnector currently only support maps " + - "that are set up via the replication channel. localIdentifier=" + localIdentifier); - this.channelList = channelList; } @@ -346,7 +340,6 @@ class StatelessWiredConnector "TRANSACTION_ID").int64(transactionId); @@ -445,7 +438,6 @@ class StatelessWiredConnector outWire.write(() -> "RESULT").text(b.toString())); - return; } @@ -468,11 +460,6 @@ class StatelessWiredConnector implements this.channelID = channelID; } + public WiredChronicleMapStatelessClientBuilder(WiredStatelessClientTcpConnectionHub hub, Class keyClass, Class valueClass, short channelID) { + this.keyClass = keyClass; + this.valueClass = valueClass; + + this.channelID = channelID; + } + // This method is meaningful, despite you could call of().create(), because Java 7 doesn't // infer type parameters if case of chained calls, like this; and then you couldn't omit class // name. Then it is @@ -56,8 +63,7 @@ public final class WiredChronicleMapStatelessClientBuilder implements // -- the second is really shorter. - - private final InetSocketAddress remoteAddress; + private InetSocketAddress remoteAddress; private boolean putReturnsNull = false; private boolean removeReturnsNull = false; private long timeoutMs = TimeUnit.SECONDS.toMillis(10); @@ -67,7 +73,6 @@ public final class WiredChronicleMapStatelessClientBuilder implements private final AtomicBoolean used = new AtomicBoolean(false); - InetSocketAddress remoteAddress() { return remoteAddress; } @@ -149,7 +154,6 @@ public final class WiredChronicleMapStatelessClientBuilder implements } - public WiredStatelessClientTcpConnectionHub hub() { return hub; } diff --git a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java index ca26ef322..419093d80 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -44,7 +44,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class); - // todo replace with an ENUM public static final String METHOD_NAME = "METHOD_NAME"; public static final String RESULT_IS_NULL = "RESULT_IS_NULL"; @@ -53,7 +52,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo public static final String ARG_1 = "ARG_1"; public static final String ARG_2 = "ARG_2"; - private final WiredStatelessClientTcpConnectionHub hub; protected Class kClass; @@ -270,7 +268,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); } - @NotNull public void createChannel(short channelID) { proxyReturnVoid(() -> { writeField(METHOD_NAME, "CREATE_CHANNEL"); @@ -278,7 +275,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo }); } - public void putAll(@NotNull Map map) { final long startTime = System.currentTimeMillis(); @@ -377,9 +373,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } else break; - - // return result; - } finally { hub.inBytesLock().unlock(); } @@ -479,7 +472,17 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo assert hub.outBytesLock().isHeldByCurrentThread(); assert !hub.inBytesLock().isHeldByCurrentThread(); - if (value instanceof CharSequence) { + if (value instanceof Byte) + wire.write(() -> fieldName).int8((Byte) value); + else if (value instanceof Character) + wire.write(() -> fieldName).text(((Character) value).toString()); + else if (value instanceof Short) + wire.write(() -> fieldName).int16((Short) value); + else if (value instanceof Integer) + wire.write(() -> fieldName).int32((Integer) value); + else if (value instanceof Long) + wire.write(() -> fieldName).int64((Long) value); + else if (value instanceof CharSequence) { wire.write(() -> fieldName).text((CharSequence) value); } else if (value instanceof Marshallable) { wire.write(() -> fieldName).marshallable((Marshallable) value); @@ -722,8 +725,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } - - @Nullable private R proxyReturnObject(@NotNull final String methodName, K key, V value, Class resultType) { diff --git a/src/test/java/net/openhft/chronicle/wire/TestWire.java b/src/test/java/net/openhft/chronicle/map/TestWire.java similarity index 85% rename from src/test/java/net/openhft/chronicle/wire/TestWire.java rename to src/test/java/net/openhft/chronicle/map/TestWire.java index 3861be50d..b53305194 100644 --- a/src/test/java/net/openhft/chronicle/wire/TestWire.java +++ b/src/test/java/net/openhft/chronicle/map/TestWire.java @@ -16,14 +16,15 @@ * limitations under the License. */ -package net.openhft.chronicle.wire; +package net.openhft.chronicle.map; import net.openhft.chronicle.hash.replication.ReplicationHub; import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig; -import net.openhft.chronicle.map.ChronicleMap; -import net.openhft.chronicle.map.WiredChronicleMapStatelessClientBuilder; -import net.openhft.chronicle.map.WiredStatelessChronicleMap; +import net.openhft.chronicle.wire.Marshallable; +import net.openhft.chronicle.wire.WireIn; +import net.openhft.chronicle.wire.WireOut; +import org.junit.Assert; import org.junit.Test; import java.io.IOException; @@ -39,7 +40,6 @@ import static org.junit.Assert.*; */ public class TestWire { - public static int SERVER_PORT = 8086; private ChronicleMap map1a; private ChronicleMap map2a; @@ -47,6 +47,66 @@ public class TestWire { private ReplicationHub hubA; + @Test + public void testUsingDetails() throws IOException, InterruptedException { + + { + final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig + .of(++SERVER_PORT) + .heartBeatInterval(1, SECONDS); + + hubA = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig) + .createWithId((byte) 1); + + // this is how you add maps after the custer is created + map1a = of(byte[].class, byte[].class) + .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); + + + byte identifier = (byte) 2; + + + final WiredChronicleMapStatelessClientBuilder builder = + new WiredChronicleMapStatelessClientBuilder<>( + new InetSocketAddress("localhost", SERVER_PORT), + String.class, + Details.class, + (short) 1); + + builder.identifier(identifier); + final ChronicleMap detailsMap = builder.create(); + + + Details details = detailsMap.get("Colours"); + + // todo add some sort of cross node locking + if (details == null) { + short channelID = (short) 2; + ((WiredStatelessChronicleMap) detailsMap).createChannel(channelID); + details = new Details(String.class, String.class, channelID); + detailsMap.put("Colours", details); + } + + + ChronicleMap colours = new WiredChronicleMapStatelessClientBuilder( + builder.hub(), + details.keyClass, + details.valueClass, + details.channelID).hub(builder.hub()).create(); + + colours.put("Rob", "Blue"); + + assertEquals(1, colours.size()); + + + Details result = detailsMap.get("Colours"); + Assert.assertEquals(2, result.channelID); + map1a.close(); + } + + } + + @Test public void testCreateChannel() throws IOException, InterruptedException { @@ -351,8 +411,8 @@ public class TestWire { Set> entries = statelessMap.entrySet(); assertEquals(2, values.size()); - assertTrue(entries.contains(Collections.singletonMap("Key1","Value1").entrySet().iterator().next())); - assertTrue(entries.contains(Collections.singletonMap("Key2","Value2").entrySet().iterator().next())); + assertTrue(entries.contains(Collections.singletonMap("Key1", "Value1").entrySet().iterator().next())); + assertTrue(entries.contains(Collections.singletonMap("Key2", "Value2").entrySet().iterator().next())); map1a.close(); @@ -420,6 +480,17 @@ public class TestWire { short channelID; + public Details() { + } + + + public Details(Class keyClass, Class valueClass, short channelID) { + this.keyClass = keyClass; + this.valueClass = valueClass; + this.channelID = channelID; + } + + @Override public void writeMarshallable(WireOut wire) { wire.write(() -> "keyClass").text(keyClass.getName()); From 61cccf1793e8b3b9784cf6a0778d18f0f2c8ca8b Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 10 Mar 2015 18:32:15 +0000 Subject: [PATCH 2/3] changed "string" to enum --- .../map/StatelessWiredConnector.java | 122 ++++++++++-------- 1 file changed, 71 insertions(+), 51 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index 4b1b097a1..6d80d6514 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -30,6 +30,7 @@ import net.openhft.chronicle.hash.replication.ReplicationHub; import net.openhft.chronicle.wire.TextWire; import net.openhft.chronicle.wire.ValueIn; import net.openhft.chronicle.wire.Wire; +import net.openhft.chronicle.wire.WireKey; import net.openhft.lang.io.ByteBufferBytes; import net.openhft.lang.io.DirectStore; import net.openhft.lang.io.IByteBufferBytes; @@ -60,6 +61,24 @@ import static java.nio.channels.SelectionKey.OP_WRITE; class StatelessWiredConnector { private static final Logger LOG = LoggerFactory.getLogger(StatelessWiredConnector.class); + + public enum Fields implements WireKey { + HAS_NEXT, + TIME_STAMP, + CHANNEL_ID, + METHOD_NAME, + TRANSACTION_ID, + RESULT, + RESULT_KEY, + RESULT_VALUE, + ARG_1, + ARG_2, + ARG_3, + IS_EXCEPTION, + EXCEPTION, + RESULT_IS_NULL + } + private final NativeBytes inLangBytes = new NativeBytes(0, 0); @NotNull @@ -88,7 +107,7 @@ class StatelessWiredConnector iterator) { - outWire.write(() -> "RESULT"); + outWire.write(Fields.RESULT); outWire.bytes().write(iterator.next()); } }; @@ -100,10 +119,10 @@ class StatelessWiredConnector entry = iterator.next(); - outWire.write(() -> "RESULT_KEY"); + outWire.write(Fields.RESULT_KEY); outWire.bytes().write(entry.getKey()); - outWire.write(() -> "RESULT_VALUE"); + outWire.write(Fields.RESULT_VALUE); outWire.bytes().write(entry.getValue()); } }; @@ -277,12 +296,12 @@ class StatelessWiredConnector { // each chunk has its own transaction-id - outWire.write(() -> "TRANSACTION_ID").int64(transactionId); + outWire.write(Fields.TRANSACTION_ID).int64(transactionId); write(map -> { boolean hasNext = iterator.hasNext(); - outWire.write(() -> "HAS_NEXT").bool(hasNext); + outWire.write(Fields.HAS_NEXT).bool(hasNext); if (hasNext) c.accept(iterator); @@ -301,14 +320,14 @@ class StatelessWiredConnector "TRANSACTION_ID").int64(); - timestamp = inWire.read(() -> "TIME_STAMP").int64(); - channelId = inWire.read(() -> "CHANNEL_ID").int16(); - inWire.read(() -> "METHOD_NAME").text(methodName); + long transactionId = inWire.read(Fields.TRANSACTION_ID).int64(); + timestamp = inWire.read(Fields.TIME_STAMP).int64(); + channelId = inWire.read(Fields.CHANNEL_ID).int16(); + inWire.read(Fields.METHOD_NAME).text(methodName); if ("PUT_WITHOUT_ACC".contentEquals(methodName)) { writeVoid(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2"); + final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); bytesMap.put(reader, reader, timestamp, identifier()); }); return; @@ -346,7 +365,7 @@ class StatelessWiredConnector { - short channelId1 = inWire.read(() -> "ARG_1").int16(); + short channelId1 = inWire.read(Fields.ARG_1).int16(); chronicleHashInstanceBuilder.replicatedViaChannel(hub.createChannel(channelId1)).create(); return null; }); @@ -355,47 +374,47 @@ class StatelessWiredConnector "RESULT").int8(); + this.remoteIdentifier = inWire.read(Fields.RESULT).int8(); return; } if ("LONG_SIZE".contentEquals(methodName)) { - write(b -> outWire.write(() -> "RESULT").int64(b.longSize())); + write(b -> outWire.write(Fields.RESULT).int64(b.longSize())); return; } if ("IS_EMPTY".contentEquals(methodName)) { - write(b -> outWire.write(() -> "RESULT").bool(b.isEmpty())); + write(b -> outWire.write(Fields.RESULT).bool(b.isEmpty())); return; } if ("CONTAINS_KEY".contentEquals(methodName)) { - write(b -> outWire.write(() -> "RESULT"). - bool(b.containsKey(toReader(inWire, "ARG_1")))); + write(b -> outWire.write(Fields.RESULT). + bool(b.containsKey(toReader(inWire, Fields.ARG_1)))); return; } if ("CONTAINS_VALUE".contentEquals(methodName)) { - write(b -> outWire.write(() -> "RESULT"). - bool(b.containsKey(toReader(inWire, "ARG_1")))); + write(b -> outWire.write(Fields.RESULT). + bool(b.containsKey(toReader(inWire, Fields.ARG_1)))); return; } if ("GET".contentEquals(methodName)) { - writeValue(b -> b.get(toReader(inWire, "ARG_1"))); + writeValue(b -> b.get(toReader(inWire, Fields.ARG_1))); return; } if ("PUT".contentEquals(methodName)) { writeValue(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2"); + final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); bytesMap.put(reader, reader, timestamp, identifier()); }); return; } if ("REMOVE".contentEquals(methodName)) { - writeValue(b -> b.remove(toReader(inWire, "ARG_1"))); + writeValue(b -> b.remove(toReader(inWire, Fields.ARG_1))); return; } @@ -406,7 +425,7 @@ class StatelessWiredConnector { - final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2"); + final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); bytesMap.replace(reader, reader); }); return; @@ -414,7 +433,7 @@ class StatelessWiredConnector { - final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2", "ARG_3"); + final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2, Fields.ARG_3); bytesMap.replace(reader, reader, reader); }); return; @@ -422,7 +441,7 @@ class StatelessWiredConnector { - final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2"); + final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); }); return; @@ -430,34 +449,33 @@ class StatelessWiredConnector { - final net.openhft.lang.io.Bytes reader = toReader(inWire, "ARG_1", "ARG_2"); + final net.openhft.lang.io.Bytes reader = toReader(inWire, Fields.ARG_1, Fields.ARG_2); bytesMap.remove(reader, reader, timestamp, identifier()); }); return; } if ("TO_STRING".contentEquals(methodName)) { - write(b -> outWire.write(() -> "RESULT").text(b.toString())); + write(b -> outWire.write(Fields.RESULT).text(b.toString())); return; } if ("APPLICATION_VERSION".contentEquals(methodName)) { - write(b -> outWire.write(() -> "RESULT").text(applicationVersion())); + write(b -> outWire.write(Fields.RESULT).text(applicationVersion())); return; } if ("PERSISTED_DATA_VERSION".contentEquals(methodName)) { - write(b -> outWire.write(() -> "RESULT").text(persistedDataVersion())); + write(b -> outWire.write(Fields.RESULT).text(persistedDataVersion())); return; } if ("HASH_CODE".contentEquals(methodName)) { - write(b -> outWire.write(() -> "RESULT").int32(b.hashCode())); + write(b -> outWire.write(Fields.RESULT).int32(b.hashCode())); return; } - throw new IllegalStateException("unsupported event=" + methodName); } finally { @@ -478,10 +496,12 @@ class StatelessWiredConnector fieldName); + System.out.println(Bytes.toDebugString(inWire.bytes())); + + ValueIn read = wire.read(fieldName); long l = read.readLength(); if (l > Integer.MAX_VALUE) @@ -527,18 +547,18 @@ class StatelessWiredConnector "HAS_NEXT").bool()) { - collectData.put(toBytes("ARG_1"), toBytes("ARG_2")); + if (inWire.read(Fields.HAS_NEXT).bool()) { + collectData.put(toBytes(Fields.ARG_1), toBytes(Fields.ARG_2)); } else { // the old code assumed that all the data would fit into a single buffer // this assumption is invalid if (!collectData.isEmpty()) { bytesMap.delegate.putAll((Map) collectData); incompleteWork.remove(transactionId); - outWire.write(() -> "TRANSACTION_ID").int64(transactionId); + outWire.write(Fields.TRANSACTION_ID).int64(transactionId); // todo handle the case where there is an exception - outWire.write(() -> "IS_EXCEPTION").bool(false); + outWire.write(Fields.IS_EXCEPTION).bool(false); nofityDataWritten(); } @@ -577,15 +597,15 @@ class StatelessWiredConnector field); + ValueIn read = wire.read(field); long fieldLength = read.readLength(); long endPos = wire.bytes().position() + fieldLength; @@ -626,7 +646,6 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(true); - outWire.write(() -> "EXCEPTION").text(toString(e)); + outWire.write(Fields.IS_EXCEPTION).bool(true); + outWire.write(Fields.EXCEPTION).text(toString(e)); LOG.error("", e); return; } @@ -712,7 +731,7 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(false); + outWire.write(Fields.IS_EXCEPTION).bool(false); try { c.accept(bytesMap); @@ -720,8 +739,8 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(true); - outWire.write(() -> "EXCEPTION").text(toString(e)); + outWire.write(Fields.IS_EXCEPTION).bool(true); + outWire.write(Fields.EXCEPTION).text(toString(e)); LOG.error("", e); return; } @@ -742,7 +761,7 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(false); + outWire.write(Fields.IS_EXCEPTION).bool(false); try { r.call(); @@ -750,8 +769,8 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(true); - outWire.write(() -> "EXCEPTION").text(toString(e)); + outWire.write(Fields.IS_EXCEPTION).bool(true); + outWire.write(Fields.EXCEPTION).text(toString(e)); LOG.error("", e); return; } @@ -812,6 +831,7 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(false); + wire.write(Fields.IS_EXCEPTION).bool(false); // flips calls flip on this message so that we can read it flipMessage(); @@ -883,10 +903,10 @@ class StatelessWiredConnector "RESULT_IS_NULL").bool(isNull); + wire.write(Fields.RESULT_IS_NULL).bool(isNull); if (!isNull) { // write the result - wire.write(() -> "RESULT"); + wire.write(Fields.RESULT); wire.bytes().write(outChronBytes); } } From d4be5c7a83d9f7e41861b5867dbd039ecb02e5ab Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 10 Mar 2015 19:55:40 +0000 Subject: [PATCH 3/3] updated to use enum --- .../map/StatelessWiredConnector.java | 2 +- .../map/WiredStatelessChronicleMap.java | 129 +++++++++--------- .../net/openhft/chronicle/map/TestWire.java | 6 - 3 files changed, 62 insertions(+), 75 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index 6d80d6514..9d9590b24 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java @@ -62,7 +62,7 @@ class StatelessWiredConnector implements ChronicleMap, Clo private static final Logger LOG = LoggerFactory.getLogger(WiredStatelessChronicleMap.class); - // todo replace with an ENUM - public static final String METHOD_NAME = "METHOD_NAME"; - public static final String RESULT_IS_NULL = "RESULT_IS_NULL"; - public static final String HAS_NEXT = "HAS_NEXT"; - public static final String RESULT = "RESULT"; - public static final String ARG_1 = "ARG_1"; - public static final String ARG_2 = "ARG_2"; - private final WiredStatelessClientTcpConnectionHub hub; protected Class kClass; @@ -269,10 +261,11 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } public void createChannel(short channelID) { - proxyReturnVoid(() -> { - writeField(METHOD_NAME, "CREATE_CHANNEL"); - hub.outWire().write(() -> "ARG_1").int16(channelID); - }); + proxyReturnVoid( + () -> { + writeField(StatelessWiredConnector.Fields.METHOD_NAME, "CREATE_CHANNEL"); + hub.outWire().write(StatelessWiredConnector.Fields.ARG_1).int16(channelID); + }); } public void putAll(@NotNull Map map) { @@ -301,18 +294,18 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.markSize(wire); hub.startTime(startTime); - wire.write(() -> "TRANSACTION_ID").int64(transactionId); - wire.write(() -> "TIME_STAMP").int64(startTime); - wire.write(() -> "CHANNEL_ID").int16(channelID); + wire.write(StatelessWiredConnector.Fields.TRANSACTION_ID).int64(transactionId); + wire.write(StatelessWiredConnector.Fields.TIME_STAMP).int64(startTime); + wire.write(StatelessWiredConnector.Fields.CHANNEL_ID).int16(channelID); - writeField(METHOD_NAME, "PUT_ALL"); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, "PUT_ALL"); - hub.outWire().write(() -> "HAS_NEXT").bool(hasNext); + hub.outWire().write(StatelessWiredConnector.Fields.HAS_NEXT).bool(hasNext); if (hasNext) { Map.Entry e = iterator.next(); - writeField(ARG_1, e.getKey()); - writeField(ARG_2, e.getValue()); + writeField(StatelessWiredConnector.Fields.ARG_1, e.getKey()); + writeField(StatelessWiredConnector.Fields.ARG_2, e.getValue()); } hub.writeSocket(hub.outWire()); @@ -347,7 +340,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } public void clear() { - proxyReturnVoid(() -> writeField(METHOD_NAME, CLEAR.toString())); + proxyReturnVoid(() -> writeField(StatelessWiredConnector.Fields.METHOD_NAME, CLEAR.toString())); } @@ -367,8 +360,8 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(() -> HAS_NEXT).bool()) { - K k = readK(RESULT, wireIn, null); + if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) { + K k = readK(StatelessWiredConnector.Fields.RESULT, wireIn, null); result.add(k); } else break; @@ -397,8 +390,8 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(() -> HAS_NEXT).bool()) { - V v = readV(RESULT, wireIn, null); + if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) { + V v = readV(StatelessWiredConnector.Fields.RESULT, wireIn, null); result.add(v); } else break; @@ -427,10 +420,10 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(() -> HAS_NEXT).bool()) { + if (wireIn.read(StatelessWiredConnector.Fields.HAS_NEXT).bool()) { result.put( - readK("RESULT_KEY", wireIn, null), - readV("RESULT_VALUE", wireIn, null)); + readK(StatelessWiredConnector.Fields.RESULT_KEY, wireIn, null), + readV(StatelessWiredConnector.Fields.RESULT_VALUE, wireIn, null)); } else break; @@ -456,43 +449,43 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo // receive hub.inBytesLock().lock(); try { - return hub.proxyReply(timeoutTime, transactionId).read(() -> RESULT).int64(); + return hub.proxyReply(timeoutTime, transactionId).read(StatelessWiredConnector.Fields.RESULT).int64(); } finally { hub.inBytesLock().unlock(); } } - private void writeField(String fieldName, Object value) { + private void writeField(StatelessWiredConnector.Fields fieldName, Object value) { writeField(fieldName, value, hub.outWire()); } - private void writeField(String fieldName, Object value, Wire wire) { + private void writeField(StatelessWiredConnector.Fields fieldName, Object value, Wire wire) { assert hub.outBytesLock().isHeldByCurrentThread(); assert !hub.inBytesLock().isHeldByCurrentThread(); if (value instanceof Byte) - wire.write(() -> fieldName).int8((Byte) value); + wire.write(fieldName).int8((Byte) value); else if (value instanceof Character) - wire.write(() -> fieldName).text(((Character) value).toString()); + wire.write(fieldName).text(((Character) value).toString()); else if (value instanceof Short) - wire.write(() -> fieldName).int16((Short) value); + wire.write(fieldName).int16((Short) value); else if (value instanceof Integer) - wire.write(() -> fieldName).int32((Integer) value); + wire.write(fieldName).int32((Integer) value); else if (value instanceof Long) - wire.write(() -> fieldName).int64((Long) value); + wire.write(fieldName).int64((Long) value); else if (value instanceof CharSequence) { - wire.write(() -> fieldName).text((CharSequence) value); + wire.write(fieldName).text((CharSequence) value); } else if (value instanceof Marshallable) { - wire.write(() -> fieldName).marshallable((Marshallable) value); + wire.write(fieldName).marshallable((Marshallable) value); } else { throw new IllegalStateException("type=" + value.getClass() + " is unsupported, it must either be of type Marshallable or CharSequence"); } } - private V readValue(String argName, long transactionId, long startTime, final V usingValue) { + private V readValue(StatelessWiredConnector.Fields argName, long transactionId, long startTime, final V usingValue) { assert !hub.outBytesLock().isHeldByCurrentThread(); long timeoutTime = startTime + hub.timeoutMs; @@ -501,7 +494,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(() -> RESULT_IS_NULL).bool()) + if (wireIn.read(StatelessWiredConnector.Fields.RESULT_IS_NULL).bool()) return null; return readV(argName, wireIn, usingValue); @@ -511,54 +504,54 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } } - private V readV(String argName, Wire wireIn, V usingValue) { + private V readV(StatelessWiredConnector.Fields argName, Wire wireIn, V usingValue) { if (StringBuilder.class.isAssignableFrom(vClass)) { - wireIn.read(() -> argName).text((StringBuilder) usingValue); + wireIn.read(argName).text((StringBuilder) usingValue); return usingValue; } else if (Marshallable.class.isAssignableFrom(vClass)) { if (usingValue == null) try { V v = vClass.newInstance(); - wireIn.read(() -> argName).marshallable((Marshallable) v); + wireIn.read(argName).marshallable((Marshallable) v); return v; } catch (Exception e) { throw new RuntimeException(e); } - wireIn.read(() -> argName).marshallable((Marshallable) usingValue); + wireIn.read(argName).marshallable((Marshallable) usingValue); return usingValue; } else if (String.class.isAssignableFrom(vClass)) { //noinspection unchecked - return (V) wireIn.read(() -> argName).text(); + return (V) wireIn.read(argName).text(); } else { throw new IllegalStateException("unsupported type"); } } - private K readK(String argName, Wire wireIn, K usingValue) { + private K readK(StatelessWiredConnector.Fields argName, Wire wireIn, K usingValue) { if (StringBuilder.class.isAssignableFrom(kClass)) { - wireIn.read(() -> argName).text((StringBuilder) usingValue); + wireIn.read(argName).text((StringBuilder) usingValue); return usingValue; } else if (Marshallable.class.isAssignableFrom(kClass)) { if (usingValue == null) try { K v = kClass.newInstance(); - wireIn.read(() -> argName).marshallable((Marshallable) v); + wireIn.read(argName).marshallable((Marshallable) v); return v; } catch (Exception e) { throw new RuntimeException(e); } - wireIn.read(() -> argName).marshallable((Marshallable) usingValue); + wireIn.read(argName).marshallable((Marshallable) usingValue); return usingValue; } else if (String.class.isAssignableFrom(vClass)) { //noinspection unchecked - return (K) wireIn.read(() -> argName).text(); + return (K) wireIn.read(argName).text(); } else { throw new IllegalStateException("unsupported type"); @@ -574,7 +567,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - return wireIn.read(() -> RESULT).bool(); + return wireIn.read(StatelessWiredConnector.Fields.RESULT).bool(); } finally { hub.inBytesLock().unlock(); } @@ -589,7 +582,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - return wireIn.read(() -> RESULT).int32(); + return wireIn.read(StatelessWiredConnector.Fields.RESULT).int32(); } finally { hub.inBytesLock().unlock(); } @@ -604,9 +597,9 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); - writeField(ARG_1, key); - writeField(ARG_2, value); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName); + writeField(StatelessWiredConnector.Fields.ARG_1, key); + writeField(StatelessWiredConnector.Fields.ARG_2, value); hub.writeSocket(hub.outWire()); } finally { @@ -625,9 +618,9 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); - writeField(ARG_1, value1); - writeField(ARG_2, value2); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName); + writeField(StatelessWiredConnector.Fields.ARG_1, value1); + writeField(StatelessWiredConnector.Fields.ARG_2, value2); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -659,7 +652,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -677,7 +670,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -715,7 +708,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -737,9 +730,9 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo assert !hub.inBytesLock().isHeldByCurrentThread(); transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); - writeField(ARG_1, key); - writeField(ARG_2, value); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName); + writeField(StatelessWiredConnector.Fields.ARG_1, key); + writeField(StatelessWiredConnector.Fields.ARG_2, value); hub.writeSocket(hub.outWire()); } finally { @@ -750,7 +743,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo return null; if (resultType == vClass) - return (R) readValue(RESULT, transactionId, startTime, null); + return (R) readValue(StatelessWiredConnector.Fields.RESULT, transactionId, startTime, null); else throw new UnsupportedOperationException("class of type class=" + resultType + " is not " + @@ -769,7 +762,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo return null; if (rClass == vClass) - return (R) readValue(RESULT, transactionId, startTime, null); + return (R) readValue(StatelessWiredConnector.Fields.RESULT, transactionId, startTime, null); else throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + "supported"); @@ -780,8 +773,8 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); - writeField(ARG_1, arg1); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName); + writeField(StatelessWiredConnector.Fields.ARG_1, arg1); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -795,7 +788,7 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(METHOD_NAME, methodName); + writeField(StatelessWiredConnector.Fields.METHOD_NAME, methodName); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); diff --git a/src/test/java/net/openhft/chronicle/map/TestWire.java b/src/test/java/net/openhft/chronicle/map/TestWire.java index b53305194..4e86856be 100644 --- a/src/test/java/net/openhft/chronicle/map/TestWire.java +++ b/src/test/java/net/openhft/chronicle/map/TestWire.java @@ -62,10 +62,8 @@ public class TestWire { map1a = of(byte[].class, byte[].class) .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); - byte identifier = (byte) 2; - final WiredChronicleMapStatelessClientBuilder builder = new WiredChronicleMapStatelessClientBuilder<>( new InetSocketAddress("localhost", SERVER_PORT), @@ -76,7 +74,6 @@ public class TestWire { builder.identifier(identifier); final ChronicleMap detailsMap = builder.create(); - Details details = detailsMap.get("Colours"); // todo add some sort of cross node locking @@ -87,7 +84,6 @@ public class TestWire { detailsMap.put("Colours", details); } - ChronicleMap colours = new WiredChronicleMapStatelessClientBuilder( builder.hub(), details.keyClass, @@ -121,8 +117,6 @@ public class TestWire { // this is how you add maps after the custer is created map1a = of(byte[].class, byte[].class) .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); - - byte identifier = (byte) 2; final WiredChronicleMapStatelessClientBuilder builder =