diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index ee09139cc..788c63c57 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -18,6 +18,7 @@ package net.openhft.chronicle.map; +import net.openhft.chronicle.bytes.Bytes; import net.openhft.chronicle.hash.function.SerializableFunction; import net.openhft.chronicle.map.MapWireHandlerBuilder.Fields; import net.openhft.chronicle.wire.Marshallable; @@ -282,7 +283,7 @@ class ClientWiredStatelessChronicleMap public void putAll(@NotNull Map map) { final long startTime = System.currentTimeMillis(); - long transactionId = hub.nextUniqueTransaction(startTime); + long tid = hub.nextUniqueTransaction(startTime); Set> entries = map.entrySet(); Iterator> iterator = entries.iterator(); @@ -301,7 +302,7 @@ class ClientWiredStatelessChronicleMap hub.markSize(wire); hub.startTime(startTime); wire.write(Fields.type).text("MAP"); - wire.write(Fields.transactionId).int64(transactionId); + wire.write(Fields.tid).int64(tid); wire.write(Fields.timeStamp).int64(startTime); wire.write(Fields.channelId).int16(channelID); hub.outWire().write(Fields.methodName).text(putAll.toString()); @@ -315,7 +316,8 @@ class ClientWiredStatelessChronicleMap writeField(Fields.arg1, e.getKey()); writeField(Fields.arg2, e.getValue()); - if (wire.bytes().remaining() < wire.bytes().capacity() * 0.5) + final Bytes bytes = wire.bytes(); + if (bytes.remaining() < bytes.capacity() * 0.5) break OUTER; } @@ -329,7 +331,7 @@ class ClientWiredStatelessChronicleMap // wait for the transaction id to be received, this will only be received once the // last chunk has been processed - readVoid(transactionId, startTime); + readVoid(tid, startTime); } @@ -374,7 +376,7 @@ class ClientWiredStatelessChronicleMap final long startTime = System.currentTimeMillis(); // send - final long transactionId = proxySend(values, startTime); + final long tid = proxySend(values, startTime); assert !hub.outBytesLock().isHeldByCurrentThread(); final long timeoutTime = startTime + hub.timeoutMs; @@ -384,9 +386,10 @@ class ClientWiredStatelessChronicleMap // receive hub.inBytesLock().lock(); try { - final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + final Wire wireIn = hub.proxyReply(timeoutTime, tid); - while (wireIn.bytes().remaining() > 0) { + final Bytes bytes = wireIn.bytes(); + while (bytes.remaining() > 0) { boolean bool = wireIn.read(Fields.hasNext).bool(); if (bool) { usingCollection.add(readObject(Fields.result, wireIn, null, vClass1)); @@ -413,7 +416,7 @@ class ClientWiredStatelessChronicleMap final Map result = new HashMap(); final long startTime = System.currentTimeMillis(); // send - final long transactionId = proxySend(entrySet, startTime); + final long tid = proxySend(entrySet, startTime); assert !hub.outBytesLock().isHeldByCurrentThread(); final long timeoutTime = startTime + hub.timeoutMs; @@ -423,9 +426,10 @@ class ClientWiredStatelessChronicleMap // receive hub.inBytesLock().lock(); try { - final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + final Wire wireIn = hub.proxyReply(timeoutTime, tid); - while (wireIn.bytes().remaining() > 0) { + final Bytes bytes = wireIn.bytes(); + while (bytes.remaining() > 0) { boolean bool = wireIn.read(Fields.hasNext).bool(); if (bool) { result.put( @@ -435,7 +439,7 @@ class ClientWiredStatelessChronicleMap break OUTER; - if (wireIn.bytes().remaining() > 0) { + if (bytes.remaining() > 0) { // todo process the exception boolean isException = wireIn.read(Fields.isException).bool(); } @@ -460,7 +464,7 @@ class ClientWiredStatelessChronicleMap final long startTime = System.currentTimeMillis(); // send - final long transactionId = proxySend(entrySetRestricted, maxNumEntries, startTime); + final long tid = proxySend(entrySetRestricted, maxNumEntries, startTime); assert !hub.outBytesLock().isHeldByCurrentThread(); final long timeoutTime = startTime + hub.timeoutMs; @@ -470,9 +474,10 @@ class ClientWiredStatelessChronicleMap // receive hub.inBytesLock().lock(); try { - final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + final Wire wireIn = hub.proxyReply(timeoutTime, tid); - while (wireIn.bytes().remaining() > 0) { + final Bytes bytes = wireIn.bytes(); + while (bytes.remaining() > 0) { boolean bool = wireIn.read(Fields.hasNext).bool(); if (bool) { result.put( @@ -482,7 +487,7 @@ class ClientWiredStatelessChronicleMap break OUTER; - if (wireIn.bytes().remaining() > 0) { + if (bytes.remaining() > 0) { boolean isException = wireIn.read(Fields.isException).bool(); if (isException) throw new RuntimeException(wireIn.read(Fields.result).text()); @@ -506,14 +511,14 @@ class ClientWiredStatelessChronicleMap } - private long readLong(long transactionId, long startTime) { + private long readLong(long tid, long startTime) { assert !hub.outBytesLock().isHeldByCurrentThread(); final long timeoutTime = startTime + hub.timeoutMs; // receive hub.inBytesLock().lock(); try { - return hub.proxyReply(timeoutTime, transactionId).read(Fields.result).int64(); + return hub.proxyReply(timeoutTime, tid).read(Fields.result).int64(); } finally { hub.inBytesLock().unlock(); } @@ -551,14 +556,14 @@ class ClientWiredStatelessChronicleMap } - private V readValue(Fields argName, long transactionId, long startTime, final V usingValue) { + private V readValue(Fields argName, long tid, long startTime, final V usingValue) { assert !hub.outBytesLock().isHeldByCurrentThread(); long timeoutTime = startTime + hub.timeoutMs; hub.inBytesLock().lock(); try { - final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + final Wire wireIn = hub.proxyReply(timeoutTime, tid); if (wireIn.read(Fields.resultIsNull).bool()) return null; @@ -637,7 +642,7 @@ class ClientWiredStatelessChronicleMap } - private boolean readBoolean(long transactionId, long startTime) { + private boolean readBoolean(long tid, long startTime) { assert !hub.outBytesLock().isHeldByCurrentThread(); long timeoutTime = startTime + hub.timeoutMs; @@ -645,14 +650,14 @@ class ClientWiredStatelessChronicleMap // receive hub.inBytesLock().lock(); try { - final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + final Wire wireIn = hub.proxyReply(timeoutTime, tid); return wireIn.read(Fields.result).bool(); } finally { hub.inBytesLock().unlock(); } } - private int readInt(long transactionId, long startTime) { + private int readInt(long tid, long startTime) { assert !hub.outBytesLock().isHeldByCurrentThread(); long timeoutTime = startTime + hub.timeoutMs; @@ -660,7 +665,7 @@ class ClientWiredStatelessChronicleMap // receive hub.inBytesLock().lock(); try { - final Wire wireIn = hub.proxyReply(timeoutTime, transactionId); + final Wire wireIn = hub.proxyReply(timeoutTime, tid); return wireIn.read(Fields.result).int32(); } finally { hub.inBytesLock().unlock(); @@ -671,11 +676,11 @@ class ClientWiredStatelessChronicleMap private boolean proxyReturnBoolean(@NotNull final String methodName, K key, V value) { final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); writeField(Fields.arg1, key); writeField(Fields.arg2, value); @@ -685,7 +690,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().unlock(); } - return readBoolean(transactionId, startTime); + return readBoolean(tid, startTime); } @@ -694,10 +699,10 @@ class ClientWiredStatelessChronicleMap @NotNull final String methodName, K key, V value1, V value2) { final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); writeField(Fields.arg1, key); writeField(Fields.arg2, value1); @@ -707,7 +712,7 @@ class ClientWiredStatelessChronicleMap hub.outBytesLock().unlock(); } - return readBoolean(transactionId, startTime); + return readBoolean(tid, startTime); } @@ -728,36 +733,36 @@ class ClientWiredStatelessChronicleMap private long proxyReturnLong(@NotNull final String methodName) { final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); } - return readLong(transactionId, startTime); + return readLong(tid, startTime); } @SuppressWarnings("SameParameterValue") private boolean proxyReturnBoolean(@NotNull final String methodName) { final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); } - return readBoolean(transactionId, startTime); + return readBoolean(tid, startTime); } @SuppressWarnings("SameParameterValue") @@ -765,18 +770,18 @@ class ClientWiredStatelessChronicleMap final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); r.run(); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); } - readVoid(transactionId, startTime); + readVoid(tid, startTime); } @@ -784,18 +789,18 @@ class ClientWiredStatelessChronicleMap private int proxyReturnInt(@NotNull final String methodName) { final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); } - return readInt(transactionId, startTime); + return readInt(tid, startTime); } @@ -804,14 +809,14 @@ class ClientWiredStatelessChronicleMap @NotNull final EventId methodName, K key, V value, Class resultType) { final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; hub.outBytesLock().lock(); try { assert hub.outBytesLock().isHeldByCurrentThread(); assert !hub.inBytesLock().isHeldByCurrentThread(); - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); writeField(Fields.arg1, key); writeField(Fields.arg2, value); @@ -825,7 +830,7 @@ class ClientWiredStatelessChronicleMap return null; if (resultType == vClass) - return (R) readValue(Fields.result, transactionId, startTime, null); + return (R) readValue(Fields.result, tid, startTime, null); else throw new UnsupportedOperationException("class of type class=" + resultType + @@ -837,60 +842,60 @@ class ClientWiredStatelessChronicleMap Class rClass, @NotNull final EventId methodName, Object key) { final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; - transactionId = proxySend(methodName, startTime, key); + tid = proxySend(methodName, startTime, key); if (eventReturnsNull(methodName)) return null; if (rClass == vClass) - return (R) readValue(Fields.result, transactionId, startTime, null); + return (R) readValue(Fields.result, tid, startTime, null); else throw new UnsupportedOperationException("class of type class=" + rClass + " is not " + "supported"); } private long proxySend(EventId methodName, long startTime, Object arg1) { - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); writeField(Fields.arg1, arg1); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); } - return transactionId; + return tid; } private long proxySend(EventId methodName, long startTime) { - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); } - return transactionId; + return tid; } private long proxySend(EventId methodName, long arg1, long startTime) { - long transactionId; + long tid; hub.outBytesLock().lock(); try { - transactionId = hub.writeHeader(startTime, channelID, hub.outWire()); + tid = hub.writeHeader(startTime, channelID, hub.outWire()); hub.outWire().write(Fields.methodName).text(methodName.toString()); hub.outWire().write(Fields.arg1).int64(arg1); hub.writeSocket(hub.outWire()); } finally { hub.outBytesLock().unlock(); } - return transactionId; + return tid; } @@ -907,13 +912,13 @@ class ClientWiredStatelessChronicleMap } - private void readVoid(long transactionId, long startTime) { + private void readVoid(long tid, long startTime) { long timeoutTime = startTime + hub.timeoutMs; // receive hub.inBytesLock().lock(); try { - hub.proxyReply(timeoutTime, transactionId); + hub.proxyReply(timeoutTime, tid); } finally { hub.inBytesLock().unlock(); } diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index 4d9072b36..4bd3ab87c 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -289,27 +289,28 @@ public class ClientWiredStatelessTcpConnectionHub { private void writeLength(Wire outWire) { assert outBytesLock().isHeldByCurrentThread(); - long position = outWire.bytes().position(); + final Bytes bytes = outWire.bytes(); + long position = bytes.position(); if (position > Integer.MAX_VALUE || position < Integer.MIN_VALUE) throw new IllegalStateException("message too large"); - long pos = outWire.bytes().position(); + long pos = bytes.position(); try { - outWire.bytes().reset(); - int size = (int) (position - outWire.bytes().position()); - outWire.bytes().writeUnsignedShort(size - SIZE_OF_SIZE); + bytes.reset(); + int size = (int) (position - bytes.position()); + bytes.writeUnsignedShort(size - SIZE_OF_SIZE); } finally { - outWire.bytes().position(pos); + bytes.position(pos); } } - protected Wire proxyReply(long timeoutTime, final long transactionId) { + protected Wire proxyReply(long timeoutTime, final long tid) { assert inBytesLock().isHeldByCurrentThread(); try { - final Wire wire = proxyReplyThrowable(timeoutTime, transactionId); + final Wire wire = proxyReplyThrowable(timeoutTime, tid); // handle an exception if the message contains the isException field if (wire.read(() -> "isException").bool()) { @@ -333,7 +334,7 @@ public class ClientWiredStatelessTcpConnectionHub { } - private Wire proxyReplyThrowable(long timeoutTime, long transactionId) throws IOException { + private Wire proxyReplyThrowable(long timeoutTime, long tid) throws IOException { assert inBytesLock().isHeldByCurrentThread(); @@ -345,7 +346,8 @@ public class ClientWiredStatelessTcpConnectionHub { assert parkedTransactionTimeStamp == 0; // if we have processed all the bytes that we have read in - if (inWireByteBuffer().position() == intWire.bytes().position()) + final Bytes bytes = intWire.bytes(); + if (inWireByteBuffer().position() == bytes.position()) inWireClear(); // todo change the size to include the meta data bit @@ -353,7 +355,7 @@ public class ClientWiredStatelessTcpConnectionHub { readSocket(SIZE_OF_SIZE, timeoutTime); final int messageSize = - intWire.bytes().readUnsignedShort(intWire.bytes().position()); + bytes.readUnsignedShort(bytes.position()); assert messageSize > 0 : "Invalid message size " + messageSize; assert messageSize < 16 << 20 : "Invalid message size " + messageSize; @@ -362,23 +364,23 @@ public class ClientWiredStatelessTcpConnectionHub { readSocket(remainingBytes0, timeoutTime); - intWire.bytes().skip(SIZE_OF_SIZE); - intWire.bytes().limit(intWire.bytes().position() + messageSize); + bytes.skip(SIZE_OF_SIZE); + bytes.limit(bytes.position() + messageSize); System.out.println("\n--------------------------------\n" + - "client read:\n\n" + Bytes.toDebugString(intWire.bytes())); + "client read:\n\n" + Bytes.toDebugString(bytes)); - long transactionId0 = intWire.read(MapWireHandlerBuilder.Fields.transactionId).int64(); + long tid0 = intWire.read(MapWireHandlerBuilder.Fields.tid).int64(); // check the transaction id is reasonable - /* assert transactionId0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR : - "TransactionId too small " + transactionId0 + " messageSize " + messageSize; - assert transactionId0 < 2100000000000L * TcpReplicator.TIMESTAMP_FACTOR : - "TransactionId too large " + transactionId0 + " messageSize " + messageSize; + /* assert tid0 > 1410000000000L * TcpReplicator.TIMESTAMP_FACTOR : + "TransactionId too small " + tid0 + " messageSize " + messageSize; + assert tid0 < 2100000000000L * TcpReplicator.TIMESTAMP_FACTOR : + "TransactionId too large " + tid0 + " messageSize " + messageSize; */ // if the transaction id is for this thread process it - if (transactionId0 == transactionId) { + if (tid0 == tid) { clearParked(); return intWire; @@ -387,14 +389,14 @@ public class ClientWiredStatelessTcpConnectionHub { // if the transaction id is not for this thread, park it // and allow another thread to pick it up parkedTransactionTimeStamp = System.currentTimeMillis(); - parkedTransactionId = transactionId0; + parkedTransactionId = tid0; pause(); continue; } } // the transaction id was read by another thread, but is for this thread, process it - if (parkedTransactionId == transactionId) { + if (parkedTransactionId == tid) { clearParked(); return intWire; } @@ -424,7 +426,8 @@ public class ClientWiredStatelessTcpConnectionHub { */ private void inWireClear() { inWireByteBuffer().clear(); - intWire.bytes().clear(); + final Bytes bytes = intWire.bytes(); + bytes.clear(); } private void clearParked() { @@ -485,12 +488,14 @@ public class ClientWiredStatelessTcpConnectionHub { } private ByteBuffer inWireByteBuffer() { - return (ByteBuffer) intWire.bytes().underlyingObject(); + final Bytes bytes = intWire.bytes(); + return (ByteBuffer) bytes.underlyingObject(); } private ByteBuffer inWireByteBuffer(long requiredCapacity) { - intWire.bytes().ensureCapacity(requiredCapacity); - return (ByteBuffer) intWire.bytes().underlyingObject(); + final Bytes bytes = intWire.bytes(); + bytes.ensureCapacity(requiredCapacity); + return (ByteBuffer) bytes.underlyingObject(); } /** @@ -505,7 +510,8 @@ public class ClientWiredStatelessTcpConnectionHub { assert outBytesLock().isHeldByCurrentThread(); assert !inBytesLock().isHeldByCurrentThread(); - long outBytesPosition = outWire.bytes().position(); + final Bytes bytes = outWire.bytes(); + long outBytesPosition = bytes.position(); // if we have other threads waiting to send and the buffer is not full, @@ -515,8 +521,8 @@ public class ClientWiredStatelessTcpConnectionHub { return; } - final ByteBuffer outBuffer = (ByteBuffer) outWire.bytes().underlyingObject(); - outBuffer.limit((int) outWire.bytes().position()); + final ByteBuffer outBuffer = (ByteBuffer) bytes.underlyingObject(); + outBuffer.limit((int) bytes.position()); outBuffer.position(SIZE_OF_SIZE); if (EventGroup.IS_DEBUG ) { @@ -545,8 +551,8 @@ public class ClientWiredStatelessTcpConnectionHub { LOG.debug("continuing - without all the data being written to the buffer as " + "it will be written by the next thread"); outBuffer.compact(); - outWire.bytes().limit(outBuffer.limit()); - outWire.bytes().position(outBuffer.position()); + bytes.limit(outBuffer.limit()); + bytes.position(outBuffer.position()); return; } @@ -555,7 +561,7 @@ public class ClientWiredStatelessTcpConnectionHub { } outBuffer.clear(); - outWire.bytes().clear(); + bytes.clear(); } @@ -584,10 +590,10 @@ public class ClientWiredStatelessTcpConnectionHub { // send outBytesLock().lock(); try { - long transactionId = writeHeader(startTime, channelID, wire); + long tid = writeHeader(startTime, channelID, wire); wire.write(MapWireHandlerBuilder.Fields.methodName).text(methodName.toString()); writeSocket(wire); - return transactionId; + return tid; } finally { outBytesLock().unlock(); } @@ -603,11 +609,11 @@ public class ClientWiredStatelessTcpConnectionHub { @Nullable String proxyReturnString(@NotNull final EventId messageId, short channelID, Wire outWire) { final long startTime = System.currentTimeMillis(); - long transactionId; + long tid; outBytesLock().lock(); try { - transactionId = proxySend(messageId, startTime, channelID, outWire); + tid = proxySend(messageId, startTime, channelID, outWire); } finally { outBytesLock().unlock(); } @@ -617,7 +623,7 @@ public class ClientWiredStatelessTcpConnectionHub { // receive inBytesLock().lock(); try { - return proxyReply(timeoutTime, transactionId).read(() -> "result").text(); + return proxyReply(timeoutTime, tid).read(() -> "result").text(); } catch (Exception e) { throw new RuntimeException(e); } finally { @@ -637,13 +643,13 @@ public class ClientWiredStatelessTcpConnectionHub { markSize(wire); startTime(startTime); - long transactionId = nextUniqueTransaction(startTime); + long tid = nextUniqueTransaction(startTime); wire.write(type).text("MAP"); - wire.write(MapWireHandlerBuilder.Fields.transactionId).int64(transactionId); + wire.write(MapWireHandlerBuilder.Fields.tid).int64(tid); wire.write(timeStamp).int64(startTime); wire.write(channelId).int16(channelID); - return transactionId; + return tid; } @@ -657,10 +663,11 @@ public class ClientWiredStatelessTcpConnectionHub { assert outBytesLock().isHeldByCurrentThread(); // this is where the size will go - outWire.bytes().mark(); + final Bytes bytes = outWire.bytes(); + bytes.mark(); // skip the 2 bytes for the size - outWire.bytes().skip(2); + bytes.skip(2); } void startTime(long startTime) { diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 3f42f36d1..b9f575f1e 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -121,14 +121,14 @@ class MapWireHandler implements WireHandler, Consumer { } /** - * @param transactionId the transaction id + * @param tid the transaction id * @param function provides that returns items bases on a BytesChronicleMap * @param c an iterator that contains items * @param maxEntries the maximum number of items that can be written * @throws StreamCorruptedException */ private void writeChunked( - long transactionId, @NotNull final Function function, + long tid, @NotNull final Function function, @NotNull final Consumer c, long maxEntries) throws StreamCorruptedException { final BytesChronicleMap m = bytesMap(channelId); @@ -139,7 +139,7 @@ class MapWireHandler implements WireHandler, Consumer { @Override public void process(Wire in, Wire out) throws StreamCorruptedException { - outWire.write(Fields.transactionId).int64(transactionId); + outWire.write(Fields.tid).int64(tid); // this allows us to write more data than the buffer will allow for (int count = 0; ; count++) { @@ -178,9 +178,9 @@ class MapWireHandler implements WireHandler, Consumer { } private void writeChunked( - long transactionId, @NotNull final Function function, + long tid, @NotNull final Function function, @NotNull final Consumer c) throws StreamCorruptedException { - writeChunked(transactionId, function, c, Long.MAX_VALUE); + writeChunked(tid, function, c, Long.MAX_VALUE); } @@ -200,7 +200,7 @@ class MapWireHandler implements WireHandler, Consumer { // 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(); + long tid = inWire.read(Fields.tid).int64(); timestamp = inWire.read(timeStamp).int64(); channelId = inWire.read(Fields.channelId).int16(); @@ -208,7 +208,7 @@ class MapWireHandler implements WireHandler, Consumer { inWire.read(Fields.methodName).text(methodName); if (!incompleteWork.isEmpty()) { - Runnable runnable = incompleteWork.get(transactionId); + Runnable runnable = incompleteWork.get(tid); if (runnable != null) { runnable.run(); return; @@ -227,33 +227,33 @@ class MapWireHandler implements WireHandler, Consumer { } if (keySet.contentEquals(methodName)) { - writeChunked(transactionId, map -> map.keySet().iterator(), writeElement); + writeChunked(tid, map -> map.keySet().iterator(), writeElement); return; } if (values.contentEquals(methodName)) { - writeChunked(transactionId, map -> map.delegate.values().iterator(), writeElement); + writeChunked(tid, map -> map.delegate.values().iterator(), writeElement); return; } if (entrySet.contentEquals(methodName)) { - writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry); + writeChunked(tid, m -> m.delegate.entrySet().iterator(), writeEntry); return; } if (entrySetRestricted.contentEquals(methodName)) { long maxEntries = inWire.read(arg1).int64(); - writeChunked(transactionId, m -> m.delegate.entrySet().iterator(), writeEntry,maxEntries); + writeChunked(tid, m -> m.delegate.entrySet().iterator(), writeEntry,maxEntries); return; } if (putAll.contentEquals(methodName)) { - putAll(transactionId); + putAll(tid); return; } // write the transaction id - outWire.write(Fields.transactionId).int64(transactionId); + outWire.write(Fields.tid).int64(tid); if (createChannel.contentEquals(methodName)) { @@ -461,7 +461,7 @@ class MapWireHandler implements WireHandler, Consumer { } } - private void putAll(long transactionId) { + private void putAll(long tid) { final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId); @@ -469,7 +469,7 @@ class MapWireHandler implements WireHandler, Consumer { return; // note: a number of client threads can be using the same socket - Runnable runnable = incompleteWork.get(transactionId); + Runnable runnable = incompleteWork.get(tid); if (runnable != null) { runnable.run(); @@ -501,9 +501,9 @@ class MapWireHandler implements WireHandler, Consumer { if (!hasNext) { - incompleteWork.remove(transactionId); + incompleteWork.remove(tid); - outWire.write(Fields.transactionId).int64(transactionId); + outWire.write(Fields.tid).int64(tid); writeVoid(() -> { bytesMap.delegate.putAll((Map) collectData); @@ -518,7 +518,7 @@ class MapWireHandler implements WireHandler, Consumer { } }; - incompleteWork.put(transactionId, runnable); + incompleteWork.put(tid, runnable); runnable.run(); } diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java index 91eb3d7f0..dae60acba 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java @@ -45,7 +45,7 @@ public class MapWireHandlerBuilder { channelId, methodName, type, - transactionId, + tid, // transactionId result, resultKey, resultValue,