diff --git a/pom.xml b/pom.xml index 614f1df..bdc6f06 100644 --- a/pom.xml +++ b/pom.xml @@ -18,6 +18,7 @@ 21 + 2.3.0 @@ -26,6 +27,22 @@ spring-boot-starter-webflux + + + io.github.resilience4j + resilience4j-spring-boot3 + ${resilience4j.version} + + + io.github.resilience4j + resilience4j-reactor + ${resilience4j.version} + + + + org.springframework.boot + spring-boot-starter-validation + @@ -33,6 +50,11 @@ spring-boot-starter-test test + + io.projectreactor + reactor-test + test + diff --git a/src/main/java/com/example/productaffinity/domain/exception/ExternalServiceException.java b/src/main/java/com/example/productaffinity/domain/exception/ExternalServiceException.java new file mode 100644 index 0000000..a32eb12 --- /dev/null +++ b/src/main/java/com/example/productaffinity/domain/exception/ExternalServiceException.java @@ -0,0 +1,9 @@ +package com.example.productaffinity.domain.exception; + +//For exception on remote service. +public class ExternalServiceException extends RuntimeException { + + public ExternalServiceException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/src/main/java/com/example/productaffinity/domain/exception/ProductDetailUnavailableException.java b/src/main/java/com/example/productaffinity/domain/exception/ProductDetailUnavailableException.java new file mode 100644 index 0000000..834fb44 --- /dev/null +++ b/src/main/java/com/example/productaffinity/domain/exception/ProductDetailUnavailableException.java @@ -0,0 +1,15 @@ +package com.example.productaffinity.domain.exception; + +public class ProductDetailUnavailableException extends RuntimeException { + + private final String productId; + + public ProductDetailUnavailableException(String productId, Throwable cause) { + super("Product detail unavailable: " + productId, cause); + this.productId = productId; + } + + public String productId() { + return productId; + } +} diff --git a/src/main/java/com/example/productaffinity/domain/exception/ProductNotFoundException.java b/src/main/java/com/example/productaffinity/domain/exception/ProductNotFoundException.java new file mode 100644 index 0000000..aa91375 --- /dev/null +++ b/src/main/java/com/example/productaffinity/domain/exception/ProductNotFoundException.java @@ -0,0 +1,8 @@ +package com.example.productaffinity.domain.exception; + +public class ProductNotFoundException extends RuntimeException { + + public ProductNotFoundException(String productId) { + super("Product not found: " + productId); + } +} diff --git a/src/main/java/com/example/productaffinity/domain/port/outcoming/AffinityProductsIdsPort.java b/src/main/java/com/example/productaffinity/domain/port/outcoming/AffinityProductsIdsPort.java index dcfff64..6b3e739 100644 --- a/src/main/java/com/example/productaffinity/domain/port/outcoming/AffinityProductsIdsPort.java +++ b/src/main/java/com/example/productaffinity/domain/port/outcoming/AffinityProductsIdsPort.java @@ -2,8 +2,8 @@ package com.example.productaffinity.domain.port.outcoming; import reactor.core.publisher.Mono; -import java.util.Collection; +import java.util.List; public interface AffinityProductsIdsPort { - Mono> findAffinityProductIds(String productId); + Mono> findAffinityProductIds(String productId); } diff --git a/src/main/java/com/example/productaffinity/domain/service/AffinityProductsService.java b/src/main/java/com/example/productaffinity/domain/service/AffinityProductsService.java new file mode 100644 index 0000000..7958424 --- /dev/null +++ b/src/main/java/com/example/productaffinity/domain/service/AffinityProductsService.java @@ -0,0 +1,42 @@ +package com.example.productaffinity.domain.service; + +import com.example.productaffinity.domain.exception.ProductDetailUnavailableException; +import com.example.productaffinity.domain.model.Product; +import com.example.productaffinity.domain.port.incomming.GetAffinityProductsUC; +import com.example.productaffinity.domain.port.outcoming.AffinityProductsIdsPort; +import com.example.productaffinity.domain.port.outcoming.ProductDetailsPort; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.util.Collection; +import java.util.List; + +@Service +public class AffinityProductsService implements GetAffinityProductsUC { + + private final AffinityProductsIdsPort affinityProductsIdsPort; + private final ProductDetailsPort productDetailsPort; + + //Máximum number of detail req. + private static final int DETAIL_FETCH_CONCURRENCY = 10; + + public AffinityProductsService(final AffinityProductsIdsPort affinityProductsIdsPort, + final ProductDetailsPort productDetailsPort) { + this.affinityProductsIdsPort = affinityProductsIdsPort; + this.productDetailsPort = productDetailsPort; + } + + // Returns similar products in similarity order. If fails resolving the request, web propagates status + @Override + public Mono> getAffinityProducts(final String productId) { + return affinityProductsIdsPort.findAffinityProductIds(productId).flatMapMany(Flux::fromIterable) + .distinct().flatMapSequential(this::fetchDetailOrSkip, DETAIL_FETCH_CONCURRENCY) + .collectList(); + } + + private Mono fetchDetailOrSkip(String productId) { + return productDetailsPort.findProductDetail(productId) + .onErrorResume(ProductDetailUnavailableException.class, error -> Mono.empty()); + } +} diff --git a/src/main/java/com/example/productaffinity/domain/service/SimilarProductsService.java b/src/main/java/com/example/productaffinity/domain/service/SimilarProductsService.java deleted file mode 100644 index f166806..0000000 --- a/src/main/java/com/example/productaffinity/domain/service/SimilarProductsService.java +++ /dev/null @@ -1,20 +0,0 @@ -package com.example.productaffinity.domain.service; - -import com.example.productaffinity.domain.model.Product; -import com.example.productaffinity.domain.port.incomming.GetAffinityProductsUC; -import org.springframework.stereotype.Service; -import reactor.core.publisher.Mono; - -import java.util.List; - -@Service -public class SimilarProductsService implements GetAffinityProductsUC { - - public SimilarProductsService() { - } - - @Override - public Mono> getAffinityProducts(String productId) { - return null; - } -} diff --git a/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/ApiAffinityIdsAdapter.java b/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/ApiAffinityIdsAdapter.java new file mode 100644 index 0000000..e0b1331 --- /dev/null +++ b/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/ApiAffinityIdsAdapter.java @@ -0,0 +1,97 @@ +package com.example.productaffinity.infraestructure.adapter.outcomming.client; + +import com.example.productaffinity.domain.exception.ExternalServiceException; +import com.example.productaffinity.domain.exception.ProductNotFoundException; +import com.example.productaffinity.domain.port.outcoming.AffinityProductsIdsPort; +import io.github.resilience4j.bulkhead.Bulkhead; +import io.github.resilience4j.bulkhead.BulkheadRegistry; +import io.github.resilience4j.circuitbreaker.CircuitBreaker; +import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry; +import io.github.resilience4j.reactor.bulkhead.operator.BulkheadOperator; +import io.github.resilience4j.reactor.circuitbreaker.operator.CircuitBreakerOperator; +import io.github.resilience4j.reactor.retry.RetryOperator; +import io.github.resilience4j.reactor.timelimiter.TimeLimiterOperator; +import io.github.resilience4j.retry.Retry; +import io.github.resilience4j.retry.RetryRegistry; +import io.github.resilience4j.timelimiter.TimeLimiter; +import io.github.resilience4j.timelimiter.TimeLimiterRegistry; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.core.codec.DecodingException; +import org.springframework.stereotype.Component; +import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.reactive.function.client.WebClientResponseException; +import reactor.core.publisher.Mono; + +import java.util.Collection; +import java.util.List; +import java.util.Objects; + +@Component +public class ApiAffinityIdsAdapter implements AffinityProductsIdsPort { + + private static final Logger log = LoggerFactory.getLogger(ApiAffinityIdsAdapter.class); + private static final String RESILIENCE_INSTANCE = "affinityIds"; + + private final WebClient webClient; + private final TimeLimiter timeLimiter; + private final CircuitBreaker circuitBreaker; + private final Retry retry; + private final Bulkhead bulkhead; + private final RequestCoalescer> requestCoalescer; + + public ApiAffinityIdsAdapter( + final @Qualifier("affinityIdsWebClient") WebClient existingApiWebClient, + final TimeLimiterRegistry timeLimiterRegistry, + final CircuitBreakerRegistry circuitBreakerRegistry, + final RetryRegistry retryRegistry, + final BulkheadRegistry bulkheadRegistry + ) { + this.webClient = existingApiWebClient; + this.timeLimiter = timeLimiterRegistry.timeLimiter(RESILIENCE_INSTANCE); + this.circuitBreaker = circuitBreakerRegistry.circuitBreaker(RESILIENCE_INSTANCE); + this.retry = retryRegistry.retry(RESILIENCE_INSTANCE); + this.bulkhead = bulkheadRegistry.bulkhead(RESILIENCE_INSTANCE); + this.requestCoalescer = new RequestCoalescer<>(); + } + + @Override + public Mono> findAffinityProductIds(final String productId) { + return requestCoalescer.execute(productId, () -> fetchSimilarProductIds(productId)); + } + + private Mono> fetchSimilarProductIds(final String productId) { + Mono> call = webClient.get() + .uri("/product/{productId}/similarids", productId) + .retrieve() + .bodyToMono(new ParameterizedTypeReference>() { }) + .switchIfEmpty(Mono.error(new DecodingException("Empty similar product IDs response"))) + .map(this::validateSimilarProductIds) + .onErrorMap(WebClientResponseException.NotFound.class, e -> new ProductNotFoundException(productId)); + + return call.transformDeferred(TimeLimiterOperator.of(timeLimiter)) + .transformDeferred(CircuitBreakerOperator.of(circuitBreaker)) + .transformDeferred(BulkheadOperator.of(bulkhead)) + .transformDeferred(RetryOperator.of(retry)) + .onErrorMap(this::isUnexpectedError, + error -> new ExternalServiceException( + "Could not obtain similar product IDs for " + productId, error)) + .doOnError(ExternalServiceException.class, + error -> log.warn("similar_product_ids_unavailable productId={} cause={}", + productId, error.getCause().toString())); + } + + private List validateSimilarProductIds(List productIds) { + if (productIds.stream().anyMatch(Objects::isNull)) { + throw new DecodingException("Similar product IDs response contains null"); + } + return List.copyOf(productIds); + } + + private boolean isUnexpectedError(Throwable error) { + return !(error instanceof ProductNotFoundException); + } + +} diff --git a/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/ApiProductDetailAdapter.java b/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/ApiProductDetailAdapter.java new file mode 100644 index 0000000..5d59255 --- /dev/null +++ b/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/ApiProductDetailAdapter.java @@ -0,0 +1,45 @@ +package com.example.productaffinity.infraestructure.adapter.outcomming.client; + +import com.example.productaffinity.domain.model.Product; +import com.example.productaffinity.domain.port.outcoming.ProductDetailsPort; +import io.github.resilience4j.bulkhead.Bulkhead; +import io.github.resilience4j.bulkhead.BulkheadRegistry; +import io.github.resilience4j.retry.Retry; +import io.github.resilience4j.retry.RetryRegistry; +import io.github.resilience4j.timelimiter.TimeLimiter; +import io.github.resilience4j.timelimiter.TimeLimiterRegistry; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import org.springframework.web.reactive.function.client.WebClient; +import reactor.core.publisher.Mono; + +@Component +public class ApiProductDetailAdapter implements ProductDetailsPort { + + private static final String RESILIENCE_INSTANCE = "productDetail"; + + private final WebClient webClient; + private final TimeLimiter timeLimiter; + private final Retry retry; + private final Bulkhead bulkhead; + private final RequestCoalescer requestCoalescer; + + + public ApiProductDetailAdapter( + @Qualifier("productDetailWebClient") final WebClient existingApiWebClient, + final TimeLimiterRegistry timeLimiterRegistry, + final RetryRegistry retryRegistry, + final BulkheadRegistry bulkheadRegistry) { + + this.webClient = existingApiWebClient; + this.timeLimiter = timeLimiterRegistry.timeLimiter(RESILIENCE_INSTANCE); + this.retry = retryRegistry.retry(RESILIENCE_INSTANCE); + this.bulkhead = bulkheadRegistry.bulkhead(RESILIENCE_INSTANCE); + this.requestCoalescer = new RequestCoalescer<>(); + } + + @Override + public Mono findProductDetail(final String productId) { + return null; + } +} diff --git a/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/RequestCoalescer.java b/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/RequestCoalescer.java new file mode 100644 index 0000000..f22a116 --- /dev/null +++ b/src/main/java/com/example/productaffinity/infraestructure/adapter/outcomming/client/RequestCoalescer.java @@ -0,0 +1,21 @@ +package com.example.productaffinity.infraestructure.adapter.outcomming.client; + +import reactor.core.publisher.Mono; + +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.function.Supplier; + +//Shares a single active request per key and removes it inmediately when it finishes. +public class RequestCoalescer { + private final ConcurrentMap> requests = new ConcurrentHashMap<>(); + Mono execute(K key, Supplier> requestSupplier) { + return Mono.defer(() -> requests.computeIfAbsent( + key, ignored -> sharedRequest(key, requestSupplier))); + } + private Mono sharedRequest(K key, Supplier> requestSupplier) { + return Mono.defer(requestSupplier) + .doFinally(ignored -> requests.remove(key)) + .cache(); + } +} diff --git a/src/main/java/com/example/productaffinity/infraestructure/config/ApiProperties.java b/src/main/java/com/example/productaffinity/infraestructure/config/ApiProperties.java new file mode 100644 index 0000000..79d9a46 --- /dev/null +++ b/src/main/java/com/example/productaffinity/infraestructure/config/ApiProperties.java @@ -0,0 +1,24 @@ +package com.example.productaffinity.infraestructure.config; + +import jakarta.validation.Valid; +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.NotBlank; +import org.jetbrains.annotations.NotNull; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + +import java.time.Duration; + +@Validated +@ConfigurationProperties(prefix = "api") +public record ApiProperties( + @NotBlank String baseUrl, + @Valid @NotNull ClientProperties affinityIds, + @Valid @NotNull ClientProperties productDetail) { + public record ClientProperties( + @Min(1) int maxConnections, + @Min(1) int pendingAcquireMaxCount, + @NotNull Duration pendingAcquireTimeout, + @NotNull Duration responseTimeout) { + } +} diff --git a/src/main/java/com/example/productaffinity/infraestructure/config/WebClientConfig.java b/src/main/java/com/example/productaffinity/infraestructure/config/WebClientConfig.java new file mode 100644 index 0000000..dde108e --- /dev/null +++ b/src/main/java/com/example/productaffinity/infraestructure/config/WebClientConfig.java @@ -0,0 +1,61 @@ +package com.example.productaffinity.infraestructure.config; + +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.client.reactive.ReactorClientHttpConnector; +import org.springframework.web.reactive.function.client.WebClient; +import reactor.netty.http.client.HttpClient; +import reactor.netty.resources.ConnectionProvider; + +@Configuration(proxyBeanMethods = false) +@EnableConfigurationProperties(ApiProperties.class) +public class WebClientConfig { + + @Bean(name = "affinityIdsConnectionProvider", destroyMethod = "dispose") + public ConnectionProvider affinityIdsConnectionProvider(ApiProperties properties) { + return connectionProvider("similar-ids-pool", properties.affinityIds()); + } + + @Bean(name = "productDetailConnectionProvider", destroyMethod = "dispose") + public ConnectionProvider productDetailConnectionProvider(ApiProperties properties) { + return connectionProvider("product-detail-pool", properties.productDetail()); + } + + @Bean("affinityIdsWebClient") + public WebClient affinityIdsWebClient( + ApiProperties properties, + @Qualifier("affinityIdsConnectionProvider") ConnectionProvider connectionProvider) { + return webClient(properties.baseUrl(), properties.affinityIds(), connectionProvider); + } + + @Bean("productDetailWebClient") + public WebClient productDetailWebClient( + ApiProperties properties, + @Qualifier("productDetailConnectionProvider") ConnectionProvider connectionProvider) { + return webClient(properties.baseUrl(), properties.productDetail(), connectionProvider); + } + + private ConnectionProvider connectionProvider(final String name, + final ApiProperties.ClientProperties properties) { + return ConnectionProvider.builder(name) + .maxConnections(properties.maxConnections()) + .pendingAcquireMaxCount(properties.pendingAcquireMaxCount()) + .pendingAcquireTimeout(properties.pendingAcquireTimeout()) + .build(); + } + + private WebClient webClient( + final String baseUrl, + final ApiProperties.ClientProperties properties, + final ConnectionProvider connectionProvider) { + HttpClient httpClient = HttpClient.create(connectionProvider) + .responseTimeout(properties.responseTimeout()); + return WebClient.builder() + .baseUrl(baseUrl) + .clientConnector(new ReactorClientHttpConnector(httpClient)) + .build(); + } + +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index e69de29..c7f6b88 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -0,0 +1,12 @@ +api: + base-url: http://localhost:3001 + affinity-ids: + max-connections: 200 + pending-acquire-max-count: 200 + pending-acquire-timeout: 250ms + response-timeout: 3s + product-detail: + max-connections: 500 + pending-acquire-max-count: 500 + pending-acquire-timeout: 250ms + response-timeout: 3s \ No newline at end of file diff --git a/src/test/java/com/example/productaffinity/domain/service/AffinityProductsServiceTest.java b/src/test/java/com/example/productaffinity/domain/service/AffinityProductsServiceTest.java new file mode 100644 index 0000000..cc81478 --- /dev/null +++ b/src/test/java/com/example/productaffinity/domain/service/AffinityProductsServiceTest.java @@ -0,0 +1,124 @@ +package com.example.productaffinity.domain.service; + +import com.example.productaffinity.domain.exception.ExternalServiceException; +import com.example.productaffinity.domain.exception.ProductDetailUnavailableException; +import com.example.productaffinity.domain.exception.ProductNotFoundException; +import com.example.productaffinity.domain.model.Product; +import com.example.productaffinity.domain.port.outcoming.AffinityProductsIdsPort; +import com.example.productaffinity.domain.port.outcoming.ProductDetailsPort; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import java.math.BigDecimal; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + + +class AffinityProductsServiceTest { + + private AffinityProductsIdsPort affinityProductsIdsPort; + private ProductDetailsPort productDetailsPort; + private AffinityProductsService service; + + @BeforeEach + void setUp() { + affinityProductsIdsPort = mock(AffinityProductsIdsPort.class); + productDetailsPort = mock(ProductDetailsPort.class); + service = new AffinityProductsService(affinityProductsIdsPort, productDetailsPort); + } + + @Test + void returnsDetailsInSimilarityOrder() { + when(affinityProductsIdsPort.findAffinityProductIds("1")).thenReturn(Mono.just(List.of("2", "3", "4"))); + when(productDetailsPort.findProductDetail("2")).thenReturn(Mono.just(product("2"))); + when(productDetailsPort.findProductDetail("3")).thenReturn(Mono.just(product("3"))); + when(productDetailsPort.findProductDetail("4")).thenReturn(Mono.just(product("4"))); + + StepVerifier.create(service.getAffinityProducts("1")) + .assertNext(products -> assertEquals(List.of("2", "3", "4"), ids(products))) + .verifyComplete(); + } + + @Test + void removesDuplicateProductsWithoutChangingSimilarityOrder() { + when(affinityProductsIdsPort.findAffinityProductIds("1")) + .thenReturn(Mono.just(List.of("2", "3", "2"))); + when(productDetailsPort.findProductDetail("2")).thenReturn(Mono.just(product("2"))); + when(productDetailsPort.findProductDetail("3")).thenReturn(Mono.just(product("3"))); + + StepVerifier.create(service.getAffinityProducts("1")) + .assertNext(products -> assertEquals(List.of("2", "3"), ids(products))) + .verifyComplete(); + } + + @Test + void skipsItemsWhoseDetailFailsWithoutFailingTheWholeRequest() { + when(affinityProductsIdsPort.findAffinityProductIds("4")).thenReturn(Mono.just(List.of("1", "2", "5"))); + when(productDetailsPort.findProductDetail("1")).thenReturn(Mono.just(product("1"))); + when(productDetailsPort.findProductDetail("2")).thenReturn(Mono.just(product("2"))); + when(productDetailsPort.findProductDetail("5")) + .thenReturn(Mono.error(new ProductDetailUnavailableException("5", new RuntimeException("404")))); + + StepVerifier.create(service.getAffinityProducts("4")) + .assertNext(products -> assertEquals(List.of("1", "2"), ids(products))) + .verifyComplete(); + } + + @Test + void propagatesUnexpectedDetailFailure() { + when(affinityProductsIdsPort.findAffinityProductIds("1")).thenReturn(Mono.just(List.of("2"))); + when(productDetailsPort.findProductDetail("2")) + .thenReturn(Mono.error(new IllegalStateException("programming defect"))); + + StepVerifier.create(service.getAffinityProducts("1")) + .expectErrorMatches(error -> error instanceof IllegalStateException + && error.getMessage().equals("programming defect")) + .verify(); + } + + @Test + void skipsItemThatTimesOut() { + when(affinityProductsIdsPort.findAffinityProductIds("3")).thenReturn(Mono.just(List.of("100", "1000"))); + when(productDetailsPort.findProductDetail("100")).thenReturn(Mono.just(product("100"))); + when(productDetailsPort.findProductDetail("1000")) + .thenReturn(Mono.error(new ProductDetailUnavailableException( + "1000", new java.util.concurrent.TimeoutException("too slow")))); + + StepVerifier.create(service.getAffinityProducts("3")) + .assertNext(products -> assertEquals(List.of("100"), ids(products))) + .verifyComplete(); + } + + @Test + void propagatesProductNotFoundFromMainCall() { + when(affinityProductsIdsPort.findAffinityProductIds("999")) + .thenReturn(Mono.error(new ProductNotFoundException("999"))); + + StepVerifier.create(service.getAffinityProducts("999")) + .expectError(ProductNotFoundException.class) + .verify(); + } + + @Test + void propagatesExternalServiceFailureFromMainCall() { + when(affinityProductsIdsPort.findAffinityProductIds("1")) + .thenReturn(Mono.error(new ExternalServiceException("boom", new RuntimeException()))); + + StepVerifier.create(service.getAffinityProducts("1")) + .expectError(ExternalServiceException.class) + .verify(); + } + + private List ids(List products) { + return products.stream().map(Product::id).toList(); + } + + private Product product(String id) { + return new Product(id, "Product " + id, BigDecimal.TEN, true); + } +}