Make hw7

bot/pom.xml

@@ -73,6 +73,12 @@
         </dependency>
 
         <!-- Tests -->
+        <dependency>
+            <groupId>org.junit.vintage</groupId>
+            <artifactId>junit-vintage-engine</artifactId>
+            <version>5.10.1</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
@@ -130,10 +136,31 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.junit.vintage</groupId>
-            <artifactId>junit-vintage-engine</artifactId>
-            <version>5.10.1</version>
-            <scope>test</scope>
+            <groupId>com.giffing.bucket4j.spring.boot.starter</groupId>
+            <artifactId>bucket4j-spring-boot-starter</artifactId>
+            <version>0.12.5</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-cache</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>javax.cache</groupId>
+            <artifactId>cache-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>jcache</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>edu.java</groupId>
+            <artifactId>retry</artifactId>
+            <version>0.1</version>
+            <scope>compile</scope>
         </dependency>
     </dependencies>
 

bot/src/main/java/edu/java/bot/BotApplication.java

@@ -1,12 +1,15 @@
 package edu.java.bot;
 
+import edu.java.RetryQueryConfiguration;
 import edu.java.bot.configuration.ApplicationConfig;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cache.annotation.EnableCaching;
 
 @SpringBootApplication
-@EnableConfigurationProperties(ApplicationConfig.class)
+@EnableCaching
+@EnableConfigurationProperties({ApplicationConfig.class, RetryQueryConfiguration.class})
 public class BotApplication {
     public static void main(String[] args) {
         SpringApplication.run(BotApplication.class, args);

bot/src/main/java/edu/java/bot/configuration/ScrapperClientConfiguration.java

@@ -1,5 +1,7 @@
 package edu.java.bot.configuration;
 
+import edu.java.RetryFactory;
+import edu.java.RetryQueryConfiguration;
 import edu.java.bot.client.scrapper.ScrapperClient;
 import lombok.extern.log4j.Log4j2;
 import org.springframework.beans.factory.annotation.Value;
@@ -9,6 +11,7 @@
 import org.springframework.web.reactive.function.client.support.WebClientAdapter;
 import org.springframework.web.service.invoker.HttpServiceProxyFactory;
 import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
 
 @Configuration
 @Log4j2
@@ -18,10 +21,12 @@ public class ScrapperClientConfiguration {
     private String scrapperUrl;
 
     @Bean
-    public ScrapperClient scrapperClient() {
+    public ScrapperClient scrapperClient(RetryQueryConfiguration retryQueryConfiguration) {
+        Retry retry = RetryFactory.createRetry(retryQueryConfiguration, "scrapper");
         WebClient webClient = WebClient.builder()
             .defaultStatusHandler(httpStatusCode -> true, clientResponse -> Mono.empty())
             .defaultHeader("Content-Type", "application/json")
+            .filter(RetryFactory.createFilter(retry))
             .baseUrl(scrapperUrl).build();
 
         HttpServiceProxyFactory httpServiceProxyFactory = HttpServiceProxyFactory

bot/src/main/resources/application.yml

@@ -6,6 +6,13 @@ spring:
     name: bot
   jackson:
     time-zone: UTC
+  cache:
+    jcache:
+      provider: com.github.benmanes.caffeine.jcache.spi.CaffeineCachingProvider
+    caffeine:
+      spec: maximumSize=100000,expireAfterAccess=3600s
+    cache-names:
+      - rate-limit-bucket
 
 server:
   port: 8090
@@ -19,3 +26,26 @@ springdoc:
 
 scrapper:
   url: http://localhost:8080
+
+retry-query:
+  retries:
+    - target: scrapper
+      type: exponential
+      max-attempts: 3
+      min-delay: 1s
+      max-delay: 10s
+      codes: 429
+
+bucket4j:
+  enabled: true
+  filters:
+    - cache-name: rate-limit-bucket
+      url: .*
+      http-status-code: too_many_requests
+      rate-limits:
+        - bandwidths:
+            - capacity: 1000
+              time: 1
+              unit: hours
+              refill-speed: interval
+          cache-key: getRemoteAddr()

pom.xml

@@ -49,6 +49,7 @@
         <module>bot</module>
         <module>scrapper</module>
         <module>scrapper-jooq</module>
+        <module>retry</module>
     </modules>
 
     <dependencyManagement>

retry/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>edu.java</groupId>
+        <artifactId>root</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>retry</artifactId>
+    <version>${revision}</version>
+
+    <properties>
+        <maven.compiler.source>21</maven.compiler.source>
+        <maven.compiler.target>21</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-webflux</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-context</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.retry</groupId>
+            <artifactId>spring-retry</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+    </dependencies>
+
+</project>

retry/src/main/java/edu/java/ErrorFilterPredicate.java

@@ -0,0 +1,21 @@
+package edu.java;
+
+import java.util.List;
+import java.util.function.Predicate;
+import org.springframework.web.reactive.function.client.WebClientResponseException;
+
+public class ErrorFilterPredicate implements Predicate<Throwable> {
+    private final List<Integer> retryCodes;
+
+    public ErrorFilterPredicate(List<Integer> retryCodes) {
+        this.retryCodes = retryCodes;
+    }
+
+    @Override
+    public boolean test(Throwable throwable) {
+        if (throwable instanceof WebClientResponseException e) {
+            return retryCodes.contains(e.getStatusCode().value());
+        }
+        return true;
+    }
+}

retry/src/main/java/edu/java/LinearRetryBackoffSpec.java

@@ -0,0 +1,93 @@
+package edu.java;
+
+import java.time.Duration;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import lombok.RequiredArgsConstructor;
+import org.reactivestreams.Publisher;
+import org.springframework.retry.ExhaustedRetryException;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+import reactor.util.retry.Retry;
+
+
+@RequiredArgsConstructor
+public class LinearRetryBackoffSpec extends Retry {
+    private static final Duration MAX_BACKOFF = Duration.ofMillis(Long.MAX_VALUE);
+    private final Duration minBackoff;
+    private final Duration maxBackoff;
+    private final double factor;
+    private final int maxAttempts;
+    private final Predicate<Throwable> errorFilter;
+    private final Supplier<Scheduler> schedulerSupplier;
+
+    public LinearRetryBackoffSpec factor(double factor) {
+        return new LinearRetryBackoffSpec(
+            this.minBackoff,
+            this.maxBackoff,
+            factor,
+            this.maxAttempts,
+            this.errorFilter,
+            this.schedulerSupplier
+        );
+    }
+
+    public LinearRetryBackoffSpec filter(Predicate<Throwable> errorFilter) {
+        return new LinearRetryBackoffSpec(
+            this.minBackoff,
+            this.maxBackoff,
+            this.factor,
+            this.maxAttempts,
+            errorFilter,
+            this.schedulerSupplier
+        );
+    }
+
+    public static LinearRetryBackoffSpec linear(int maxAttempts, Duration minDelay) {
+        return new LinearRetryBackoffSpec(
+            minDelay,
+            MAX_BACKOFF,
+            1.0,
+            maxAttempts,
+            e -> true,
+            Schedulers::parallel
+        );
+    }
+
+    @Override
+    public Publisher<?> generateCompanion(Flux<RetrySignal> retrySignals) {
+        return Flux.deferContextual(cv ->
+            retrySignals.contextWrite(cv)
+                .concatMap(retryWhenState -> {
+                    RetrySignal copy = retryWhenState.copy();
+                    Throwable currentFailure = copy.failure();
+                    long iteration = copy.totalRetries();
+                    if (currentFailure == null) {
+                        return Mono.error(new IllegalStateException(
+                            "Retry.RetrySignal#failure() not expected to be null"));
+                    }
+                    if (!errorFilter.test(currentFailure)) {
+                        return Mono.error(currentFailure);
+                    }
+                    if (iteration >= maxAttempts) {
+                        return Mono.error(new ExhaustedRetryException("Retry exhausted: " + this));
+                    }
+
+                    Duration nextBackoff;
+                    try {
+                        nextBackoff = minBackoff.multipliedBy((long) (iteration * factor));
+                        if (nextBackoff.compareTo(maxBackoff) > 0) {
+                            nextBackoff = maxBackoff;
+                        }
+                    } catch (ArithmeticException overflow) {
+                        nextBackoff = maxBackoff;
+                    }
+
+                    return Mono.delay(nextBackoff, schedulerSupplier.get()).contextWrite(cv);
+                })
+                .onErrorStop()
+        );
+    }
+}

retry/src/main/java/edu/java/RetryElement.java

@@ -0,0 +1,16 @@
+package edu.java;
+
+import java.time.Duration;
+import java.util.List;
+import org.jetbrains.annotations.NotNull;
+
+public record RetryElement(
+    @NotNull String target,
+    @NotNull String type,
+    int maxAttempts,
+    double factor,
+    Duration minDelay,
+    Duration maxDelay,
+    List<Integer> codes
+) {
+}

retry/src/main/java/edu/java/RetryFactory.java

@@ -0,0 +1,41 @@
+package edu.java;
+
+import edu.java.builders.ExponentialRetryBuilder;
+import edu.java.builders.FixedRetryBuilder;
+import edu.java.builders.LinearRetryBuilder;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+import lombok.experimental.UtilityClass;
+import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
+import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
+
+@UtilityClass
+public class RetryFactory {
+
+    private static final Map<String, Function<RetryElement, Retry>> RETRY_BUILDERS = new HashMap<>();
+
+    static {
+        RETRY_BUILDERS.put("fixed", new FixedRetryBuilder());
+        RETRY_BUILDERS.put("linear", new LinearRetryBuilder());
+        RETRY_BUILDERS.put("exponential", new ExponentialRetryBuilder());
+    }
+
+    public static ExchangeFilterFunction createFilter(Retry retry) {
+        return (response, next) -> next.exchange(response)
+            .flatMap(clientResponse -> {
+                if (clientResponse.statusCode().isError()) {
+                    return clientResponse.createError();
+                } else {
+                    return Mono.just(clientResponse);
+                }
+            }).retryWhen(retry);
+    }
+
+    public static Retry createRetry(RetryQueryConfiguration config, String target) {
+        return config.retries().stream().filter(element -> element.target().equals(target)).findFirst()
+            .map(element -> RETRY_BUILDERS.get(element.type()).apply(element))
+            .orElseThrow(() -> new RuntimeException("Unknown target " + target));
+    }
+}

retry/src/main/java/edu/java/RetryQueryConfiguration.java

@@ -0,0 +1,10 @@
+package edu.java;
+
+import java.util.List;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+@Validated
+@ConfigurationProperties(prefix = "retry-query", ignoreUnknownFields = false)
+public record RetryQueryConfiguration(List<RetryElement> retries) {
+}

retry/src/main/java/edu/java/builders/ExponentialRetryBuilder.java

@@ -0,0 +1,16 @@
+package edu.java.builders;
+
+import edu.java.ErrorFilterPredicate;
+import edu.java.RetryElement;
+import java.util.function.Function;
+import reactor.util.retry.Retry;
+import reactor.util.retry.RetryBackoffSpec;
+
+public class ExponentialRetryBuilder implements Function<RetryElement, Retry> {
+    @Override
+    public Retry apply(RetryElement retryElement) {
+        return RetryBackoffSpec.backoff(retryElement.maxAttempts(), retryElement.minDelay())
+            .maxBackoff(retryElement.maxDelay())
+            .filter(new ErrorFilterPredicate(retryElement.codes()));
+    }
+}