JAVA-24463 Cleanup spring-reactive-modules (#14811)

Co-authored-by: timis1 <noreplay@yahoo.com>
2023-09-25 06:42:56 +03:00
parent c689708383
commit a3ab8131c9
354 changed files with 377 additions and 371 deletions
@@ -0,0 +1,12 @@
+package com.baeldung.limitrequests;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class LimitRequestsApp {
+
+    public static void main(String[] args) {
+        SpringApplication.run(LimitRequestsApp.class, args);
+    }
+}
@@ -0,0 +1,33 @@
+package com.baeldung.limitrequests.client;
+
+import java.time.Duration;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.baeldung.limitrequests.client.utils.Client;
+import com.baeldung.limitrequests.client.utils.RandomConsumer;
+
+import reactor.core.publisher.Flux;
+
+public class DelayElements {
+
+    private DelayElements() {
+    }
+
+    public static Flux<Integer> fetch(WebClient client, int requests, int delay) {
+        String clientId = Client.id(requests, DelayElements.class, delay);
+
+        return Flux.range(1, requests)
+            .log()
+            .delayElements(Duration.ofMillis(delay))
+            .flatMap(i -> RandomConsumer.get(client, clientId));
+    }
+
+    public static void main(String[] args) {
+        String baseUrl = args[0];
+        WebClient client = WebClient.create(baseUrl);
+
+        fetch(client, 12, 1050).doOnNext(System.out::println)
+            .blockLast();
+    }
+}
@@ -0,0 +1,37 @@
+package com.baeldung.limitrequests.client;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.baeldung.limitrequests.client.utils.Client;
+import com.baeldung.limitrequests.client.utils.RandomConsumer;
+import com.google.common.util.concurrent.RateLimiter;
+
+import reactor.core.publisher.Flux;
+
+public class GuavaRateLimit {
+
+    private GuavaRateLimit() {
+    }
+
+    public static Flux<Integer> fetch(WebClient client, int requests, double limit) {
+        String clientId = Client.id(requests, GuavaRateLimit.class, limit);
+
+        RateLimiter limiter = RateLimiter.create(limit);
+
+        return Flux.range(1, requests)
+            .log()
+            .flatMap(i -> {
+                limiter.acquire();
+
+                return RandomConsumer.get(client, clientId);
+            });
+    }
+
+    public static void main(String[] args) throws InterruptedException {
+        String baseUrl = args[0];
+        WebClient client = WebClient.create(baseUrl);
+
+        fetch(client, 20, 2).doOnNext(System.out::println)
+            .blockLast();
+    }
+}
@@ -0,0 +1,30 @@
+package com.baeldung.limitrequests.client;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.baeldung.limitrequests.client.utils.Client;
+import com.baeldung.limitrequests.client.utils.RandomConsumer;
+
+import reactor.core.publisher.Flux;
+
+public class LimitConcurrency {
+
+    private LimitConcurrency() {
+    }
+
+    public static Flux<Integer> fetch(WebClient client, int requests, int concurrency) {
+        String clientId = Client.id(requests, LimitConcurrency.class.getSimpleName(), concurrency);
+
+        return Flux.range(1, requests)
+            .log()
+            .flatMap(i -> RandomConsumer.get(client, clientId), concurrency);
+    }
+
+    public static void main(String[] args) throws InterruptedException {
+        String baseUrl = args[0];
+        WebClient client = WebClient.create(baseUrl);
+
+        fetch(client, 12, 5).doOnNext(System.out::println)
+            .blockLast();
+    }
+}
@@ -0,0 +1,42 @@
+package com.baeldung.limitrequests.client;
+
+import java.time.Duration;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.baeldung.limitrequests.client.utils.Client;
+import com.baeldung.limitrequests.client.utils.RandomConsumer;
+
+import io.github.resilience4j.ratelimiter.RateLimiter;
+import io.github.resilience4j.ratelimiter.RateLimiterConfig;
+import io.github.resilience4j.reactor.ratelimiter.operator.RateLimiterOperator;
+import reactor.core.publisher.Flux;
+
+public class Resilience4jRateLimit {
+
+    private Resilience4jRateLimit() {
+    }
+
+    public static Flux<Integer> fetch(WebClient client, int requests, int concurrency, int interval) {
+        RateLimiter limiter = RateLimiter.of("my-rate-limiter", RateLimiterConfig.custom()
+            .limitRefreshPeriod(Duration.ofMillis(interval))
+            .limitForPeriod(concurrency)
+            .timeoutDuration(Duration.ofMillis((long) interval * concurrency))
+            .build());
+
+        String clientId = Client.id(requests, Resilience4jRateLimit.class, concurrency, interval);
+
+        return Flux.range(1, requests)
+            .log()
+            .flatMap(i -> RandomConsumer.<Integer> get(client, clientId)
+                .transformDeferred(RateLimiterOperator.of(limiter)));
+    }
+
+    public static void main(String[] args) {
+        String baseUrl = args[0];
+        WebClient client = WebClient.create(baseUrl);
+
+        fetch(client, 10, 5, 2500).doOnNext(System.out::println)
+            .blockLast();
+    }
+}
@@ -0,0 +1,33 @@
+package com.baeldung.limitrequests.client;
+
+import java.time.Duration;
+
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.baeldung.limitrequests.client.utils.Client;
+import com.baeldung.limitrequests.client.utils.RandomConsumer;
+
+import reactor.core.publisher.Flux;
+
+public class ZipWithInterval {
+
+    private ZipWithInterval() {
+    }
+
+    public static Flux<Integer> fetch(WebClient client, int requests, int delay) {
+        String clientId = Client.id(requests, ZipWithInterval.class, delay);
+
+        return Flux.range(1, requests)
+            .log()
+            .zipWith(Flux.interval(Duration.ofMillis(delay)))
+            .flatMap(i -> RandomConsumer.get(client, clientId));
+    }
+
+    public static void main(String[] args) {
+        String baseUrl = args[0];
+        WebClient client = WebClient.create(baseUrl);
+
+        fetch(client, 15, 1100).doOnNext(System.out::println)
+            .blockLast();
+    }
+}
@@ -0,0 +1,16 @@
+package com.baeldung.limitrequests.client.utils;
+
+public interface Client {
+
+    String SEPARATOR = ":";
+
+    static String id(Object... args) {
+        StringBuilder builder = new StringBuilder();
+        for (Object object : args) {
+            builder.append(":")
+                .append(object.toString());
+        }
+        return builder.toString()
+            .replaceFirst(SEPARATOR, "");
+    }
+}
@@ -0,0 +1,19 @@
+package com.baeldung.limitrequests.client.utils;
+
+import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.baeldung.limitrequests.server.RandomController;
+
+import reactor.core.publisher.Mono;
+
+public interface RandomConsumer {
+
+    static <T> Mono<T> get(WebClient client, String id) {
+        return client.get()
+            .header(RandomController.CLIENT_ID_HEADER, id)
+            .retrieve()
+            .bodyToMono(new ParameterizedTypeReference<T>() {
+            });
+    }
+}
@@ -0,0 +1,38 @@
+package com.baeldung.limitrequests.server;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.IntSupplier;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Concurrency {
+
+    public static final int MAX_CONCURRENT_REQUESTS = 5;
+    private static final Logger logger = LoggerFactory.getLogger(Concurrency.class);
+    private static final Map<String, AtomicInteger> CONCURRENT_REQUESTS = new HashMap<>();
+
+    private Concurrency() {
+    }
+
+    public static int protect(String clientId, IntSupplier supplier) throws InterruptedException {
+        AtomicInteger counter = CONCURRENT_REQUESTS.computeIfAbsent(clientId, k -> new AtomicInteger(0));
+
+        try {
+            int n = counter.incrementAndGet();
+            if (n > MAX_CONCURRENT_REQUESTS) {
+                String message = String.format("%s - %d max concurrent requests reached [%d]. try again later", clientId, MAX_CONCURRENT_REQUESTS, n);
+                throw new UnsupportedOperationException(message);
+            }
+
+            logger.info("{} - {}", clientId, n);
+            TimeUnit.SECONDS.sleep(2);
+            return supplier.getAsInt();
+        } finally {
+            counter.decrementAndGet();
+        }
+    }
+}
@@ -0,0 +1,21 @@
+package com.baeldung.limitrequests.server;
+
+import java.util.Random;
+
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestHeader;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/random")
+public class RandomController {
+
+    public static final String CLIENT_ID_HEADER = "client-id";
+    private static final Random RANDOM = new Random();
+
+    @GetMapping
+    Integer getRandom(@RequestHeader(CLIENT_ID_HEADER) String clientId) throws InterruptedException {
+        return Concurrency.protect(clientId, () -> RANDOM.nextInt(50));
+    }
+}
@@ -0,0 +1,12 @@
+package com.baeldung.streamlargefile;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class StreamLargeFileApp {
+
+    public static void main(String... args) {
+        SpringApplication.run(StreamLargeFileApp.class, args);
+    }
+}
@@ -0,0 +1,41 @@
+package com.baeldung.streamlargefile.client;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import reactor.core.publisher.Flux;
+
+public class LargeFileDownloadWebClient {
+
+    private LargeFileDownloadWebClient() {
+    }
+
+    public static long fetch(WebClient client, String destination) throws IOException {
+        Flux<DataBuffer> flux = client.get()
+            .retrieve()
+            .bodyToFlux(DataBuffer.class);
+
+        Path path = Paths.get(destination);
+
+        DataBufferUtils.write(flux, path)
+            .block();
+
+        return Files.size(path);
+    }
+
+    public static void main(String... args) throws IOException {
+        String baseUrl = args[0];
+        String destination = args[1];
+
+        WebClient client = WebClient.create(baseUrl);
+
+        long bytes = fetch(client, destination);
+        System.out.printf("downloaded %d bytes to %s", bytes, destination);
+    }
+}
@@ -0,0 +1,55 @@
+package com.baeldung.streamlargefile.client;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import org.springframework.web.reactive.function.client.ExchangeStrategies;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import reactor.core.publisher.Mono;
+
+public class LimitedFileDownloadWebClient {
+
+    private LimitedFileDownloadWebClient() {
+    }
+
+    public static long fetch(WebClient client, String destination) throws IOException {
+        Mono<byte[]> mono = client.get()
+            .retrieve()
+            .bodyToMono(byte[].class)
+            .onErrorMap(RuntimeException::new);
+
+        byte[] bytes = mono.block();
+
+        Path path = Paths.get(destination);
+        Files.write(path, bytes);
+
+        return bytes.length;
+    }
+
+    public static void main(String... args) throws IOException {
+        String baseUrl = args[0];
+        String destination = args[1];
+
+        WebClient client = WebClient.builder()
+            .baseUrl(baseUrl)
+            .exchangeStrategies(useMaxMemory())
+            .build();
+
+        long bytes = fetch(client, destination);
+        System.out.printf("downloaded %d bytes to %s", bytes, destination);
+    }
+
+    public static ExchangeStrategies useMaxMemory() {
+        long totalMemory = Runtime.getRuntime()
+            .maxMemory();
+
+        return ExchangeStrategies.builder()
+            .codecs(configurer ->
+                configurer.defaultCodecs()
+                .maxInMemorySize((int) totalMemory))
+            .build();
+    }
+}
@@ -0,0 +1,31 @@
+package com.baeldung.streamlargefile.server;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+
+import org.springframework.core.io.FileSystemResource;
+import org.springframework.core.io.Resource;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/large-file")
+public class LargeFileController {
+
+    public static final Path downloadPath = Paths.get("/tmp/large.dat");
+
+    @GetMapping("size")
+    Long getSize() throws IOException {
+        return Files.size(downloadPath);
+    }
+
+    @GetMapping
+    ResponseEntity<Resource> get() {
+        return ResponseEntity.ok()
+            .body(new FileSystemResource(downloadPath));
+    }
+}
@@ -0,0 +1,5 @@
+logging.level.root=INFO
+
+server.port=8081
+
+logging.level.reactor.netty.http.client.HttpClient=DEBUG
@@ -0,0 +1,16 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
+        <layout class="ch.qos.logback.classic.PatternLayout">
+            # Pattern of log message for console appender
+            <Pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>
+        </layout>
+    </appender>
+    
+    <logger name="org.springframework" level="INFO" />
+
+    <root level="INFO">
+        <appender-ref ref="stdout" />
+    </root>
+    
+</configuration>
@@ -0,0 +1,12 @@
+#!/bin/sh
+
+generate() {
+	file="$1"
+	size="$2"
+	
+	fallocate -l "$size" "$file"
+	ls -lah "$file"
+}
+
+generate /tmp/small.dat 128K
+generate /tmp/large.dat 128M
@@ -0,0 +1,21 @@
+#!/bin/bash
+MYSELF="$(readlink -f "$0")"
+MYDIR="${MYSELF%/*}"
+
+client="${1:-Large}"
+url="${2:-http://localhost:8081/large-file}"
+download_destination="${3:-/tmp/download.dat}"
+xmx="${4:-32m}"
+
+module_dir="$(readlink -f "$MYDIR/../../../..")"
+
+echo "module: $module_dir"
+cd $module_dir || exit
+
+echo "packaging..."
+mvn clean package dependency:copy-dependencies
+
+echo "GET $url with $client client..."
+java -Xmx$xmx -cp target/dependency/*:target/* \
+"com.baeldung.streamlargefile.client.${client}FileDownloadWebClient" \
+"$url" "$download_destination"
@@ -0,0 +1,113 @@
+package com.baeldung.limitrequests;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClientResponseException.InternalServerError;
+
+import com.baeldung.limitrequests.client.DelayElements;
+import com.baeldung.limitrequests.client.GuavaRateLimit;
+import com.baeldung.limitrequests.client.LimitConcurrency;
+import com.baeldung.limitrequests.client.Resilience4jRateLimit;
+import com.baeldung.limitrequests.client.ZipWithInterval;
+import com.baeldung.limitrequests.server.Concurrency;
+
+class RandomControllerLiveTest {
+
+    private static final int MAX_CONCURRENT_REQUESTS = Concurrency.MAX_CONCURRENT_REQUESTS;
+    private static final int TOTAL_REQUESTS = 10;
+
+    private WebClient client = WebClient.create("http://localhost:8081/random");
+
+    @Test
+    void givenEnoughDelay_whenZipWithInterval_thenNoExceptionThrown() {
+        int delay = MAX_CONCURRENT_REQUESTS * 100;
+
+        assertDoesNotThrow(() -> {
+            ZipWithInterval.fetch(client, TOTAL_REQUESTS, delay)
+                .doOnNext(System.out::println)
+                .blockLast();
+        });
+    }
+
+    @Test
+    void givenSmallDelay_whenZipWithInterval_thenExceptionThrown() {
+        int delay = 100;
+
+        assertThrows(InternalServerError.class, () -> {
+            ZipWithInterval.fetch(client, TOTAL_REQUESTS, delay)
+                .doOnNext(System.out::println)
+                .blockLast();
+        });
+    }
+
+    @Test
+    void givenEnoughDelay_whenDelayElements_thenNoExceptionThrown() {
+        int delay = MAX_CONCURRENT_REQUESTS * 100;
+
+        assertDoesNotThrow(() -> {
+            DelayElements.fetch(client, TOTAL_REQUESTS, delay)
+                .doOnNext(System.out::println)
+                .blockLast();
+        });
+    }
+
+    @Test
+    void givenSmallDelay_whenDelayElements_thenExceptionThrown() {
+        int delay = 100;
+
+        assertThrows(InternalServerError.class, () -> {
+            DelayElements.fetch(client, TOTAL_REQUESTS, delay)
+                .doOnNext(System.out::println)
+                .blockLast();
+        });
+    }
+
+    @Test
+    void givenLimitInsideServerRange_whenLimitedConcurrency_thenNoExceptionThrown() {
+        int limit = MAX_CONCURRENT_REQUESTS;
+
+        assertDoesNotThrow(() -> {
+            LimitConcurrency.fetch(client, TOTAL_REQUESTS, limit)
+                .doOnNext(System.out::println)
+                .blockLast();
+        });
+    }
+
+    @Test
+    void givenLimitOutsideServerRange_whenLimitedConcurrency_thenExceptionThrown() {
+        int limit = MAX_CONCURRENT_REQUESTS + TOTAL_REQUESTS;
+
+        assertThrows(InternalServerError.class, () -> {
+            LimitConcurrency.fetch(client, TOTAL_REQUESTS, limit)
+                .doOnNext(System.out::println)
+                .blockLast();
+        });
+    }
+
+    @Test
+    void givenLongInterval_whenRateLimited_thenNoExceptionThrown() {
+        int interval = MAX_CONCURRENT_REQUESTS * 500;
+
+        int limit = MAX_CONCURRENT_REQUESTS;
+
+        assertDoesNotThrow(() -> {
+            Resilience4jRateLimit.fetch(client, TOTAL_REQUESTS, limit, interval)
+                .doOnNext(System.out::println)
+                .blockLast();
+        });
+    }
+
+    @Test
+    void givenShortLimit_whenUsingGuavaRateLimiter_thenNoExceptionThrown() {
+        double limit = MAX_CONCURRENT_REQUESTS / 2;
+
+        assertDoesNotThrow(() -> {
+            GuavaRateLimit.fetch(client, TOTAL_REQUESTS, limit)
+                .doOnNext(System.out::println)
+                .blockLast();
+        });
+    }
+}
@@ -0,0 +1,59 @@
+package com.baeldung.streamlargefile;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.web.reactive.function.client.WebClient;
+
+import com.baeldung.streamlargefile.client.LargeFileDownloadWebClient;
+import com.baeldung.streamlargefile.client.LimitedFileDownloadWebClient;
+import com.baeldung.streamlargefile.server.LargeFileController;
+
+class LargeFileControllerLiveTest {
+
+    private static final String BASE_URL = "http://localhost:8081/large-file";
+    private static final String DOWNLOAD_DESTINATION = LargeFileController.downloadPath.resolveSibling("download.dat")
+        .toString();
+    private static final Path downloadFile = LargeFileController.downloadPath;
+    private static final Runtime runtime = Runtime.getRuntime();
+    private static final Long xmx = runtime.maxMemory();
+
+    private WebClient client = WebClient.create(BASE_URL);
+
+    @BeforeAll
+    static void init() throws IOException {
+        if (!Files.exists(downloadFile)) {
+            ClassPathResource res = new ClassPathResource("streamlargefile/generate-sample-files.sh");
+
+            runtime.exec(res.getFile()
+                .getAbsolutePath());
+        }
+    }
+
+    @Test
+    void givenMemorySafeClient_whenFileLargerThanXmx_thenFileDownloaded() throws IOException {
+        if (xmx < Files.size(downloadFile)) {
+            long size = LargeFileDownloadWebClient.fetch(client, DOWNLOAD_DESTINATION);
+            assertTrue(size > xmx);
+        }
+    }
+
+    @Test
+    void givenLimitedClient_whenXmxLargerThanFile_thenFileDownloaded() throws IOException {
+        WebClient client = WebClient.builder()
+            .baseUrl(BASE_URL)
+            .exchangeStrategies(LimitedFileDownloadWebClient.useMaxMemory())
+            .build();
+
+        if (xmx > Files.size(downloadFile)) {
+            long size = LimitedFileDownloadWebClient.fetch(client, DOWNLOAD_DESTINATION);
+            assertTrue(size < xmx);
+        }
+    }
+}
@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="stdout" class="ch.qos.logback.core.ConsoleAppender">
+        <layout class="ch.qos.logback.classic.PatternLayout">
+            # Pattern of log message for console appender
+            <Pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>
+        </layout>
+    </appender>
+
+    <logger name="org.springframework" level="INFO" />
+
+    <logger name="com.baeldung.reactive.logging.jetty" level="DEBUG" />
+    <logger name="reactor.netty.http.client.HttpClient" level="DEBUG" />
+    <logger name="com.baeldung.reactive.logging" level="DEBUG" />
+
+    <root level="INFO">
+        <appender-ref ref="stdout" />
+    </root>
+
+</configuration>