HCOLL-167 added support for keyset

src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java

@@ -570,14 +570,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
 
     }
 
-    @NotNull
-    public synchronized Set<K> keySet() {
-        long sizeLocation = writeEvent(KEY_SET);
-
-        // get the data back from the server
-        return readKeySet(blockingFetch(sizeLocation));
-    }
-
 
     @NotNull
     public synchronized Collection<V> values() {
@@ -637,6 +629,42 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
     }
 
 
+    @NotNull
+    public synchronized Set<K> 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<K> 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();

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<K, V> {
         return null;
     }
 
-    private Work keySet(Bytes reader, final Bytes writer) {
-        final long sizeLocation = reflectTransactionId(reader, writer);
-        final Set<K> ks;
-        try {
-            ks = map.keySet();
-        } catch (RuntimeException e) {
-            return sendException(writer, sizeLocation, e);
-        }
-
-        writer.writeStopBit(ks.size());
-
-        final Iterator<K> 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<K, V> {
         };
     }
 
+    private Work keySet(Bytes reader, final Bytes writer) {
+        final long transactionId = reader.readLong();
+
+        Set<K> ks;
+
+
+        try {
+            ks = map.keySet();
+        } catch (RuntimeException e) {
+            return sendException(reader, writer, e);
+        }
+
+
+        final Iterator<K> 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();