diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 37b6b2e93..80267adbf 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -53,6 +53,7 @@ abstract class AbstractChannelReplicator implements Closeable { public static final int SIZE_OF_SIZE = 4; public static final int SIZE_OF_TRANSACTION_ID = 8; + public static final int SIZE_OF_IS_EXCEPTION = 1; private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class); diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 2c88efaff..789169a58 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -982,8 +982,9 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon // block until we have received all the bytes in this chunk receive(size, timeoutTime); - final boolean isException = bytes.readBoolean(); + final long inTransactionId = bytes.readLong(); + final boolean isException = bytes.readBoolean(); if (inTransactionId != transactionId) { LOG.error("", new IllegalStateException("Skipped Message with transaction-id=" + diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index cd24a9757..9d78cb591 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1520,7 +1520,7 @@ class StatelessServerConnector { writeException(writer, e); - writeSizeAndFlags(sizeLocation, true, writer); + writeSizeAndFlags(sizeLocation+SIZE_OF_TRANSACTION_ID, true, writer); return null; } @@ -1845,17 +1845,19 @@ class StatelessServerConnector { } private long reflectTransactionId(Bytes reader, Bytes writer) { - final long sizeLocation = writer.position(); - writer.skip(TRANSACTION_ID_OFFSET); // isException + SIZE_OF_SIZE final long transactionId = reader.readLong(); + final long sizeLocation = writer.position(); + writer.skip(SIZE_OF_SIZE); writer.writeLong(transactionId); + writer.writeBoolean(false); // isException return sizeLocation; } private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { final long size = out.position() - locationOfSize; out.writeInt(locationOfSize, (int) size); // size - out.writeBoolean(locationOfSize + SIZE_OF_SIZE, isException); // isException + out.skip(SIZE_OF_TRANSACTION_ID); + out.writeBoolean(isException); // isException } private void writeException(Bytes out, Throwable e) { @@ -1887,10 +1889,11 @@ class StatelessServerConnector { private long header(Bytes writer, final long transactionId) { final long sizeLocation = writer.position(); - writer.skip(TRANSACTION_ID_OFFSET); + writer.skip(SIZE_OF_SIZE); + writer.writeLong(transactionId); // exception - writer.writeLong(transactionId); + writer.skip(1); // hasAnotherChunk writer.skip(1); @@ -1909,12 +1912,12 @@ class StatelessServerConnector { // size in bytes writer.writeInt(size); - // is exception - writer.writeBoolean(false); - //transaction id; writer.skip(8); + // is exception + writer.writeBoolean(false); + writer.writeBoolean(hasAnotherChunk); // count