Осокин Дмитрий Hw6 Политех

src/main/java/ru/vk/itmo/test/osokindm/ChunkedResponse.java

@@ -0,0 +1,26 @@
+package ru.vk.itmo.test.osokindm;
+
+import one.nio.http.Response;
+import ru.vk.itmo.test.osokindm.dao.Entry;
+
+import java.lang.foreign.MemorySegment;
+import java.util.Iterator;
+
+public class ChunkedResponse extends Response {
+
+    private static final String CHUNKED_HEADER = "Transfer-Encoding: chunked";
+    private static final String CONTENT_TYPE_HEADER = "Content-Type: result-range";
+    private final Iterator<Entry<MemorySegment>> rangeResult;
+
+    public ChunkedResponse(String resultCode, Iterator<Entry<MemorySegment>> rangeResult) {
+        super(resultCode);
+        super.addHeader(CHUNKED_HEADER);
+        super.addHeader(CONTENT_TYPE_HEADER);
+        this.rangeResult = rangeResult;
+    }
+
+    public Iterator<Entry<MemorySegment>> getResultIterator() {
+        return rangeResult;
+    }
+
+ }

src/main/java/ru/vk/itmo/test/osokindm/CustomHttpSession.java

@@ -0,0 +1,59 @@
+package ru.vk.itmo.test.osokindm;
+
+import one.nio.http.HttpServer;
+import one.nio.http.HttpSession;
+import one.nio.http.Response;
+import one.nio.net.Socket;
+import ru.vk.itmo.test.osokindm.dao.Entry;
+
+import java.io.IOException;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Iterator;
+
+public class CustomHttpSession extends HttpSession {
+
+    private static final byte[] CHUNK_SEPARATOR = "\r\n".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] DELIMITER = "\n".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] ZERO = "0".getBytes(StandardCharsets.UTF_8);
+
+    public CustomHttpSession(Socket socket, HttpServer server) {
+        super(socket, server);
+    }
+
+    @Override
+    protected void writeResponse(Response response, boolean includeBody) throws IOException {
+        if (response instanceof ChunkedResponse) {
+            super.writeResponse(response, false);
+            Iterator<Entry<MemorySegment>> iterator = ((ChunkedResponse) response).getResultIterator();
+            while (iterator.hasNext()) {
+                writeChunk(iterator.next());
+            }
+
+            super.write(ZERO, 0, ZERO.length);
+            super.write(CHUNK_SEPARATOR, 0, CHUNK_SEPARATOR.length);
+            super.write(CHUNK_SEPARATOR, 0, CHUNK_SEPARATOR.length);
+            super.close();
+        } else {
+            super.writeResponse(response, includeBody);
+        }
+    }
+
+    private void writeChunk(Entry<MemorySegment> entry) throws IOException {
+        byte[] keyBytes = entry.key().toArray(ValueLayout.JAVA_BYTE);
+        byte[] valueBytes = entry.value().toArray(ValueLayout.JAVA_BYTE);
+        int entryLength = keyBytes.length + valueBytes.length + DELIMITER.length;
+        ByteBuffer buffer = ByteBuffer.allocate(Integer.BYTES + entryLength + CHUNK_SEPARATOR.length * 2);
+        buffer.put(Integer.toHexString(entryLength).getBytes(StandardCharsets.UTF_8));
+        buffer.put(CHUNK_SEPARATOR);
+        buffer.put(entry.key().toArray(ValueLayout.JAVA_BYTE));
+        buffer.put(DELIMITER);
+        buffer.put(entry.value().toArray(ValueLayout.JAVA_BYTE));
+        buffer.put(CHUNK_SEPARATOR);
+
+        super.write(buffer.array(), 0, buffer.position());
+    }
+
+}

src/main/java/ru/vk/itmo/test/osokindm/DaoWrapper.java

@@ -1,14 +1,15 @@
 package ru.vk.itmo.test.osokindm;
 
 import one.nio.http.Request;
-import ru.vk.itmo.dao.BaseEntry;
 import ru.vk.itmo.dao.Config;
-import ru.vk.itmo.dao.Entry;
+import ru.vk.itmo.test.osokindm.dao.BaseEntry;
+import ru.vk.itmo.test.osokindm.dao.Entry;
 import ru.vk.itmo.test.osokindm.dao.ReferenceDao;
 
 import java.io.IOException;
 import java.lang.foreign.MemorySegment;
 import java.nio.charset.StandardCharsets;
+import java.util.Iterator;
 
 public class DaoWrapper {
 
@@ -23,15 +24,21 @@ public Entry<MemorySegment> get(String id) {
         return storage.get(key);
     }
 
-    public void delete(String id) {
+    public Iterator<Entry<MemorySegment>> get(String start, String end) {
+        MemorySegment startKey = getMemorySegment(start);
+        MemorySegment endKey = getMemorySegment(end);
+        return storage.get(startKey, endKey);
+    }
+
+    public void delete(String id, long timestamp) {
         MemorySegment key = getMemorySegment(id);
-        storage.upsert(new BaseEntry<>(key, null));
+        storage.upsert(new BaseEntry<>(key, null, timestamp));
     }
 
-    public void upsert(String id, Request request) {
+    public void upsert(String id, Request request, long timestamp) {
         MemorySegment key = getMemorySegment(id);
         MemorySegment value = getMemorySegment(request.getBody());
-        storage.upsert(new BaseEntry<>(key, value));
+        storage.upsert(new BaseEntry<>(key, value, timestamp));
     }
 
     public void stop() throws IOException {

src/main/java/ru/vk/itmo/test/osokindm/HttpServerImpl.java

@@ -5,6 +5,8 @@
 import one.nio.http.HttpSession;
 import one.nio.http.Request;
 import one.nio.http.Response;
+import one.nio.net.Socket;
+import one.nio.server.RejectedSessionException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -18,6 +20,7 @@
 public class HttpServerImpl extends HttpServer {
 
     private static final String ID_REQUEST = "id=";
+    private static final String START_REQUEST = "start=";
     private static final long KEEP_ALIVE_TIME = 60L;
     private static final Logger LOGGER = LoggerFactory.getLogger(HttpServerImpl.class);
     private final ThreadPoolExecutor requestWorkers;
@@ -34,6 +37,11 @@ public HttpServerImpl(HttpServerConfig config) throws IOException {
         requestWorkers.prestartAllCoreThreads();
     }
 
+    @Override
+    public HttpSession createSession(Socket socket) throws RejectedSessionException {
+        return new CustomHttpSession(socket, this);
+    }
+
     @Override
     public synchronized void stop() {
         super.stop();
@@ -52,7 +60,8 @@ public synchronized void stop() {
     @Override
     public void handleRequest(Request request, HttpSession session) {
         String id = request.getParameter(ID_REQUEST);
-        if (id == null || id.isEmpty()) {
+        String start = request.getParameter(START_REQUEST);
+        if ((id == null || id.isEmpty()) && (start == null || start.isEmpty())) {
             try {
                 handleDefault(request, session);
             } catch (IOException e) {

src/main/java/ru/vk/itmo/test/osokindm/Node.java

@@ -3,9 +3,9 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 public class Node {
+    private static final int MAX_ERRORS = 10;
     public final String address;
     public final int name;
-    private static final int MAX_ERRORS = 10;
     private final AtomicInteger errors;
     private volatile boolean isAlive;
 

src/main/java/ru/vk/itmo/test/osokindm/RendezvousRouter.java

@@ -3,7 +3,9 @@
 import one.nio.util.Hash;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.TreeMap;
 
 public class RendezvousRouter {
 
@@ -32,4 +34,25 @@ public Node getNode(String key) {
         return maxHashNode;
     }
 
+    @SuppressWarnings("MixedMutabilityReturnType")
+    public List<Node> getNodes(String key, int nodeAmount) {
+        if (key == null) {
+            return Collections.emptyList();
+        }
+        TreeMap<Integer, Node> sortedNodes = new TreeMap<>();
+        for (Node node : nodes) {
+            int hash = Hash.murmur3(node.name + key);
+            sortedNodes.put(hash, node);
+        }
+        List<Node> selectedNodes = new ArrayList<>();
+        int nodesNeeded = nodeAmount;
+        for (Node node : sortedNodes.values()) {
+            if (nodesNeeded == 0) break;
+            selectedNodes.add(node);
+            nodesNeeded--;
+        }
+
+        return selectedNodes;
+    }
+
 }