Stage 6, Андрей Чешев, Политех

src/main/java/ru/vk/itmo/test/andreycheshev/AsyncActions.java

@@ -0,0 +1,225 @@
+package ru.vk.itmo.test.andreycheshev;
+
+import one.nio.http.HttpSession;
+import one.nio.http.Request;
+import one.nio.http.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+public class AsyncActions {
+    private static final Logger LOGGER = LoggerFactory.getLogger(AsyncActions.class);
+
+    private static final int CPU_THREADS_COUNT = Runtime.getRuntime().availableProcessors();
+
+    public static final String FUTURE_CREATION_ERROR = "Error when CompletableFuture creation";
+
+    private final Executor internalExecutor = Executors.newFixedThreadPool(
+            CPU_THREADS_COUNT / 2,
+            new WorkerThreadFactory("Internal-thread")
+    );
+    private final Executor senderExecutor = Executors.newFixedThreadPool(
+            CPU_THREADS_COUNT / 2,
+            new WorkerThreadFactory("Sender-thread")
+    );
+    private final Executor localCallExecutor = Executors.newFixedThreadPool(
+            CPU_THREADS_COUNT / 2,
+            new WorkerThreadFactory("LocalCall-thread")
+    );
+    private final Executor remoteCallExecutor = Executors.newFixedThreadPool(
+            CPU_THREADS_COUNT,
+            new WorkerThreadFactory("RemoteCall-thread")
+    );
+    private final Executor streamingExecutor = Executors.newFixedThreadPool(
+            CPU_THREADS_COUNT,
+            new WorkerThreadFactory("Streaming-thread")
+    );
+
+    private final HttpClient httpClient = HttpClient.newBuilder()
+            .executor(remoteCallExecutor)
+            .connectTimeout(Duration.ofMillis(500))
+            .version(HttpClient.Version.HTTP_1_1)
+            .build();
+
+    private final HttpProvider httpProvider;
+
+    public AsyncActions(HttpProvider httpProvider) {
+        this.httpProvider = httpProvider;
+    }
+
+    public CompletableFuture<Void> processLocallyToSend(
+            String id,
+            Request request,
+            long timestamp,
+            HttpSession session) {
+
+        int method = request.getMethod();
+
+        return getLocalFuture(method, id, request, timestamp)
+                .thenAcceptAsync(
+                        elements -> HttpUtils.sendResponse(
+                                HttpUtils.getOneNioResponse(method, elements),
+                                session
+                        ),
+                        senderExecutor
+                )
+                .exceptionallyAsync(
+                        exception -> {
+                            if (exception.getCause() instanceof UncheckedIOException) {
+                                LOGGER.error("Error when sending a request", exception);
+                            }
+                            return null;
+                        },
+                        internalExecutor
+                );
+    }
+
+    public CompletableFuture<Void> processLocallyToCollect(
+            int method,
+            String id,
+            Request request,
+            long timestamp,
+            ResponseCollector collector) {
+
+        CompletableFuture<Void> future = getLocalFuture(method, id, request, timestamp)
+                .thenApplyAsync(
+                        collector::add,
+                        internalExecutor
+                )
+                .exceptionallyAsync(
+                        exception -> {
+                            LOGGER.error("Internal error of the DAO operation", exception);
+                            return collector.incrementResponsesCounter();
+                        },
+                        internalExecutor
+                )
+                .thenAcceptAsync(
+                        condition -> {
+                            if (condition) {
+                                collector.sendResponse();
+                            }
+                        },
+                        senderExecutor
+                );
+
+        return withSendingErrorProcessing(future);
+    }
+
+    public CompletableFuture<Void> processRemotelyToCollect(
+            String node,
+            Request request,
+            long timestamp,
+            ResponseCollector collector) {
+
+        HttpRequest httpRequest = HttpRequest.newBuilder(URI.create(node + request.getURI()))
+                .method(
+                        request.getMethodName(),
+                        request.getBody() == null
+                                ? HttpRequest.BodyPublishers.noBody()
+                                : HttpRequest.BodyPublishers.ofByteArray(request.getBody())
+                )
+                .header(HttpUtils.TIMESTAMP_JAVA_NET_HEADER, String.valueOf(timestamp))
+                .timeout(Duration.ofMillis(1500))
+                .build();
+
+        CompletableFuture<Void> future = httpClient
+                .sendAsync(httpRequest, HttpResponse.BodyHandlers.ofByteArray())
+                .thenApplyAsync(
+                        response -> { // java.net response.
+                            return collector.add(
+                                    HttpUtils.getElementsFromJavaNetResponse(response)
+                            );
+                        },
+                        internalExecutor
+                )
+                .exceptionallyAsync(
+                        exception -> {
+                            LOGGER.error("Error when sending a request to the remote node", exception);
+                            return collector.incrementResponsesCounter();
+                        },
+                        internalExecutor
+                )
+                .thenAcceptAsync(
+                        condition -> {
+                            if (condition) {
+                                collector.sendResponse();
+                            }
+                        },
+                        senderExecutor
+                );
+
+        return withSendingErrorProcessing(future);
+    }
+
+    public void stream(Runnable runnable) {
+        CompletableFuture<Void> future = CompletableFuture.runAsync(
+                runnable,
+                streamingExecutor
+        ).exceptionallyAsync(
+                exception -> {
+                    LOGGER.error("Error while streaming process", exception);
+                    return null;
+                },
+                internalExecutor
+        );
+
+        checkFuture(future);
+    }
+
+    private CompletableFuture<ResponseElements> getLocalFuture(
+            int method,
+            String id,
+            Request request,
+            long timestamp) {
+
+        return CompletableFuture.supplyAsync(
+                () -> switch (method) {
+                    case Request.METHOD_GET -> httpProvider.get(id);
+                    case Request.METHOD_PUT -> httpProvider.put(id, request.getBody(), timestamp);
+                    default -> httpProvider.delete(id, timestamp);
+                },
+                localCallExecutor
+        );
+    }
+
+    private CompletableFuture<Void> withSendingErrorProcessing(
+            CompletableFuture<Void> future) {
+
+        return future
+                .exceptionallyAsync(
+                        exception -> {
+                            LOGGER.error("Error when sending a request", exception);
+                            return null;
+                        },
+                        internalExecutor
+                );
+    }
+
+    public void sendAsync(Response response, HttpSession session) {
+        CompletableFuture<Void> future = withSendingErrorProcessing(
+                CompletableFuture.runAsync(
+                        () -> HttpUtils.sendResponse(response, session),
+                        senderExecutor
+                )
+        );
+
+        checkFuture(future);
+    }
+
+    public static boolean checkFuture(CompletableFuture<?> future) {
+        boolean isFutureNull = future == null;
+        if (isFutureNull) {
+            LOGGER.error(FUTURE_CREATION_ERROR);
+        }
+        return isFutureNull;
+    }
+}

src/main/java/ru/vk/itmo/test/andreycheshev/HttpProvider.java

@@ -0,0 +1,9 @@
+package ru.vk.itmo.test.andreycheshev;
+
+public interface HttpProvider {
+    ResponseElements get(String id);
+
+    ResponseElements put(String id, byte[] body, long timestamp);
+
+    ResponseElements delete(String id, long timestamp);
+}

src/main/java/ru/vk/itmo/test/andreycheshev/HttpUtils.java

@@ -0,0 +1,97 @@
+package ru.vk.itmo.test.andreycheshev;
+
+import one.nio.http.HttpSession;
+import one.nio.http.Request;
+import one.nio.http.Response;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.net.http.HttpResponse;
+import java.util.Map;
+import java.util.Optional;
+
+public class HttpUtils {
+    public static final String TIMESTAMP_JAVA_NET_HEADER = "X-Timestamp";
+    public static final String TIMESTAMP_ONE_NIO_HEADER = TIMESTAMP_JAVA_NET_HEADER + ": ";
+
+    private static final String NOT_ENOUGH_REPLICAS_STATUS = "504 Not Enough Replicas";
+    private static final String TOO_MANY_REQUESTS_STATUS = "429 Too many requests";
+
+    public static final int EMPTY_TIMESTAMP = -1;
+
+    private static final Map<Integer, String> AVAILABLE_RESPONSES = Map.of(
+            200, Response.OK,
+            201, Response.CREATED,
+            202, Response.ACCEPTED,
+            404, Response.NOT_FOUND,
+            410, Response.GONE
+    ); // Immutable map.
+
+    private HttpUtils() {
+
+    }
+
+    public static Response getBadRequest() {
+        return new Response(Response.BAD_REQUEST, Response.EMPTY);
+    }
+
+    public static Response getMethodNotAllowed() {
+        return new Response(Response.METHOD_NOT_ALLOWED, Response.EMPTY);
+    }
+
+    public static Response getInternalError() {
+        return new Response(Response.INTERNAL_ERROR, Response.EMPTY);
+    }
+
+    public static Response getNotEnoughReplicas() {
+        return new Response(NOT_ENOUGH_REPLICAS_STATUS, Response.EMPTY);
+    }
+
+    public static Response getTooManyRequests() {
+        return new Response(TOO_MANY_REQUESTS_STATUS, Response.EMPTY);
+    }
+
+    public static void sendResponse(Response response, HttpSession session) {
+        try {
+            session.sendResponse(response);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+
+    public static ResponseElements getElementsFromJavaNetResponse(HttpResponse<byte[]> response) {
+        Optional<String> optTimestamp = response.headers().firstValue(TIMESTAMP_JAVA_NET_HEADER);
+
+        long responseTimestamp = optTimestamp.isPresent()
+                ? Long.parseLong(optTimestamp.get())
+                : EMPTY_TIMESTAMP;
+
+        return new ResponseElements(
+                response.statusCode(),
+                response.body(),
+                responseTimestamp
+        );
+    }
+
+    public static Response getOneNioResponse(int method, ResponseElements elements) {
+        switch (method) {
+            case Request.METHOD_GET -> {
+                int status = elements.getStatus();
+
+                Response response = status == 410
+                        ? new Response(Response.NOT_FOUND, Response.EMPTY)
+                        : new Response(AVAILABLE_RESPONSES.get(status), elements.getBody());
+
+                response.addHeader(TIMESTAMP_ONE_NIO_HEADER + elements.getTimestamp());
+
+                return response;
+            }
+            case Request.METHOD_PUT -> {
+                return new Response(Response.CREATED, Response.EMPTY);
+            }
+            default -> { // For delete method.
+                return new Response(Response.ACCEPTED, Response.EMPTY);
+            }
+        }
+    }
+}