diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index d89882dba..fc78a9607 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -34,7 +34,8 @@ import java.util.function.Consumer; import java.util.function.Predicate; import static java.util.Collections.emptyList; -import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId.*; +import static net.openhft.chronicle.map.MapWireHandler.EventId; +import static net.openhft.chronicle.map.MapWireHandler.EventId.*; /** @@ -135,7 +136,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon if (key == null || value == null) throw new NullPointerException(); - return proxyReturnObject(PUT_IF_ABSENT.toString(), key, value, vClass); + return proxyReturnObject(PUT_IF_ABSENT, key, value, vClass); } @SuppressWarnings("NullableProblems") @@ -159,7 +160,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon public V replace(K key, V value) { if (key == null || value == null) throw new NullPointerException(); - return proxyReturnObject(REPLACE.toString(), key, value, vClass); + return proxyReturnObject(REPLACE, key, value, vClass); } public int size() { @@ -204,7 +205,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon @NotNull public String serverPersistedDataVersion() { - return hub.proxyReturnString(PERSISTED_DATA_VERSION.toString(), channelID); + return hub.proxyReturnString(PERSISTED_DATA_VERSION, channelID); } public boolean isEmpty() { @@ -212,7 +213,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } public boolean containsKey(Object key) { - return proxyReturnBooleanK(CONTAINS_KEY.toString(), (K) key); + return proxyReturnBooleanK(CONTAINS_KEY, (K) key); } @NotNull @@ -221,7 +222,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } public boolean containsValue(Object value) { - return proxyReturnBooleanV(CONTAINS_VALUE.toString(), (V) value); + return proxyReturnBooleanV(CONTAINS_VALUE, (V) value); } public long longSize() { @@ -234,7 +235,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } public V get(Object key) { - return proxyReturnObject(vClass, GET.toString(), (K) key); + return proxyReturnObject(vClass, GET, (K) key); } @Nullable @@ -257,15 +258,15 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon public V remove(Object key) { if (key == null) throw keyNotNullNPE(); - return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC.toString() : REMOVE.toString(), (K) key); + return proxyReturnObject(vClass, removeReturnsNull ? REMOVE_WITHOUT_ACC : REMOVE, (K) key); } @Override public void createChannel(short channelID) { proxyReturnVoid( () -> { - writeField(Fields.METHOD_NAME, "CREATE_CHANNEL"); - hub.outWire().write(Fields.ARG_1).int16(channelID); + hub.outWire().write(Fields.methodName).text(CREATE_CHANNEL.toString()); + hub.outWire().write(Fields.arg1).int16(channelID); }); } @@ -290,22 +291,20 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon assert hub.outBytesLock().isHeldByCurrentThread(); hub.markSize(wire); hub.startTime(startTime); - wire.write(Fields.TYPE).text("MAP"); - wire.write(Fields.TRANSACTION_ID).int64(transactionId); - wire.write(Fields.TIME_STAMP).int64(startTime); - wire.write(Fields.CHANNEL_ID).int16(channelID); - writeField(Fields.METHOD_NAME, "PUT_ALL"); - - + wire.write(Fields.type).text("MAP"); + wire.write(Fields.transactionId).int64(transactionId); + wire.write(Fields.timeStamp).int64(startTime); + wire.write(Fields.channelId).int16(channelID); + hub.outWire().write(Fields.methodName).text(PUT_ALL.toString()); while (iterator.hasNext()) { Map.Entry e = iterator.next(); - hub.outWire().write(Fields.HAS_NEXT).bool(iterator.hasNext()); - writeField(Fields.ARG_1, e.getKey()); - writeField(Fields.ARG_2, e.getValue()); + hub.outWire().write(Fields.hasNext).bool(iterator.hasNext()); + writeField(Fields.arg1, e.getKey()); + writeField(Fields.arg2, e.getValue()); if (wire.bytes().remaining() < wire.bytes().capacity() * 0.5) break OUTER; @@ -331,7 +330,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon public V put(K key, V value) { if (key == null || value == null) throw new NullPointerException(); - return proxyReturnObject(putReturnsNull ? PUT_WITHOUT_ACC.toString() : PUT.toString(), key, value, vClass); + return proxyReturnObject(putReturnsNull ? PUT_WITHOUT_ACC : PUT, key, value, vClass); } @Nullable @@ -346,14 +345,14 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } public void clear() { - proxyReturnVoid(() -> writeField(Fields.METHOD_NAME, CLEAR.toString())); + proxyReturnVoid(() -> writeField(Fields.methodName, CLEAR.toString())); } @NotNull public Set keySet() { final Set usingCollection = new HashSet(); - readChunked(kClass, "KEY_SET", usingCollection); + readChunked(kClass, KEY_SET, usingCollection); return usingCollection; } @@ -361,10 +360,10 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon @NotNull public Collection values() { final List usingCollection = new ArrayList<>(); - return readChunked(vClass, "VALUES", usingCollection); + return readChunked(vClass, VALUES, usingCollection); } - private > Collection readChunked(Class vClass1, String values, A usingCollection) { + private > Collection readChunked(Class vClass1, EventId values, A usingCollection) { final long startTime = System.currentTimeMillis(); // send @@ -381,14 +380,14 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); while (wireIn.bytes().remaining() > 0) { - boolean bool = wireIn.read(Fields.HAS_NEXT).bool(); + boolean bool = wireIn.read(Fields.hasNext).bool(); if (bool) { - usingCollection.add(readObject(Fields.RESULT, wireIn, null, vClass1)); + usingCollection.add(readObject(Fields.result, wireIn, null, vClass1)); } else break OUTER; // todo process the exception - boolean isException = wireIn.read(Fields.IS_EXCEPTION).bool(); + boolean isException = wireIn.read(Fields.isException).bool(); } } finally { @@ -407,7 +406,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon final Map result = new HashMap(); final long startTime = System.currentTimeMillis(); // send - final long transactionId = proxySend("ENTRY_SET", startTime); + final long transactionId = proxySend(ENTRY_SET, startTime); assert !hub.outBytesLock().isHeldByCurrentThread(); final long timeoutTime = startTime + hub.timeoutMs; @@ -420,18 +419,18 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); while (wireIn.bytes().remaining() > 0) { - boolean bool = wireIn.read(Fields.HAS_NEXT).bool(); + boolean bool = wireIn.read(Fields.hasNext).bool(); if (bool) { result.put( - readK(Fields.RESULT_KEY, wireIn, null), - readV(Fields.RESULT_VALUE, wireIn, null)); + readK(Fields.resultKey, wireIn, null), + readV(Fields.resultValue, wireIn, null)); } else break OUTER; if (wireIn.bytes().remaining() > 0) { // todo process the exception - boolean isException = wireIn.read(Fields.IS_EXCEPTION).bool(); + boolean isException = wireIn.read(Fields.isException).bool(); } } @@ -457,7 +456,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon // receive hub.inBytesLock().lock(); try { - return hub.proxyReply(timeoutTime, transactionId).read(Fields.RESULT).int64(); + return hub.proxyReply(timeoutTime, transactionId).read(Fields.result).int64(); } finally { hub.inBytesLock().unlock(); } @@ -503,7 +502,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - if (wireIn.read(Fields.RESULT_IS_NULL).bool()) + if (wireIn.read(Fields.resultIsNull).bool()) return null; return readV(argName, wireIn, usingValue); @@ -589,7 +588,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - return wireIn.read(Fields.RESULT).bool(); + return wireIn.read(Fields.result).bool(); } finally { hub.inBytesLock().unlock(); } @@ -604,7 +603,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon hub.inBytesLock().lock(); try { final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); - return wireIn.read(Fields.RESULT).int32(); + return wireIn.read(Fields.result).int32(); } finally { hub.inBytesLock().unlock(); } @@ -619,9 +618,9 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); - writeField(Fields.ARG_1, key); - writeField(Fields.ARG_2, value); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + writeField(Fields.arg1, key); + writeField(Fields.arg2, value); hub.writeSocket(hub.outWire()); } finally { @@ -640,10 +639,10 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); - writeField(Fields.ARG_1, key); - writeField(Fields.ARG_2, value1); - writeField(Fields.ARG_3, value2); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + writeField(Fields.arg1, key); + writeField(Fields.arg2, value1); + writeField(Fields.arg3, value2); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -654,13 +653,13 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } @SuppressWarnings("SameParameterValue") - private boolean proxyReturnBooleanV(@NotNull final String methodName, V value) { + private boolean proxyReturnBooleanV(@NotNull final EventId methodName, V value) { final long startTime = System.currentTimeMillis(); return readBoolean(proxySend(methodName, startTime, value), startTime); } @SuppressWarnings("SameParameterValue") - private boolean proxyReturnBooleanK(@NotNull final String methodName, K key) { + private boolean proxyReturnBooleanK(@NotNull final EventId methodName, K key) { final long startTime = System.currentTimeMillis(); return readBoolean(proxySend(methodName, startTime, key), startTime); @@ -675,7 +674,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); + hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -693,7 +692,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); + hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -731,7 +730,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); + hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -742,7 +741,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon @Nullable - private R proxyReturnObject(@NotNull final String methodName, K key, V value, Class resultType) { + private R proxyReturnObject(@NotNull final EventId methodName, K key, V value, Class resultType) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -753,9 +752,9 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon assert !hub.inBytesLock().isHeldByCurrentThread(); transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); - writeField(Fields.ARG_1, key); - writeField(Fields.ARG_2, value); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + writeField(Fields.arg1, key); + writeField(Fields.arg2, value); hub.writeSocket(hub.outWire()); } finally { @@ -766,7 +765,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon return null; if (resultType == vClass) - return (R) readValue(Fields.RESULT, transactionId, startTime, null); + return (R) readValue(Fields.result, transactionId, startTime, null); else throw new UnsupportedOperationException("class of type class=" + resultType + " is not " + @@ -774,7 +773,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } @Nullable - private R proxyReturnObject(Class rClass, @NotNull final String methodName, Object key) { + private R proxyReturnObject(Class rClass, @NotNull final EventId methodName, Object key) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -785,19 +784,19 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon return null; if (rClass == vClass) - return (R) readValue(Fields.RESULT, transactionId, startTime, null); + return (R) readValue(Fields.result, transactionId, startTime, null); else throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + "supported"); } - private long proxySend(String methodName, long startTime, Object arg1) { + private long proxySend(EventId methodName, long startTime, Object arg1) { long transactionId; hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); - writeField(Fields.ARG_1, arg1); + hub.outWire().write(Fields.methodName).text(methodName.toString()); + writeField(Fields.arg1, arg1); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -806,12 +805,12 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } - private long proxySend(String methodName, long startTime) { + private long proxySend(EventId methodName, long startTime) { long transactionId; hub.outBytesLock().lock(); try { transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); - writeField(Fields.METHOD_NAME, methodName); + hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); @@ -819,12 +818,12 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon return transactionId; } - private boolean eventReturnsNull(@NotNull String methodName) { + private boolean eventReturnsNull(@NotNull EventId methodName) { switch (methodName) { - case "PUT_ALL_WITHOUT_ACC": - case "PUT_WITHOUT_ACC": - case "REMOVE_WITHOUT_ACC": + case PUT_ALL_WITHOUT_ACC: + case PUT_WITHOUT_ACC: + case REMOVE_WITHOUT_ACC: return true; default: return false; @@ -845,7 +844,8 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon } - static enum EventId { +/* static enum EventId { + CREATE_CHANNEL, HEARTBEAT, STATEFUL_UPDATE, LONG_SIZE, @@ -876,7 +876,7 @@ class ClientWiredStatelessChronicleMap implements ChronicleMap, Clon PUT_MAPPED, KEY_BUILDER, VALUE_BUILDER - } + }*/ class Entry implements Map.Entry { diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index ca042457e..f5976c242 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -23,6 +23,7 @@ import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.RemoteCallTimeoutException; import net.openhft.chronicle.hash.impl.util.BuildVersion; import net.openhft.chronicle.hash.impl.util.CloseablesManager; +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; @@ -38,7 +39,7 @@ import java.nio.channels.SocketChannel; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.ReentrantLock; -import static net.openhft.chronicle.map.ClientWiredStatelessChronicleMap.EventId.APPLICATION_VERSION; +import static net.openhft.chronicle.map.MapWireHandler.EventId.APPLICATION_VERSION; import static net.openhft.chronicle.map.MapWireHandler.SIZE_OF_SIZE; import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; @@ -233,7 +234,7 @@ public class ClientWiredStatelessTcpConnectionHub { @NotNull public String serverApplicationVersion(short channelID) { TextWire wire = new TextWire(Bytes.elasticByteBuffer()); - String result = proxyReturnString(APPLICATION_VERSION.toString(), channelID, wire); + String result = proxyReturnString(APPLICATION_VERSION, channelID, wire); return (result == null) ? "" : result; } @@ -306,9 +307,9 @@ public class ClientWiredStatelessTcpConnectionHub { final Wire wire = proxyReplyThrowable(timeoutTime, transactionId); - // handle an exception if the message contains the IS_EXCEPTION field - if (wire.read(() -> "IS_EXCEPTION").bool()) { - final String text = wire.read(() -> "EXCEPTION").text(); + // handle an exception if the message contains the isException field + if (wire.read(() -> "isException").bool()) { + final String text = wire.read(() -> "exception").text(); throw new RuntimeException(text); } return wire; @@ -363,7 +364,7 @@ public class ClientWiredStatelessTcpConnectionHub { System.out.println("\n--------------------------------\n" + "client read:\n\n" + Bytes.toDebugString(intWire.bytes())); - long transactionId0 = intWire.read(() -> "TRANSACTION_ID").int64(); + long transactionId0 = intWire.read(MapWireHandlerBuilder.Fields.transactionId).int64(); // check the transaction id is reasonable @@ -552,7 +553,7 @@ public class ClientWiredStatelessTcpConnectionHub { } - private long proxySend(@NotNull final String methodName, + private long proxySend(@NotNull final EventId methodName, final long startTime, short channelID, @NotNull final Wire wire) { @@ -564,7 +565,7 @@ public class ClientWiredStatelessTcpConnectionHub { outBytesLock().lock(); try { long transactionId = writeHeader(startTime, channelID, wire); - wire.write(() -> "METHOD_NAME").text(methodName); + wire.write(MapWireHandlerBuilder.Fields.methodName).text(methodName.toString()); writeSocket(wire); return transactionId; } finally { @@ -574,13 +575,13 @@ public class ClientWiredStatelessTcpConnectionHub { @SuppressWarnings("SameParameterValue") @Nullable - String proxyReturnString(@NotNull final String messageId, short channelID) { + String proxyReturnString(@NotNull final EventId messageId, short channelID) { return proxyReturnString(messageId, channelID, outWire); } @SuppressWarnings("SameParameterValue") @Nullable - String proxyReturnString(@NotNull final String messageId, short channelID, Wire outWire) { + String proxyReturnString(@NotNull final EventId messageId, short channelID, Wire outWire) { final long startTime = System.currentTimeMillis(); long transactionId; @@ -596,7 +597,7 @@ public class ClientWiredStatelessTcpConnectionHub { // receive inBytesLock().lock(); try { - return proxyReply(timeoutTime, transactionId).read(() -> "RESULT").text(); + return proxyReply(timeoutTime, transactionId).read(() -> "result").text(); } catch (Exception e) { throw new RuntimeException(e); } finally { @@ -617,10 +618,10 @@ public class ClientWiredStatelessTcpConnectionHub { startTime(startTime); long transactionId = nextUniqueTransaction(startTime); - wire.write(TYPE).text("MAP"); - wire.write(TRANSACTION_ID).int64(transactionId); - wire.write(TIME_STAMP).int64(startTime); - wire.write(CHANNEL_ID).int16(channelID); + wire.write(type).text("MAP"); + wire.write(MapWireHandlerBuilder.Fields.transactionId).int64(transactionId); + wire.write(timeStamp).int64(startTime); + wire.write(channelId).int16(channelID); return transactionId; } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index b28f10601..ba906c76d 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -26,6 +26,7 @@ import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; import net.openhft.chronicle.hash.impl.util.BuildVersion; import net.openhft.chronicle.hash.replication.ReplicationHub; +import net.openhft.chronicle.map.MapWireHandlerBuilder.Fields; import net.openhft.chronicle.network2.WireHandler; import net.openhft.chronicle.network2.event.EventGroup; import net.openhft.chronicle.network2.event.WireHandlers; @@ -49,6 +50,7 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; +import static net.openhft.chronicle.map.MapWireHandler.EventId.*; import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; /** @@ -56,9 +58,10 @@ import static net.openhft.chronicle.map.MapWireHandlerBuilder.Fields.*; */ class MapWireHandler implements WireHandler, Consumer { - private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); - public static final int SIZE_OF_SIZE = 2; + public static final int SIZE_OF_SIZE = 2; + private static final Logger LOG = LoggerFactory.getLogger(MapWireHandler.class); + final Map incompleteWork = new HashMap(); private final ArrayList bytesChronicleMaps = new ArrayList<>(); @NotNull @@ -70,19 +73,10 @@ class MapWireHandler implements WireHandler, Consumer { @Override public void accept(Iterator iterator) { - outWire.write(RESULT); + outWire.write(result); outWire.bytes().write(iterator.next()); } }; - - private WireHandlers publishLater; - - @Override - public void accept(WireHandlers wireHandlers) { - this.publishLater = wireHandlers; - } - - private final Consumer writeEntry = new Consumer>>() { @Override @@ -90,21 +84,20 @@ class MapWireHandler implements WireHandler, Consumer { final Map.Entry entry = iterator.next(); - outWire.write(RESULT_KEY); + outWire.write(resultKey); outWire.bytes().write(entry.getKey()); - outWire.write(RESULT_VALUE); + outWire.write(resultValue); outWire.bytes().write(entry.getValue()); } }; + private WireHandlers publishLater; private byte localIdentifier; private long timestamp; private short channelId; private List channelList; private ReplicationHub hub; private byte remoteIdentifier; - - public MapWireHandler(@NotNull final Supplier>> chronicleHashInstanceBuilder, @NotNull final ReplicationHub hub, byte localIdentifier, @@ -112,28 +105,19 @@ class MapWireHandler implements WireHandler, Consumer { this(chronicleHashInstanceBuilder, hub); this.channelList = channelList; this.localIdentifier = localIdentifier; - - } + public MapWireHandler(Supplier>> chronicleHashInstanceBuilder, ReplicationHub hub) { this.chronicleHashInstanceBuilder = chronicleHashInstanceBuilder; this.hub = hub; } - @Override - public void process(Wire in, Wire out) throws StreamCorruptedException { - try { - this.inWire = in; - this.outWire = out; - onEvent(); - } catch (Exception e) { - LOG.error("", e); - } + public void accept(WireHandlers wireHandlers) { + this.publishLater = wireHandlers; } - private void writeChunked(long transactionId, @NotNull final Function function, @NotNull final Consumer c) throws StreamCorruptedException { @@ -145,7 +129,7 @@ class MapWireHandler implements WireHandler, Consumer { @Override public void process(Wire in, Wire out) throws StreamCorruptedException { - outWire.write(TRANSACTION_ID).int64(transactionId); + outWire.write(Fields.transactionId).int64(transactionId); // this allows us to write more data than the buffer will allow for (; ; ) { @@ -154,7 +138,7 @@ class MapWireHandler implements WireHandler, Consumer { write(map -> { - outWire.write(HAS_NEXT).bool(hasNext); + outWire.write(Fields.hasNext).bool(hasNext); if (hasNext) c.accept(iterator); @@ -178,8 +162,224 @@ class MapWireHandler implements WireHandler, Consumer { } + @Override + public void process(Wire in, Wire out) throws StreamCorruptedException { + try { + this.inWire = in; + this.outWire = out; + onEvent(); + } catch (Exception e) { + LOG.error("", e); + } + } - final Map incompleteWork = new HashMap(); + @SuppressWarnings("UnusedReturnValue") + void onEvent() throws StreamCorruptedException { + + // it is assumed by this point that the buffer has all the bytes in it for this message + + long transactionId = inWire.read(Fields.transactionId).int64(); + timestamp = inWire.read(timeStamp).int64(); + channelId = inWire.read(Fields.channelId).int16(); + + final StringBuilder methodName = Wires.acquireStringBuilder(); + inWire.read(Fields.methodName).text(methodName); + + if (!incompleteWork.isEmpty()) { + Runnable runnable = incompleteWork.get(transactionId); + if (runnable != null) { + runnable.run(); + return; + } + } + try { + + if (PUT_WITHOUT_ACC.toString().contentEquals(methodName)) { + writeVoid(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, arg1, arg2); + // todo bytesMap.put(reader, reader, timestamp, identifier()); + bytesMap.put(reader, reader); + }); + return; + } + + if (KEY_SET.toString().contentEquals(methodName)) { + writeChunked(transactionId, map -> map.keySet().iterator(), writeElement); + return; + } + + if (VALUES.toString().contentEquals(methodName)) { + writeChunked(transactionId, map -> map.delegate.values().iterator(), writeElement); + return; + } + + if (ENTRY_SET.toString().contentEquals(methodName)) { + writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry); + return; + } + + if (PUT_ALL.toString().contentEquals(methodName)) { + putAll(transactionId); + return; + } + + // write the transaction id + outWire.write(Fields.transactionId).int64(transactionId); + + + if (CREATE_CHANNEL.toString().contentEquals(methodName)) { + writeVoid(() -> { + short channelId1 = inWire.read(arg1).int16(); + chronicleHashInstanceBuilder.get().replicatedViaChannel(hub.createChannel(channelId1)).create(); + return null; + }); + return; + } + + + if (REMOTE_IDENTIFIER.toString().contentEquals(methodName)) { + this.remoteIdentifier = inWire.read(result).int8(); + return; + } + + if (LONG_SIZE.toString().contentEquals(methodName)) { + write(b -> outWire.write(result).int64(b.longSize())); + return; + } + + if (IS_EMPTY.toString().contentEquals(methodName)) { + write(b -> outWire.write(result).bool(b.isEmpty())); + return; + } + + if (CONTAINS_KEY.toString().contentEquals(methodName)) { + write(b -> outWire.write(result).bool(b.delegate.containsKey(toByteArray(inWire, arg1)))); + return; + } + + if (CONTAINS_VALUE.toString().contentEquals(methodName)) { + write(b -> outWire.write(result).bool(b.delegate.containsValue(toByteArray(inWire, arg1)))); + return; + } + + if (GET.toString().contentEquals(methodName)) { + writeValueUsingDelegate(map -> map.get(toByteArray(inWire, arg1))); + return; + } + + if (PUT.toString().contentEquals(methodName)) { + writeValue(b -> { + final net.openhft.lang.io.Bytes reader = MapWireHandler.this.toReader(inWire, arg1, arg2); + + // todo call return b.put(reader, reader, timestamp, identifier()); + return b.put(reader, reader); + }); + + return; + } + + if (REMOVE.toString().contentEquals(methodName)) { + writeValue(b -> b.remove(toReader(inWire, arg1))); + return; + } + + if (CLEAR.toString().contentEquals(methodName)) { + writeVoid(BytesChronicleMap::clear); + return; + } + + if (REPLACE.toString().contentEquals(methodName)) { + write(bytesMap -> { + + + // todo fix this this is a hack to get to work for now. + // todo may use somehting like : + // todo bytesMap.replace(reader, reader, timestamp, identifier()); + + + VanillaChronicleMap map = bytesMap.delegate; + byte[] result = (byte[]) map.replace( + toByteArray(inWire, arg1), + toByteArray(inWire, arg2)); + + boolean isNull = result == null || result.length == 0; + outWire.write(resultIsNull).bool(isNull); + if (!isNull) { + outWire.write(Fields.result); + outWire.bytes().write(result); + } + + }); + return; + } + + if (REPLACE_WITH_OLD_AND_NEW_VALUE.toString().contentEquals(methodName)) { + + write(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, arg1, arg2, arg3); + boolean result = bytesMap.replace(reader, reader, reader); + outWire.write(Fields.result).bool(result); + }); + + return; + } + + if (PUT_IF_ABSENT.toString().contentEquals(methodName)) { + + + // todo call bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); + writeValueFromBytes(b -> { + byte[] key = MapWireHandler.this.toByteArray(inWire, arg1); + byte[] value = MapWireHandler.this.toByteArray(inWire, arg2); + return ((Map) b.delegate).putIfAbsent(key, value); + }); + + + return; + } + + if (REMOVE_WITH_VALUE.toString().contentEquals(methodName)) { + write(bytesMap -> { + final net.openhft.lang.io.Bytes reader = toReader(inWire, arg1, arg2); + // todo call outWire.write(result).bool(bytesMap.remove(reader, reader, timestamp, identifier())); + outWire.write(result).bool(bytesMap.remove(reader, reader)); + }); + return; + } + + + if (APPLICATION_VERSION.toString().contentEquals(methodName)) { + write(b -> outWire.write(result).text(applicationVersion())); + return; + } + + if (PERSISTED_DATA_VERSION.toString().contentEquals(methodName)) { + write(b -> outWire.write(result).text(persistedDataVersion())); + + return; + } + + if (HASH_CODE.toString().contentEquals(methodName)) { + write(b -> outWire.write(result).int32(b.hashCode())); + return; + } + + throw new IllegalStateException("unsupported event=" + methodName); + + } finally { + + // if (len > 4) + if (EventGroup.IS_DEBUG) { + long len = outWire.bytes().position() - SIZE_OF_SIZE; + if (len == 0) + System.out.println("--------------------------------------------\nserver wrote:\n\n"); + else + System.out.println("--------------------------------------------\nserver wrote:\n\n" + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len)); + } + } + + + } private byte[] toByteArray(net.openhft.lang.io.Bytes bytes) { if (bytes == null || bytes.remaining() == 0) @@ -219,214 +419,6 @@ class MapWireHandler implements WireHandler, Consumer { } } - @SuppressWarnings("UnusedReturnValue") - void onEvent() throws StreamCorruptedException { - - // it is assumed by this point that the buffer has all the bytes in it for this message - - long transactionId = inWire.read(TRANSACTION_ID).int64(); - timestamp = inWire.read(TIME_STAMP).int64(); - channelId = inWire.read(CHANNEL_ID).int16(); - - final StringBuilder methodName = Wires.acquireStringBuilder(); - inWire.read(METHOD_NAME).text(methodName); - - if (!incompleteWork.isEmpty()) { - Runnable runnable = incompleteWork.get(transactionId); - if (runnable != null) { - runnable.run(); - return; - } - } - try { - - if ("PUT_WITHOUT_ACC".contentEquals(methodName)) { - writeVoid(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); - // todo bytesMap.put(reader, reader, timestamp, identifier()); - bytesMap.put(reader, reader); - }); - return; - } - - if ("KEY_SET".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.keySet().iterator(), writeElement); - return; - } - - if ("VALUES".contentEquals(methodName)) { - writeChunked(transactionId, map -> map.delegate.values().iterator(), writeElement); - return; - } - - if ("ENTRY_SET".contentEquals(methodName)) { - writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry); - return; - } - - if ("PUT_ALL".contentEquals(methodName)) { - putAll(transactionId); - return; - } - - // write the transaction id - outWire.write(() -> "TRANSACTION_ID").int64(transactionId); - - - if ("CREATE_CHANNEL".contentEquals(methodName)) { - writeVoid(() -> { - short channelId1 = inWire.read(ARG_1).int16(); - chronicleHashInstanceBuilder.get().replicatedViaChannel(hub.createChannel(channelId1)).create(); - return null; - }); - return; - } - - - if ("REMOTE_IDENTIFIER".contentEquals(methodName)) { - this.remoteIdentifier = inWire.read(RESULT).int8(); - return; - } - - if ("LONG_SIZE".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).int64(b.longSize())); - return; - } - - if ("IS_EMPTY".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).bool(b.isEmpty())); - return; - } - - if ("CONTAINS_KEY".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).bool(b.delegate.containsKey(toByteArray(inWire, ARG_1)))); - return; - } - - if ("CONTAINS_VALUE".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).bool(b.delegate.containsValue(toByteArray(inWire, ARG_1)))); - return; - } - - if ("GET".contentEquals(methodName)) { - writeValueUsingDelegate(map -> map.get(toByteArray(inWire, ARG_1))); - return; - } - - if ("PUT".contentEquals(methodName)) { - writeValue(b -> { - final net.openhft.lang.io.Bytes reader = MapWireHandler.this.toReader(inWire, ARG_1, ARG_2); - - // todo call return b.put(reader, reader, timestamp, identifier()); - return b.put(reader, reader); - }); - - return; - } - - if ("REMOVE".contentEquals(methodName)) { - writeValue(b -> b.remove(toReader(inWire, ARG_1))); - return; - } - - if ("CLEAR".contentEquals(methodName)) { - writeVoid(BytesChronicleMap::clear); - return; - } - - if ("REPLACE".contentEquals(methodName)) { - write(bytesMap -> { - - - // todo fix this this is a hack to get to work for now. - // todo may use somehting like : - // todo bytesMap.replace(reader, reader, timestamp, identifier()); - - - VanillaChronicleMap map = bytesMap.delegate; - byte[] result = (byte[]) map.replace( - toByteArray(inWire, ARG_1), - toByteArray(inWire, ARG_2)); - - boolean isNull = result == null || result.length == 0; - outWire.write(() -> "RESULT_IS_NULL").bool(isNull); - if (!isNull) { - outWire.write(() -> "RESULT"); - outWire.bytes().write(result); - } - - }); - return; - } - - if ("REPLACE_WITH_OLD_AND_NEW_VALUE".contentEquals(methodName)) { - - write(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2, ARG_3); - boolean result = bytesMap.replace(reader, reader, reader); - outWire.write(RESULT).bool(result); - }); - - return; - } - - if ("PUT_IF_ABSENT".contentEquals(methodName)) { - - - // todo call bytesMap.putIfAbsent(reader, reader, timestamp, identifier()); - writeValueFromBytes(b -> { - byte[] key = MapWireHandler.this.toByteArray(inWire, ARG_1); - byte[] value = MapWireHandler.this.toByteArray(inWire, ARG_2); - return ((Map) b.delegate).putIfAbsent(key, value); - }); - - - return; - } - - if ("REMOVE_WITH_VALUE".contentEquals(methodName)) { - write(bytesMap -> { - final net.openhft.lang.io.Bytes reader = toReader(inWire, ARG_1, ARG_2); - // todo call outWire.write(RESULT).bool(bytesMap.remove(reader, reader, timestamp, identifier())); - outWire.write(RESULT).bool(bytesMap.remove(reader, reader)); - }); - return; - } - - - if ("APPLICATION_VERSION".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).text(applicationVersion())); - return; - } - - if ("PERSISTED_DATA_VERSION".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).text(persistedDataVersion())); - - return; - } - - if ("HASH_CODE".contentEquals(methodName)) { - write(b -> outWire.write(RESULT).int32(b.hashCode())); - return; - } - - throw new IllegalStateException("unsupported event=" + methodName); - - } finally { - - // if (len > 4) - if (EventGroup.IS_DEBUG) { - long len = outWire.bytes().position() - SIZE_OF_SIZE; - if (len == 0) - System.out.println("--------------------------------------------\nserver wrote:\n\n"); - else - System.out.println("--------------------------------------------\nserver wrote:\n\n" + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len)); - } - } - - - } - private void putAll(long transactionId) { final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId); @@ -457,17 +449,17 @@ class MapWireHandler implements WireHandler, Consumer { // the old code assumed that ALL of the entries would fit into a single buffer // this assumption is invalid boolean hasNext; - for(;;) { + for (; ; ) { - hasNext = inWire.read(HAS_NEXT).bool(); + hasNext = inWire.read(Fields.hasNext).bool(); - collectData.put(toBytes(ARG_1), toBytes(ARG_2)); + collectData.put(toBytes(arg1), toBytes(arg2)); if (!hasNext) { incompleteWork.remove(transactionId); - outWire.write(TRANSACTION_ID).int64(transactionId); + outWire.write(Fields.transactionId).int64(transactionId); writeVoid(() -> { bytesMap.delegate.putAll((Map) collectData); @@ -486,6 +478,23 @@ class MapWireHandler implements WireHandler, Consumer { } + @SuppressWarnings("SameReturnValue") + private void writeValueFromBytes(final Function f) { + + write(b -> { + + byte[] fromBytes = f.apply(b); + boolean isNull = fromBytes == null || fromBytes.length == 0; + outWire.write(resultIsNull).bool(isNull); + if (isNull) + return; + + outWire.write(result); + outWire.bytes().write(fromBytes); + + }); + } + private byte identifier() { // if the client provides a remote identifier then we will use that otherwise we will use the local identifier return remoteIdentifier == 0 ? localIdentifier : remoteIdentifier; @@ -640,24 +649,6 @@ class MapWireHandler implements WireHandler, Consumer { writeValueFromBytes(b -> toByteArray(f.apply(b))); } - - @SuppressWarnings("SameReturnValue") - private void writeValueFromBytes(final Function f) { - - write(b -> { - - byte[] fromBytes = f.apply(b); - boolean isNull = fromBytes == null || fromBytes.length == 0; - outWire.write(RESULT_IS_NULL).bool(isNull); - if (isNull) - return; - - outWire.write(RESULT); - outWire.bytes().write(fromBytes); - - }); - } - @SuppressWarnings("SameReturnValue") private void writeValueUsingDelegate(final Function, byte[]> f) { @@ -666,13 +657,13 @@ class MapWireHandler implements WireHandler, Consumer { byte[] result = f.apply((ChronicleMap) b.delegate); boolean isNull = result == null; - outWire.write(RESULT_IS_NULL).bool(isNull); + outWire.write(resultIsNull).bool(isNull); if (isNull) return; isNull = result.length == 0; - outWire.write(RESULT); + outWire.write(Fields.result); outWire.bytes().write(result); }); @@ -680,7 +671,6 @@ class MapWireHandler implements WireHandler, Consumer { } - @SuppressWarnings("SameReturnValue") private void write(@NotNull Consumer c) { @@ -693,7 +683,7 @@ class MapWireHandler implements WireHandler, Consumer { bytesMap.output = null; outWire.bytes().mark(); - outWire.write(IS_EXCEPTION).bool(false); + outWire.write(isException).bool(false); try { c.accept(bytesMap); @@ -701,8 +691,8 @@ class MapWireHandler implements WireHandler, Consumer { outWire.bytes().reset(); // the idea of wire is that is platform independent, // so we wil have to send the exception as a String - outWire.write(IS_EXCEPTION).bool(true); - outWire.write(EXCEPTION).text(toString(e)); + outWire.write(isException).bool(true); + outWire.write(exception).text(toString(e)); LOG.error("", e); return; } @@ -724,27 +714,19 @@ class MapWireHandler implements WireHandler, Consumer { try { r.call(); - outWire.write(IS_EXCEPTION).bool(false); + outWire.write(isException).bool(false); } catch (Exception e) { outWire.bytes().reset(); // the idea of wire is that is platform independent, // so we wil have to send the exception as a String - outWire.write(IS_EXCEPTION).bool(true); - outWire.write(EXCEPTION).text(toString(e)); + outWire.write(isException).bool(true); + outWire.write(exception).text(toString(e)); LOG.error("", e); return; } } - /** - * only used for debugging - */ - @SuppressWarnings("UnusedDeclaration") - private void showOutWire() { - System.out.println("pos=" + outWire.bytes().position() + ",bytes=" + Bytes.toDebugString(outWire.bytes(), 0, outWire.bytes().position())); - } - @SuppressWarnings("SameReturnValue") private void writeVoid(@NotNull Consumer process) { @@ -759,14 +741,22 @@ class MapWireHandler implements WireHandler, Consumer { bytesMap.output = null; try { process.accept(bytesMap); - outWire.write(IS_EXCEPTION).bool(false); + outWire.write(isException).bool(false); } catch (Exception e) { - outWire.write(IS_EXCEPTION).bool(true); + outWire.write(isException).bool(true); LOG.error("", e); } } + /** + * only used for debugging + */ + @SuppressWarnings("UnusedDeclaration") + private void showOutWire() { + System.out.println("pos=" + outWire.bytes().position() + ",bytes=" + Bytes.toDebugString(outWire.bytes(), 0, outWire.bytes().position())); + } + /** * converts the exception into a String, so that it can be sent to c# clients */ @@ -781,5 +771,40 @@ class MapWireHandler implements WireHandler, Consumer { this.localIdentifier = localIdentifier; } + static enum EventId { + HEARTBEAT, + STATEFUL_UPDATE, + LONG_SIZE, + SIZE, + IS_EMPTY, + CONTAINS_KEY, + CONTAINS_VALUE, + GET, + PUT, + PUT_WITHOUT_ACC, + REMOVE, + REMOVE_WITHOUT_ACC, + CLEAR, + KEY_SET, + VALUES, + ENTRY_SET, + REPLACE, + REPLACE_WITH_OLD_AND_NEW_VALUE, + PUT_IF_ABSENT, + REMOVE_WITH_VALUE, + TO_STRING, + APPLICATION_VERSION, + PERSISTED_DATA_VERSION, + PUT_ALL, + PUT_ALL_WITHOUT_ACC, + HASH_CODE, + MAP_FOR_KEY, + PUT_MAPPED, + KEY_BUILDER, + VALUE_BUILDER, + CREATE_CHANNEL, + REMOTE_IDENTIFIER + } + } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java index f67efcda8..4dfd992db 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java @@ -21,7 +21,6 @@ package net.openhft.chronicle.map; import net.openhft.chronicle.hash.ChronicleHashInstanceBuilder; import net.openhft.chronicle.hash.replication.ReplicationHub; import net.openhft.chronicle.network2.WireHandler; - import net.openhft.chronicle.wire.WireKey; import org.jetbrains.annotations.NotNull; @@ -41,22 +40,23 @@ public class MapWireHandlerBuilder { return new MapWireHandler(chronicleHashInstanceBuilder, hub, localIdentifier, channelList); } + // note : peter has asked for these to be in camel case public static enum Fields implements WireKey { - HAS_NEXT, - TIME_STAMP, - CHANNEL_ID, - METHOD_NAME, - TYPE, - TRANSACTION_ID, - RESULT, - RESULT_KEY, - RESULT_VALUE, - ARG_1, - ARG_2, - ARG_3, - IS_EXCEPTION, - EXCEPTION, - RESULT_IS_NULL + hasNext, + timeStamp, + channelId, + methodName, + type, + transactionId, + result, + resultKey, + resultValue, + arg1, + arg2, + arg3, + isException, + exception, + resultIsNull } }