diff --git a/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java b/src/main/java/net/openhft/chronicle/map/StatelessWiredConnector.java index d27045ff9..9d9590b24 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,16 +61,35 @@ import static java.nio.channels.SelectionKey.OP_WRITE; class StatelessWiredConnector { private static final Logger LOG = LoggerFactory.getLogger(StatelessWiredConnector.class); + + static 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 private final OutMessageAdapter outMessageAdapter = new OutMessageAdapter(); + @NotNull private final ArrayList 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 +100,6 @@ class StatelessWiredConnector channelList; private ReplicationHub hub; @@ -88,7 +107,7 @@ class StatelessWiredConnector iterator) { - outWire.write(() -> "RESULT"); + outWire.write(Fields.RESULT); outWire.bytes().write(iterator.next()); } }; @@ -100,16 +119,14 @@ 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()); } }; - private Runnable in = null; - private byte remoteIdentifier; private byte localIdentifer; private Runnable out = null; @@ -123,10 +140,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; } @@ -283,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); @@ -307,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,14 +359,13 @@ class StatelessWiredConnector "TRANSACTION_ID").int64(transactionId); if ("CREATE_CHANNEL".contentEquals(methodName)) { writeVoid(() -> { - short channelId1 = inWire.read(() -> "ARG_1").int16(); + short channelId1 = inWire.read(Fields.ARG_1).int16(); chronicleHashInstanceBuilder.replicatedViaChannel(hub.createChannel(channelId1)).create(); return null; }); @@ -362,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; } @@ -413,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; @@ -421,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; @@ -429,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; @@ -437,42 +449,35 @@ 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); - // else if ("MAP_FOR_KEY".contentEquals(methodName) - // else if ("PUT_MAPPED".contentEquals(methodName)) - // else if ("VALUE_BUILDER".contentEquals(methodName)) - - } finally { @@ -491,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) @@ -540,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(); } @@ -590,15 +597,15 @@ class StatelessWiredConnector field); + ValueIn read = wire.read(field); long fieldLength = read.readLength(); long endPos = wire.bytes().position() + fieldLength; @@ -639,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; } @@ -725,7 +731,7 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(false); + outWire.write(Fields.IS_EXCEPTION).bool(false); try { c.accept(bytesMap); @@ -733,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; } @@ -755,7 +761,7 @@ class StatelessWiredConnector "IS_EXCEPTION").bool(false); + outWire.write(Fields.IS_EXCEPTION).bool(false); try { r.call(); @@ -763,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; } @@ -825,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(); @@ -896,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); } } diff --git a/src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java b/src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java index 1b9f4db51..68f17ec8e 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/WiredChronicleMapStatelessClientBuilder.java @@ -47,6 +47,13 @@ public final class WiredChronicleMapStatelessClientBuilder 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..1203bda14 100644 --- a/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/WiredStatelessChronicleMap.java @@ -44,16 +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"; - 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; @@ -270,15 +260,14 @@ 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"); - 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) { final long startTime = System.currentTimeMillis(); @@ -305,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()); @@ -351,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())); } @@ -371,15 +360,12 @@ 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; - - // return result; - } finally { hub.inBytesLock().unlock(); } @@ -404,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; @@ -434,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; @@ -463,33 +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 CharSequence) { - wire.write(() -> fieldName).text((CharSequence) value); + 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); + 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; @@ -498,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); @@ -508,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"); @@ -571,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(); } @@ -586,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(); } @@ -601,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 { @@ -622,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(); @@ -656,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(); @@ -674,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(); @@ -712,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(); @@ -722,8 +718,6 @@ public class WiredStatelessChronicleMap implements ChronicleMap, Clo } - - @Nullable private R proxyReturnObject(@NotNull final String methodName, K key, V value, Class resultType) { @@ -736,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 { @@ -749,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 " + @@ -768,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"); @@ -779,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(); @@ -794,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/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..4e86856be 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,62 @@ 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 { @@ -61,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 = @@ -351,8 +405,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 +474,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());