Плясов Климентий, ИТМО DWS, stage 6

src/main/java/ru/vk/itmo/test/klimplyasov/ChunkGenerator.java

@@ -0,0 +1,57 @@
+package ru.vk.itmo.test.klimplyasov;
+
+import one.nio.http.HttpSession;
+import ru.vk.itmo.dao.Entry;
+
+import java.io.IOException;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.charset.StandardCharsets;
+
+public final class ChunkGenerator {
+    private static final byte[] HTTP_RESPONSE_LINE = "HTTP/1.1 200 OK".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] CRLF = "\r\n".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] CONTENT_TYPE_HEADER = "Content-Type: text/plain".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] TRANSFER_ENCODING_HEADER =
+            "Transfer-Encoding: chunked".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] CONNECTION_HEADER = "Connection: keep-alive".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] LF = "\n".getBytes(StandardCharsets.UTF_8);
+
+    private ChunkGenerator() {
+        //:)
+    }
+
+    public static void writeResponseHeaders(HttpSession session) throws IOException {
+        session.write(HTTP_RESPONSE_LINE, 0, HTTP_RESPONSE_LINE.length);
+        session.write(CRLF, 0, CRLF.length);
+        session.write(CONTENT_TYPE_HEADER, 0, CONTENT_TYPE_HEADER.length);
+        session.write(CRLF, 0, CRLF.length);
+        session.write(TRANSFER_ENCODING_HEADER, 0, TRANSFER_ENCODING_HEADER.length);
+        session.write(CRLF, 0, CRLF.length);
+        session.write(CONNECTION_HEADER, 0, CONNECTION_HEADER.length);
+        session.write(CRLF, 0, CRLF.length);
+        session.write(CRLF, 0, CRLF.length);
+    }
+
+    public static void writeDataChunk(HttpSession session, Entry<MemorySegment> entry) throws IOException {
+        byte[] key = entry.key().toArray(ValueLayout.JAVA_BYTE);
+        byte[] value = entry.value().toArray(ValueLayout.JAVA_BYTE);
+        byte[] lengthBytes =
+                Integer
+                .toHexString(key.length + value.length + LF.length)
+                .getBytes(StandardCharsets.UTF_8);
+        session.write(lengthBytes, 0, lengthBytes.length);
+        session.write(CRLF, 0, CRLF.length);
+        session.write(key, 0, key.length);
+        session.write(LF, 0, LF.length);
+        session.write(value, 0, value.length);
+        session.write(CRLF, 0, CRLF.length);
+    }
+
+    public static void writeEmptyChunk(HttpSession session) throws IOException {
+        byte[] dataSizeBytes = Integer.toHexString(0).getBytes(StandardCharsets.UTF_8);
+        session.write(dataSizeBytes, 0, dataSizeBytes.length);
+        session.write(CRLF, 0, CRLF.length);
+        session.write(CRLF, 0, CRLF.length);
+    }
+}

src/main/java/ru/vk/itmo/test/klimplyasov/HandleResult.java

@@ -0,0 +1,14 @@
+package ru.vk.itmo.test.klimplyasov;
+
+public record HandleResult(int status, byte[] data, long timestamp) {
+
+    public HandleResult(int status, byte[] data, long timestamp) {
+        this.status = status;
+        this.data = data;
+        this.timestamp = timestamp;
+    }
+
+    public HandleResult(int status, byte[] data) {
+        this(status, data, 0);
+    }
+}

src/main/java/ru/vk/itmo/test/klimplyasov/MergeHandleResult.java

@@ -0,0 +1,74 @@
+package ru.vk.itmo.test.klimplyasov;
+
+import one.nio.http.HttpSession;
+import one.nio.http.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class MergeHandleResult {
+    private static final Logger log = LoggerFactory.getLogger(MergeHandleResult.class);
+    private final HandleResult[] handleResults;
+    private final AtomicInteger totalCount;
+    private final AtomicInteger validCount;
+    private final int ackThreshold;
+    private final int totalExpected;
+    private final HttpSession currentSession;
+
+    public MergeHandleResult(HttpSession session, int size, int ackThreshold) {
+        this.currentSession = session;
+        this.handleResults = new HandleResult[size];
+        this.totalCount = new AtomicInteger();
+        this.validCount = new AtomicInteger();
+        this.ackThreshold = ackThreshold;
+        this.totalExpected = size;
+    }
+
+    public boolean add(int index, HandleResult handleResult) {
+        handleResults[index] = handleResult;
+        int valid = handleResult.status() == HttpURLConnection.HTTP_OK
+                || handleResult.status() == HttpURLConnection.HTTP_CREATED
+                || handleResult.status() == HttpURLConnection.HTTP_ACCEPTED
+                || handleResult.status() == HttpURLConnection.HTTP_NOT_FOUND
+                ? validCount.incrementAndGet() : validCount.get();
+        if (valid >= ackThreshold || totalCount.incrementAndGet() == totalExpected) {
+            sendResult();
+            return true;
+        }
+        return false;
+    }
+
+    private void sendResult() {
+        HandleResult mergedResult = new HandleResult(HttpURLConnection.HTTP_GATEWAY_TIMEOUT, null);
+        int localValidCount = 0;
+        for (HandleResult handleResult : handleResults) {
+            if (handleResult.status() == HttpURLConnection.HTTP_OK
+                    || handleResult.status() == HttpURLConnection.HTTP_CREATED
+                    || handleResult.status() == HttpURLConnection.HTTP_ACCEPTED
+                    || handleResult.status() == HttpURLConnection.HTTP_NOT_FOUND) {
+                localValidCount++;
+                if (mergedResult.timestamp() <= handleResult.timestamp()) {
+                    mergedResult = handleResult;
+                }
+            }
+        }
+        try {
+            if (localValidCount < ackThreshold) {
+                currentSession.sendResponse(new Response(Response.GATEWAY_TIMEOUT, Response.EMPTY));
+            } else {
+                currentSession.sendResponse(new Response(String.valueOf(mergedResult.status()), mergedResult.data()));
+            }
+        } catch (Exception e) {
+            log.error("Exception during handleRequest", e);
+            try {
+                currentSession.sendResponse(new Response(Response.INTERNAL_ERROR, Response.EMPTY));
+            } catch (IOException ex) {
+                log.error("Exception while sending close connection", e);
+                currentSession.scheduleClose();
+            }
+        }
+    }
+}