CE-14 fixed a bug in the restricted entry set

This commit is contained in:
Rob Austin
2015-03-16 15:26:36 +00:00
parent e1fc723ae4
commit 93b8a969d0
3 changed files with 32 additions and 11 deletions
@@ -478,11 +478,14 @@ class ClientWiredStatelessChronicleMap<K, V>
if (wireIn.bytes().remaining() > 0) {
// todo process the exception
boolean isException = wireIn.read(Fields.isException).bool();
if (isException)
throw new RuntimeException(wireIn.read(Fields.result).text());
}
}
} catch (Exception e) {
throw e;
} finally {
hub.inBytesLock().unlock();
}
@@ -107,12 +107,12 @@ public class ClientWiredStatelessTcpConnectionHub {
try {
SocketChannel socketChannel = AbstractChannelReplicator.openSocketChannel(closeables);
if (socketChannel.connect(remoteAddress)) {
if (socketChannel.connect(remoteAddress))
clientChannel = socketChannel;
}
} catch (IOException e) {
LOG.error("", e);
if (closeables != null) closeables.closeQuietly();
clientChannel = null;
}
@@ -457,12 +457,23 @@ public class ClientWiredStatelessTcpConnectionHub {
private void readSocket(int requiredNumberOfBytes, long timeoutTime) throws IOException {
assert inBytesLock().isHeldByCurrentThread();
ByteBuffer buffer = inWireByteBuffer();
buffer.limit(buffer.position() + requiredNumberOfBytes);
ByteBuffer buffer = inWireByteBuffer();
int position = buffer.position();
try {
buffer.limit(position + requiredNumberOfBytes);
} catch (IllegalArgumentException e) {
buffer = inWireByteBuffer(position + requiredNumberOfBytes);
buffer.limit(position + requiredNumberOfBytes);
buffer.position(position);
}
long start = buffer.position();
while (buffer.position() - start < requiredNumberOfBytes) {
assert clientChannel != null;
int len = clientChannel.read(buffer);
@@ -477,6 +488,11 @@ public class ClientWiredStatelessTcpConnectionHub {
return (ByteBuffer) intWire.bytes().underlyingObject();
}
private ByteBuffer inWireByteBuffer(long requiredCapacity) {
intWire.bytes().ensureCapacity(requiredCapacity);
return (ByteBuffer) intWire.bytes().underlyingObject();
}
/**
* writes the bytes to the socket
*
@@ -505,7 +521,7 @@ public class ClientWiredStatelessTcpConnectionHub {
if (EventGroup.IS_DEBUG ) {
System.out.println("\n--------------------------------------------\n" +
"client wrote:\n\n" + Bytes.toDebugString(outBuffer));
"client writes:\n\n" + Bytes.toDebugString(outBuffer));
}
outBuffer.position(0);
@@ -139,9 +139,11 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
outWire.write(Fields.transactionId).int64(transactionId);
// this allows us to write more data than the buffer will allow
for (int count = 0; count < maxEntries; count++) {
for (int count = 0; ; count++) {
final boolean hasNext = iterator.hasNext();
boolean finished = count == maxEntries;
final boolean hasNext = iterator.hasNext() && !finished;
write(map -> {
@@ -401,10 +403,10 @@ class MapWireHandler<K, V> implements WireHandler, Consumer<WireHandlers> {
long len = outWire.bytes().position() - SIZE_OF_SIZE;
if (len == 0) {
System.out.println("--------------------------------------------\n" +
"server wrote:\n\n<EMPTY>");
"server writes:\n\n<EMPTY>");
} else {
System.out.println("--------------------------------------------\n" +
"server wrote:\n\n" +
"server writes:\n\n" +
Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len));
}
}