Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

changed "transactionId" to "tid" inline with the spec at /Users/robaustin/git-projects/Chronicle-Map/src/test

Browse Source
This commit is contained in:
Rob Austin
2015-03-31 15:30:44 +01:00
parent 4f857e1226
commit 250c5fdd46
4 changed files with 131 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java
+63 -58
View File
@@ -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<K, V>
public void putAll(@NotNull Map<? extends K, ? extends V> map) {
final long startTime = System.currentTimeMillis();
long transactionId = hub.nextUniqueTransaction(startTime);
long tid = hub.nextUniqueTransaction(startTime);
Set<? extends Map.Entry<? extends K, ? extends V>> entries = map.entrySet();
Iterator<? extends Map.Entry<? extends K, ? extends V>> iterator = entries.iterator();
@@ -301,7 +302,7 @@ class ClientWiredStatelessChronicleMap<K, V>
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<K, V>
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<K, V>
// 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<K, V>
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<K, V>
// 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<K, V>
final Map<K, V> result = new HashMap<K, V>();
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<K, V>
// 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<K, V>
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<K, V>
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<K, V>
// 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<K, V>
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<K, V>
}
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<K, V>
}
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<K, V>
}
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<K, V>
// 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<K, V>
// 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<K, V>
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<K, V>
hub.outBytesLock().unlock();
}
return readBoolean(transactionId, startTime);
return readBoolean(tid, startTime);
}
@@ -694,10 +699,10 @@ class ClientWiredStatelessChronicleMap<K, V>
@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<K, V>
hub.outBytesLock().unlock();
}
return readBoolean(transactionId, startTime);
return readBoolean(tid, startTime);
}
@@ -728,36 +733,36 @@ class ClientWiredStatelessChronicleMap<K, V>
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<K, V>
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<K, V>
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<K, V>
@NotNull final EventId methodName, K key, V value, Class<V> 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<K, V>
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<K, V>
Class<R> 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<K, V>
}
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();
}
src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java
+49 -42
View File
@@ -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) {
src/main/java/net/openhft/chronicle/map/MapWireHandler.java
+18 -18
View File
@@ -121,14 +121,14 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
}
/**
* @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<BytesChronicleMap, Iterator> function,
long tid, @NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator> c, long maxEntries) throws StreamCorruptedException {
final BytesChronicleMap m = bytesMap(channelId);
@@ -139,7 +139,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
@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<K, V> implements WireHandler, Consumer<WireHandlers> {
}
private void writeChunked(
long transactionId, @NotNull final Function<BytesChronicleMap, Iterator> function,
long tid, @NotNull final Function<BytesChronicleMap, Iterator> function,
@NotNull final Consumer<Iterator> c) throws StreamCorruptedException {
writeChunked(transactionId, function, c, Long.MAX_VALUE);
writeChunked(tid, function, c, Long.MAX_VALUE);
}
@@ -200,7 +200,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
// 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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
}
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<K, V> implements WireHandler, Consumer<WireHandlers> {
}
}
private void putAll(long transactionId) {
private void putAll(long tid) {
final BytesChronicleMap bytesMap = bytesMap(MapWireHandler.this.channelId);
@@ -469,7 +469,7 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
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<K, V> implements WireHandler, Consumer<WireHandlers> {
}
};
incompleteWork.put(transactionId, runnable);
incompleteWork.put(tid, runnable);
runnable.run();
}
src/main/java/net/openhft/chronicle/map/MapWireHandlerBuilder.java
+1 -1
View File
@@ -45,7 +45,7 @@ public class MapWireHandlerBuilder {
channelId,
methodName,
type,
transactionId,
tid, // transactionId
result,
resultKey,
resultValue,