diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 80399efc0..82e35f25b 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -570,14 +570,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - @NotNull - public synchronized Set keySet() { - long sizeLocation = writeEvent(KEY_SET); - - // get the data back from the server - return readKeySet(blockingFetch(sizeLocation)); - } - @NotNull public synchronized Collection values() { @@ -637,6 +629,42 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } + @NotNull + public synchronized Set keySet() { + final long sizeLocation = writeEvent(KEY_SET); + + final long startTime = System.currentTimeMillis(); + final long transactionId = nextUniqueTransaction(startTime); + final long timeoutTime = System.currentTimeMillis() + builder.timeoutMs(); + + // get the data back from the server + Bytes in = blockingFetch0(sizeLocation, transactionId, startTime); + + final Set result = new HashSet<>(); + + for (; ; ) { + + boolean hasMoreEntries = in.readBoolean(); + LOG.info("entrySet - hasMoreEntries=" + hasMoreEntries); + + // number of entries in the chunk + long size = in.readInt(); + + for (int i = 0; i < size; i++) { + K k = keyValueSerializer.readKey(in); + result.add(k); + } + + if (!hasMoreEntries) + break; + + in = blockingFetchReadOnly(timeoutTime, transactionId); + } + + return result; + } + + private long writeEvent(EventId event) { buffer.clear(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 18fe6b89b..0b7504b3f 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -548,7 +548,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { if (completed) attached.entryWriter.workCompleted(); - attached.hasRemoteHeartbeatInterval = false; + attached.hasRemoteHeartbeatInterval = false; } else if (attached.remoteModificationIterator != null) attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); @@ -1573,40 +1573,6 @@ class StatelessServerConnector { return null; } - private Work keySet(Bytes reader, final Bytes writer) { - final long sizeLocation = reflectTransactionId(reader, writer); - final Set ks; - try { - ks = map.keySet(); - } catch (RuntimeException e) { - return sendException(writer, sizeLocation, e); - } - - writer.writeStopBit(ks.size()); - - final Iterator iterator = ks.iterator(); - - // this allows us to write more data than the buffer will allow - return new Work() { - - @Override - public boolean doWork(@NotNull final Bytes writer) { - - 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) - return false; - - writeKey(iterator.next(), writer); - } - - writeSizeAndFlags(sizeLocation, false, writer); - return true; - } - }; - } private Work values(Bytes reader, Bytes writer) { final long sizeLocation = reflectTransactionId(reader, writer); @@ -1644,6 +1610,50 @@ class StatelessServerConnector { }; } + private Work keySet(Bytes reader, final Bytes writer) { + final long transactionId = reader.readLong(); + + Set ks; + + + try { + ks = map.keySet(); + } catch (RuntimeException e) { + return sendException(reader, writer, e); + } + + + final Iterator iterator = ks.iterator(); + + // this allows us to write more data than the buffer will allow + return new Work() { + + @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) { + writeHeader(writer, sizeLocation, count, true); + return false; + } + + count++; + K key = iterator.next(); + writeKey(key, writer); + } + + writeHeader(writer, sizeLocation, count, false); + return true; + } + }; + } + private Work entrySet(Bytes reader, Bytes writer) { final long transactionId = reader.readLong();