From 93b8a969d075b991bfd7d7943a3569f75dffd3e3 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Mon, 16 Mar 2015 15:26:36 +0000 Subject: [PATCH] CE-14 fixed a bug in the restricted entry set --- .../map/ClientWiredStatelessChronicleMap.java | 5 +++- .../ClientWiredStatelessTcpConnectionHub.java | 28 +++++++++++++++---- .../openhft/chronicle/map/MapWireHandler.java | 10 ++++--- 3 files changed, 32 insertions(+), 11 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java index 979ebff9c..b8ee06ded 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessChronicleMap.java @@ -478,11 +478,14 @@ class ClientWiredStatelessChronicleMap 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(); } diff --git a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java index 0fa15984f..4d9072b36 100644 --- a/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java +++ b/src/main/java/net/openhft/chronicle/map/ClientWiredStatelessTcpConnectionHub.java @@ -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); diff --git a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java index 48b31d924..6a3007d7a 100644 --- a/src/main/java/net/openhft/chronicle/map/MapWireHandler.java +++ b/src/main/java/net/openhft/chronicle/map/MapWireHandler.java @@ -139,9 +139,11 @@ class MapWireHandler implements WireHandler, Consumer { 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 implements WireHandler, Consumer { long len = outWire.bytes().position() - SIZE_OF_SIZE; if (len == 0) { System.out.println("--------------------------------------------\n" + - "server wrote:\n\n"); + "server writes:\n\n"); } else { System.out.println("--------------------------------------------\n" + - "server wrote:\n\n" + + "server writes:\n\n" + Bytes.toDebugString(outWire.bytes(), SIZE_OF_SIZE, len)); } }