Андрей Смирнов ИТМО КТ stage6

src/main/java/ru/vk/itmo/test/smirnovandrew/Main.java

@@ -16,7 +16,7 @@ public static void main(String[] args) throws IOException {
         ReferenceDao dao = new ReferenceDao(
                 new Config(
                         data,
-                        2 * 1024 * 1024
+                        2L * 1024 * 1024
                 )
         );
         String localhost = "http://localhost";

src/main/java/ru/vk/itmo/test/smirnovandrew/MyServer.java

@@ -1,44 +1,39 @@
 package ru.vk.itmo.test.smirnovandrew;
 
 import one.nio.http.Header;
-import one.nio.http.HttpClient;
-import one.nio.http.HttpException;
 import one.nio.http.HttpServer;
 import one.nio.http.HttpSession;
 import one.nio.http.Param;
 import one.nio.http.Path;
 import one.nio.http.Request;
 import one.nio.http.RequestMethod;
 import one.nio.http.Response;
-import one.nio.pool.PoolException;
 import ru.vk.itmo.ServiceConfig;
 import ru.vk.itmo.test.reference.dao.ReferenceDao;
 
 import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.function.Function;
 import java.util.logging.Logger;
-import java.util.stream.Collectors;
 
 public class MyServer extends HttpServer {
-    private static final String ROOT = "/v0/entity";
-    private static final String X_SENDER_NODE = "X-SenderNode";
-    private static final String NOT_ENOUGH_REPLICAS = "504 Not Enough Replicas";
-    private static final long DURATION = 1000L;
-    private static final int OK_STATUS = 300;
-    private static final int NOT_FOUND_STATUS = 404;
-    private static final String HEADER_DELIMITER = ": ";
     private final MyServerDao dao;
     private final MyExecutor executor;
     private final Logger logger;
-    private final Map<String, HttpClient> httpClients;
+    private final HttpClient httpClient;
     private final RendezvousClusterManager rendezvousClustersManager;
     private final ServiceConfig config;
 
@@ -64,17 +59,20 @@ public MyServer(
         this.dao = new MyServerDao(dao);
         this.executor = new MyExecutor(corePoolSize, availableProcessors);
         this.logger = Logger.getLogger(MyServer.class.getName());
-        this.httpClients = config.clusterUrls().stream()
-                .filter(url -> !Objects.equals(url, config.selfUrl()))
-                .collect(Collectors.toMap(s -> s, MyServerUtil::createClient, (c, c1) -> c));
+        this.httpClient = HttpClient.newHttpClient();
     }
 
     @Override
     public void handleRequest(Request request, HttpSession session) throws IOException {
         try {
-            long exp = System.currentTimeMillis() + DURATION;
+            long exp = System.currentTimeMillis() + MyServerUtil.DURATION;
             executor.execute(() -> {
                 try {
+                    if (Objects.equals(request.getPath(), MyServerUtil.ROOT_ENTITIES)) {
+                        entities(request, session);
+                        return;
+                    }
+
                     if (System.currentTimeMillis() > exp) {
                         MyServerUtil.sendEmpty(session, logger, Response.SERVICE_UNAVAILABLE);
                     } else {
@@ -98,27 +96,31 @@ private static int quorum(int from) {
         return from / 2 + 1;
     }
 
-    private Response sendToAnotherNode(
+    private HttpRequest toHttpRequest(Request request, String nodeUrl, String params) {
+        return HttpRequest.newBuilder(URI.create(nodeUrl + MyServerUtil.ROOT + "?" + params))
+                .method(request.getMethodName(), request.getBody() == null
+                        ? HttpRequest.BodyPublishers.noBody()
+                        : HttpRequest.BodyPublishers.ofByteArray(request.getBody()))
+                .setHeader(MyServerUtil.X_SENDER_NODE, config.selfUrl())
+                .build();
+    }
+
+    private CompletableFuture<Response> sendToAnotherNode(
             Request request,
+            int ack,
+            int from,
+            String id,
             String clusterUrl,
             Function<MyServerDao, Response> operation
     ) {
         if (Objects.equals(clusterUrl, config.selfUrl())) {
-            return operation.apply(dao);
+            return CompletableFuture.completedFuture(operation.apply(dao));
         }
 
-        var httpClient = httpClients.get(clusterUrl);
+        var httpRequest = toHttpRequest(request, clusterUrl, String.format("id=%s&from=%d&ack=%d", id, from, ack));
 
-        try {
-            return httpClient.invoke(request);
-        } catch (InterruptedException e) {
-            logger.info(e.getMessage());
-            Thread.currentThread().interrupt();
-            return new Response(Response.INTERNAL_ERROR, Response.EMPTY);
-        } catch (HttpException | IOException | PoolException e1) {
-            logger.info(e1.getMessage());
-            return new Response(Response.INTERNAL_ERROR, Response.EMPTY);
-        }
+        return httpClient.sendAsync(httpRequest, HttpResponse.BodyHandlers.ofByteArray())
+                .thenApplyAsync(MyServerUtil::processingResponse);
     }
 
     private Response handleLocalRequest(
@@ -128,18 +130,22 @@ private Response handleLocalRequest(
             Integer ackParam,
             String senderNode,
             Function<MyServerDao, Response> operation
-    ) {
-        Integer from = fromParam;
-        if (Objects.isNull(from)) {
+    ) throws ExecutionException, InterruptedException, TimeoutException {
+        final int from;
+        if (Objects.isNull(fromParam)) {
             from = config.clusterUrls().size();
+        } else {
+            from = fromParam;
         }
 
-        Integer ack = ackParam;
-        if (Objects.isNull(ack)) {
+        final int ack;
+        if (Objects.isNull(ackParam)) {
             ack = quorum(from);
+        } else {
+            ack = ackParam;
         }
 
-        String paramError = getParametersError(id, from, ack);
+        String paramError = MyServerUtil.getParametersError(id, from, ack, config.clusterUrls().size());
         if (Objects.nonNull(paramError)) {
             return new Response(Response.BAD_REQUEST, paramError.getBytes(StandardCharsets.UTF_8));
         }
@@ -154,62 +160,40 @@ private Response handleLocalRequest(
             return new Response(Response.BAD_REQUEST, Response.EMPTY);
         }
 
-        var sortedNodes = RendezvousClusterManager.getSortedNodes(from, config);
+        var sortedNodes = RendezvousClusterManager.getSortedNodes(id, from, config);
 
         if (sortedNodes.stream().map(config.clusterUrls()::get).noneMatch(config.selfUrl()::equals)) {
-            return sendToAnotherNode(request, clusterUrl, operation);
-        }
-
-        request.addHeader(String.join(HEADER_DELIMITER, X_SENDER_NODE, config.selfUrl()));
-        var responses = new ArrayList<Response>();
-        for (int nodeNumber : sortedNodes) {
-            var r = sendToAnotherNode(request, config.clusterUrls().get(nodeNumber), operation);
-            if (r.getStatus() < OK_STATUS
-                    || (r.getStatus() == NOT_FOUND_STATUS && request.getMethod() == Request.METHOD_GET)) {
-                responses.add(r);
-            }
-        }
-
-        if (responses.size() < ack) {
-            return new Response(NOT_ENOUGH_REPLICAS, Response.EMPTY);
-        }
-        return MyServerUtil.getMaxTimestampResponse(responses);
-    }
-
-    private String getParametersError(String id, Integer from, Integer ack) {
-        if (Objects.isNull(id) || id.isEmpty()) {
-            return "Invalid id provided";
-        }
-
-        if (ack <= 0) {
-            return "Too small ack";
+            return sendToAnotherNode(request, ack, from, id, clusterUrl, operation)
+                    .get(MyServerUtil.DURATION, TimeUnit.MILLISECONDS);
         }
 
-        if (from <= 0) {
-            return "Too small from";
-        }
-
-        int clusterSize = config.clusterUrls().size();
-        if (from > clusterSize) {
-            return String.format("From is greater than cluster size: from=%d, clusterSize=%d", from, clusterSize);
-        }
-
-        if (ack > from) {
-            return String.format("Ack is greater than from: ack=%d, from=%d", ack, from);
-        }
-
-        return null;
+        var completableResults = sortedNodes.stream()
+                .map(nodeNumber -> sendToAnotherNode(
+                        request,
+                        ack,
+                        from,
+                        id,
+                        config.clusterUrls().get(nodeNumber),
+                        operation))
+                .toList();
+
+        return MyServerUtil.getResults(
+                from,
+                ack,
+                completableResults,
+                logger
+        );
     }
 
-    @Path(ROOT)
+    @Path(MyServerUtil.ROOT)
     @RequestMethod(Request.METHOD_GET)
     public Response get(
             @Param(value = "id", required = true) String id,
             @Param(value = "from") Integer from,
             @Param(value = "ack") Integer ack,
-            @Header(value = X_SENDER_NODE) String senderNode,
+            @Header(value = MyServerUtil.X_SENDER_NODE) String senderNode,
             Request request
-    ) {
+    ) throws ExecutionException, InterruptedException, TimeoutException {
         return handleLocalRequest(
                 request,
                 id,
@@ -220,15 +204,15 @@ public Response get(
         );
     }
 
-    @Path(ROOT)
+    @Path(MyServerUtil.ROOT)
     @RequestMethod(Request.METHOD_DELETE)
     public Response delete(
             @Param(value = "id", required = true) String id,
             @Param(value = "from") Integer from,
             @Param(value = "ack") Integer ack,
-            @Header(value = X_SENDER_NODE) String senderNode,
+            @Header(value = MyServerUtil.X_SENDER_NODE) String senderNode,
             Request request
-    ) {
+    ) throws ExecutionException, InterruptedException, TimeoutException {
         return handleLocalRequest(
                 request,
                 id,
@@ -239,15 +223,15 @@ public Response delete(
         );
     }
 
-    @Path(ROOT)
+    @Path(MyServerUtil.ROOT)
     @RequestMethod(Request.METHOD_PUT)
     public Response put(
             @Param(value = "id", required = true) String id,
             @Param(value = "from") Integer from,
             @Param(value = "ack") Integer ack,
-            @Header(value = X_SENDER_NODE) String senderNode,
+            @Header(value = MyServerUtil.X_SENDER_NODE) String senderNode,
             Request request
-    ) {
+    ) throws ExecutionException, InterruptedException, TimeoutException {
         request.addHeader("Content-Length: " + request.getBody().length);
         request.setBody(request.getBody());
 
@@ -261,6 +245,20 @@ public Response put(
         );
     }
 
+    private void entities(
+            Request request,
+            HttpSession session
+    ) {
+        String start = request.getParameter("start=");
+        String end = request.getParameter("end=");
+        if (start.isEmpty()) {
+            MyServerUtil.sendEmpty(session, logger, Response.BAD_REQUEST);
+            return;
+        }
+
+        MyServerUtil.handleLocalEntitiesRequest(start, end, session, dao, logger);
+    }
+
     @Override
     public void handleDefault(Request request, HttpSession session) throws IOException {
         session.sendResponse(
@@ -275,6 +273,6 @@ public void handleDefault(Request request, HttpSession session) throws IOExcepti
     public synchronized void stop() {
         this.executor.shutdown();
         super.stop();
-        httpClients.values().forEach(HttpClient::close);
+        httpClient.close();
     }
 }

src/main/java/ru/vk/itmo/test/smirnovandrew/MyServerDao.java

@@ -14,6 +14,7 @@
 import java.io.ObjectOutputStream;
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.ValueLayout;
+import java.util.Iterator;
 
 public class MyServerDao {
     private final Dao<MemorySegment, Entry<MemorySegment>> dao;
@@ -22,6 +23,10 @@ public MyServerDao(Dao<MemorySegment, Entry<MemorySegment>> dao) {
         this.dao = dao;
     }
 
+    Iterator<Entry<MemorySegment>> getEntriesFromDao(String start, String end) {
+        return dao.get(DaoFactory.fromString(start), DaoFactory.fromString(end));
+    }
+
     Response getEntryFromDao(String id) {
         Entry<MemorySegment> entry = dao.get(DaoFactory.fromString(id));
         if (entry == null) {
@@ -31,11 +36,11 @@ Response getEntryFromDao(String id) {
             ValWithTime valueWithTimestamp = byteArrayToObject(entry.value().toArray(ValueLayout.JAVA_BYTE));
             if (valueWithTimestamp.value() == null) {
                 Response response = new Response(Response.NOT_FOUND, Response.EMPTY);
-                response.addHeader(MyServerUtil.X_TIMESTAMP + valueWithTimestamp.timestamp());
+                response.addHeader("X-Timestamp: " + valueWithTimestamp.timestamp());
                 return response;
             }
             Response response = new Response(Response.OK, valueWithTimestamp.value());
-            response.addHeader(MyServerUtil.X_TIMESTAMP + valueWithTimestamp.timestamp());
+            response.addHeader("X-Timestamp: " + valueWithTimestamp.timestamp());
             return response;
         } catch (IOException | ClassNotFoundException e) {
             return new Response(Response.INTERNAL_ERROR, Response.EMPTY);
@@ -64,7 +69,7 @@ Response deleteValueFromDao(String id) {
         return new Response(Response.ACCEPTED, Response.EMPTY);
     }
 
-    private static ValWithTime byteArrayToObject(byte[] bytes) throws IOException, ClassNotFoundException {
+    static ValWithTime byteArrayToObject(byte[] bytes) throws IOException, ClassNotFoundException {
         var byteArrayInputStream = new ByteArrayInputStream(bytes);
         try (var objectInputStream = new ObjectInputStream(byteArrayInputStream)) {
             return (ValWithTime) objectInputStream.readObject();