Тузиков Александр, ИТМО DWS, Stage 6

src/main/java/ru/vk/itmo/test/tuzikovalexandr/ConsistentHashing.java

@@ -3,19 +3,19 @@
 import one.nio.util.Hash;
 
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
-import java.util.Map;
 import java.util.NavigableMap;
 import java.util.SortedMap;
 import java.util.TreeMap;
-import java.util.stream.Collectors;
 
 public class ConsistentHashing {
     private final NavigableMap<Integer, String> circle;
+    private final List<String> clusterUrls;
 
     public ConsistentHashing(List<String> clusterUrls, int numbOfVirtualNodes) {
         circle = new TreeMap<>();
-
+        this.clusterUrls = clusterUrls;
         for (String clusterUrl : clusterUrls) {
             for (int i = 0; i < numbOfVirtualNodes; i++) {
                 addNode(i, clusterUrl);
@@ -39,28 +39,45 @@ public String getNode(String key) {
     }
 
     public List<String> getNodes(String key, int from) {
-        if (circle.isEmpty()) {
-            return new ArrayList<>();
+        List<String> res = new ArrayList<>();
+
+        if (key != null && from > 0) {
+            if (from < clusterUrls.size()) {
+                int slot = getHash(key);
+                Iterator<String> it = new ClockwiseIterator(slot);
+                while (it.hasNext() && res.size() < from) {
+                    String part = it.next();
+                    res.add(part);
+                }
+            } else {
+                res.addAll(clusterUrls);
+            }
         }
 
-        final int hash = getHash(key);
-        SortedMap<Integer, String> tailMap = circle.tailMap(hash);
-        return (tailMap.isEmpty() ? circle : tailMap).values()
-                .stream().limit(from).collect(Collectors.toList());
+        return res;
     }
 
-    List<String> getNodes(String key, List<String> clusterUrls, int from) {
-        Map<Integer, String> nodesHashes = new TreeMap<>();
+    private int getHash(String key) {
+        return Hash.murmur3(key);
+    }
+
+    private class ClockwiseIterator implements Iterator<String> {
+        private final Iterator<String> head;
+        private final Iterator<String> tail;
 
-        for (String nodeUrl : clusterUrls) {
-            nodesHashes.put(Hash.murmur3(nodeUrl + key), nodeUrl);
+        public ClockwiseIterator(int slot) {
+            this.head = circle.headMap(slot).values().iterator();
+            this.tail = circle.tailMap(slot).values().iterator();
         }
 
-        return nodesHashes.values().stream().limit(from)
-                .collect(Collectors.toList());
-    }
+        @Override
+        public boolean hasNext() {
+            return head.hasNext() || tail.hasNext();
+        }
 
-    private int getHash(String key) {
-        return Hash.murmur3(key);
+        @Override
+        public String next() {
+            return tail.hasNext() ? tail.next() : head.next();
+        }
     }
 }

src/main/java/ru/vk/itmo/test/tuzikovalexandr/Constants.java

@@ -4,6 +4,7 @@
 import one.nio.http.Response;
 
 import java.net.HttpURLConnection;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.Set;
 
@@ -13,7 +14,8 @@ public final class Constants {
     );
     public static final String TOO_MANY_REQUESTS = "429 Too Many Requests";
     public static final String NOT_ENOUGH_REPLICAS = "504 Not Enough Replicas";
-    public static final int REQUEST_TIMEOUT = 300;
+    public static final int SERVER_ERROR = 500;
+    public static final int REQUEST_TIMEOUT = 500;
     public static final Map<Integer, String> HTTP_CODE = Map.of(
             HttpURLConnection.HTTP_OK, Response.OK,
             HttpURLConnection.HTTP_ACCEPTED, Response.ACCEPTED,
@@ -25,10 +27,22 @@ public final class Constants {
     public static final String BASE_URL = "http://localhost";
     public static final int BASE_PORT = 8080;
     public static final long FLUSHING_THRESHOLD_BYTES = 1024 * 1024;
-    public static final int NUMBER_OF_VIRTUAL_NODES = 5;
+    public static final int NUMBER_OF_VIRTUAL_NODES = 100;
     public static final String HTTP_TIMESTAMP_HEADER = "X-Timestamp";
     public static final String NIO_TIMESTAMP_HEADER = "x-timestamp:";
     public static final String HTTP_TERMINATION_HEADER = "X-Termination";
+    public static final String RANGE_REQUEST = "/v0/entities?start=";
+    public static final String ID_REQUEST = "/v0/entity?id=";
+    public static final byte[] CRLF = "\r\n".getBytes(StandardCharsets.UTF_8);
+    public static final byte[] LAST_STRING = "0\r\n\r\n".getBytes(StandardCharsets.UTF_8);
+    public static final byte[] NEW_LINE = "\n".getBytes(StandardCharsets.UTF_8);
+    public static final byte[] HEADER =
+            """
+                    HTTP/1.1 200 OK\r
+                    Content-Type: text/plain\r
+                    Transfer-Encoding: chunked\r
+                    \r
+                    """.getBytes(StandardCharsets.UTF_8);
 
     private Constants() {
     }

src/main/java/ru/vk/itmo/test/tuzikovalexandr/RequestHandler.java

@@ -10,6 +10,7 @@
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.ValueLayout;
 import java.nio.charset.StandardCharsets;
+import java.util.Iterator;
 
 public class RequestHandler {
     private final Dao dao;
@@ -84,4 +85,11 @@ private Response sendResponseWithTimestamp(String resultCode, byte[] body, long
     private MemorySegment fromString(String data) {
         return data == null ? null : MemorySegment.ofArray(data.getBytes(StandardCharsets.UTF_8));
     }
+
+    public Iterator<EntryWithTimestamp<MemorySegment>> getEntries(String start, String end) {
+        MemorySegment startVal = fromString(start);
+        MemorySegment endVal = end != null ? fromString(end) : null;
+
+        return dao.get(startVal, endVal);
+    }
 }