diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 3d6bdeff0..63a0b8470 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -546,7 +546,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { assert buffer.capacity() == bytes.capacity(); assert bytes.limit() == bytes.capacity(); - } @@ -561,21 +560,38 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { @NotNull public synchronized Collection values() { - long sizeLocation = writeEvent(VALUES); + final long sizeLocation = writeEvent(VALUES); + + final long startTime = System.currentTimeMillis(); + final long transactionId = nextUniqueTransaction(startTime); + final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs(); // get the data back from the server - final Bytes in = blockingFetch(sizeLocation); + Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); - final long size = in.readStopBit(); + final Collection result = new ArrayList(); - if (size > Integer.MAX_VALUE) - throw new IllegalStateException("size=" + size + " is too large."); + for (; ; ) { - final ArrayList result = new ArrayList((int) size); + boolean hasMoreEntries = in.readBoolean(); - for (int i = 0; i < size; i++) { - result.add(keyValueSerializer.readValue(in)); + // number of entries in the chunk + long size = in.readInt(); + + for (int i = 0; i < size; i++) { + V v = keyValueSerializer.readValue(in); + result.add(v); + } + + if (!hasMoreEntries) + break; + + compact(in); + + in = blockingFetchReadOnly(timeoutTime, transactionId); } + + return result; } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 2aacb2084..aa664510b 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -1561,18 +1561,18 @@ class StatelessServerConnector { private Work values(Bytes reader, Bytes writer) { - final long sizeLocation = reflectTransactionId(reader, writer); - final Collection values; + final long transactionId = reader.readLong(); + + Collection ks; try { - values = map.values(); + ks = map.values(); } catch (RuntimeException e) { - return sendException(writer, sizeLocation, e); + return sendException(reader, writer, e); } - writer.writeStopBit(values.size()); - final Iterator iterator = values.iterator(); + final Iterator iterator = ks.iterator(); // this allows us to write more data than the buffer will allow return new Work() { @@ -1580,17 +1580,24 @@ class StatelessServerConnector { @Override public boolean doWork(Bytes writer) { + final long sizeLocation = header(writer, transactionId); + + int count = 0; while (iterator.hasNext()) { // we've filled up the buffer, so lets give another channel a chance to send // some data, we don't know the max key size, we will use the entrySize instead - if (writer.remaining() <= maxEntrySizeBytes) + if (writer.remaining() <= maxEntrySizeBytes) { + writeHeader(writer, sizeLocation, count, true); return false; + } + + count++; writeValue(iterator.next(), writer); } - writeSizeAndFlags(sizeLocation, false, writer); + writeHeader(writer, sizeLocation, count, false); return true; } }; @@ -1601,7 +1608,6 @@ class StatelessServerConnector { Set ks; - try { ks = map.keySet(); } catch (RuntimeException e) { diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index effba2381..43fcecc83 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -23,6 +23,7 @@ import org.testng.Assert; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.Collection; import java.util.HashMap; import java.util.Map; import java.util.Set; @@ -82,6 +83,47 @@ public class StatelessClientTest { } + @Test + public void testBufferOverFlowPutAllAndValues() throws IOException, InterruptedException { + + final ChronicleMap serverMap; + final ChronicleMap statelessMap; + + + // stateless client + { + statelessMap = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .stateless(remoteAddress(new InetSocketAddress("localhost", 8076))).create(); + } + + // server + { + serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replicators((byte) 2, TcpReplicationConfig.of(8076)).create(); + } + + + Map payload = new HashMap(); + + for (int i = 0; i < SIZE; i++) { + payload.put(i, "some value=" + i); + } + + + statelessMap.putAll(payload); + + + Collection values = statelessMap.values(); + + + Assert.assertEquals(values.size(), SIZE); + + serverMap.close(); + statelessMap.close(); + } + + @Test public void testBufferOverFlowPutAllAndKeySet() throws IOException, InterruptedException {