getProduct(int id);
/**
* Delete the product and all its relate reviews and recommendations from their repositories.
diff --git a/store-api/src/main/java/com/siriusxi/ms/store/api/core/product/ProductEndpoint.java b/store-api/src/main/java/com/siriusxi/ms/store/api/core/product/ProductEndpoint.java
index b5ae4d31..2c44fccb 100644
--- a/store-api/src/main/java/com/siriusxi/ms/store/api/core/product/ProductEndpoint.java
+++ b/store-api/src/main/java/com/siriusxi/ms/store/api/core/product/ProductEndpoint.java
@@ -2,6 +2,7 @@
import com.siriusxi.ms.store.api.core.product.dto.Product;
import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Mono;
import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
@@ -12,7 +13,7 @@
*
* @see ProductService
* @author mohamed.taman
- * @version v1.0
+ * @version v4.0
* @since v3.0 codename Storm
*/
@RequestMapping("products")
@@ -29,31 +30,5 @@ public interface ProductEndpoint extends ProductService {
*/
@Override
@GetMapping(value = "{productId}", produces = APPLICATION_JSON_VALUE)
- Product getProduct(@PathVariable("productId") int id);
-
- /**
- * Sample usage:
- *
- * curl -X POST $HOST:$PORT/products \ -H "Content-Type: application/json" --data \
- * '{"productId":123,"name":"product 123","weight":123}'
- *
- * @param body product to save.
- * @return Product just created.
- * @since v3.0 codename Storm
- */
- @Override
- @PostMapping(produces = APPLICATION_JSON_VALUE, consumes = APPLICATION_JSON_VALUE)
- Product createProduct(@RequestBody Product body);
-
- /**
- * Sample usage:
- *
- *
curl -X DELETE $HOST:$PORT/products/1
- *
- * @param id to be deleted.
- * @since v3.0 codename Storm
- */
- @Override
- @DeleteMapping("{productId}")
- void deleteProduct(@PathVariable("productId") int id);
+ Mono getProduct(@PathVariable("productId") int id);
}
diff --git a/store-api/src/main/java/com/siriusxi/ms/store/api/core/product/ProductService.java b/store-api/src/main/java/com/siriusxi/ms/store/api/core/product/ProductService.java
index 04a8b095..4a36ecce 100644
--- a/store-api/src/main/java/com/siriusxi/ms/store/api/core/product/ProductService.java
+++ b/store-api/src/main/java/com/siriusxi/ms/store/api/core/product/ProductService.java
@@ -1,6 +1,7 @@
package com.siriusxi.ms.store.api.core.product;
import com.siriusxi.ms.store.api.core.product.dto.Product;
+import reactor.core.publisher.Mono;
/**
* Interface that define the general service contract (methods) for the Product
@@ -18,21 +19,21 @@ public interface ProductService {
/**
* Get the product with Id from repository.
+ * It is a Non-Blocking API.
*
* @param id is the product id that you are looking for.
* @return the product, if found, else null.
* @since v0.1
*/
- Product getProduct(int id);
+ Mono getProduct(int id);
/**
* Add product to the repository.
*
* @param body product to save.
- * @return just created product.
* @since v0.1
*/
- Product createProduct(Product body);
+ default Product createProduct(Product body){ return null;}
/**
* Delete the product from repository.
@@ -41,5 +42,5 @@ public interface ProductService {
* @param id to be deleted.
* @since v0.1
*/
- void deleteProduct(int id);
+ default void deleteProduct(int id){}
}
diff --git a/store-api/src/main/java/com/siriusxi/ms/store/api/core/recommendation/RecommendationEndpoint.java b/store-api/src/main/java/com/siriusxi/ms/store/api/core/recommendation/RecommendationEndpoint.java
index 9fca72c5..a95e1c55 100644
--- a/store-api/src/main/java/com/siriusxi/ms/store/api/core/recommendation/RecommendationEndpoint.java
+++ b/store-api/src/main/java/com/siriusxi/ms/store/api/core/recommendation/RecommendationEndpoint.java
@@ -1,9 +1,10 @@
package com.siriusxi.ms.store.api.core.recommendation;
import com.siriusxi.ms.store.api.core.recommendation.dto.Recommendation;
-import org.springframework.web.bind.annotation.*;
-
-import java.util.List;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import reactor.core.publisher.Flux;
import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;
/**
@@ -13,7 +14,7 @@
*
* @see RecommendationService
* @author mohamed.taman
- * @version v1.0
+ * @version v4.0
* @since v3.0 codename Storm
*/
@RequestMapping("recommendations")
@@ -29,31 +30,5 @@ public interface RecommendationEndpoint extends RecommendationService {
* @since v3.0 codename Storm
*/
@GetMapping(produces = APPLICATION_JSON_VALUE)
- List getRecommendations(@RequestParam("productId") int productId);
-
- /**
- * Sample usage:
- *
- * curl -X POST $HOST:$PORT/recommendations \
- * -H "Content-Type: application/json" --data \
- * '{"productId":123,"recommendationId":456,"author":"me","rate":5,"content":"yada, yada, yada"
- * }'
- *
- * @param body the recommendation to add.
- * @return currently created recommendation.
- * @since v3.0 codename Storm
- */
- @PostMapping(produces = APPLICATION_JSON_VALUE, consumes = APPLICATION_JSON_VALUE)
- Recommendation createRecommendation(@RequestBody Recommendation body);
-
- /**
- * Sample usage:
- *
- *
curl -X DELETE $HOST:$PORT/recommendations?productId=1
- *
- * @param productId to delete recommendations for.
- * @since version 0.1
- */
- @DeleteMapping
- void deleteRecommendations(@RequestParam("productId") int productId);
+ Flux getRecommendations(@RequestParam("productId") int productId);
}
diff --git a/store-api/src/main/java/com/siriusxi/ms/store/api/core/recommendation/RecommendationService.java b/store-api/src/main/java/com/siriusxi/ms/store/api/core/recommendation/RecommendationService.java
index 2037a9b0..bf9c5c49 100644
--- a/store-api/src/main/java/com/siriusxi/ms/store/api/core/recommendation/RecommendationService.java
+++ b/store-api/src/main/java/com/siriusxi/ms/store/api/core/recommendation/RecommendationService.java
@@ -1,8 +1,7 @@
package com.siriusxi.ms.store.api.core.recommendation;
import com.siriusxi.ms.store.api.core.recommendation.dto.Recommendation;
-
-import java.util.List;
+import reactor.core.publisher.Flux;
/**
* Interface that define the general service contract (methods) for the Recommendation
@@ -18,13 +17,13 @@
*/
public interface RecommendationService {
/**
- * Get all recommendations for specific product by product id.
+ * Get all recommendations for specific product by product id. It is a Non-Blocking API.
*
* @param productId that you are looking for its recommendations.
* @return list of product recommendations, or empty list if there are no recommendations.
* @since v0.1
*/
- List getRecommendations(int productId);
+ Flux getRecommendations(int productId);
/**
* Create a new recommendation for a product.
@@ -33,7 +32,9 @@ public interface RecommendationService {
* @return currently created recommendation.
* @since v0.1
*/
- Recommendation createRecommendation(Recommendation body);
+ default Recommendation createRecommendation(Recommendation body) {
+ return null;
+ }
/**
* Delete all product recommendations.
@@ -41,5 +42,5 @@ public interface RecommendationService {
* @param productId to delete recommendations for.
* @since v0.1
*/
- void deleteRecommendations(int productId);
+ default void deleteRecommendations(int productId) {}
}
diff --git a/store-api/src/main/java/com/siriusxi/ms/store/api/core/review/ReviewEndpoint.java b/store-api/src/main/java/com/siriusxi/ms/store/api/core/review/ReviewEndpoint.java
index 48dbcaea..dd724f5e 100644
--- a/store-api/src/main/java/com/siriusxi/ms/store/api/core/review/ReviewEndpoint.java
+++ b/store-api/src/main/java/com/siriusxi/ms/store/api/core/review/ReviewEndpoint.java
@@ -2,6 +2,7 @@
import com.siriusxi.ms.store.api.core.review.dto.Review;
import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Flux;
import java.util.List;
@@ -14,27 +15,12 @@
*
* @see ReviewService
* @author mohamed.taman
- * @version v1.0
+ * @version v4.0
* @since v3.0 codename Storm
*/
@RequestMapping("reviews")
public interface ReviewEndpoint extends ReviewService {
- /**
- * Sample usage:
- *
- * curl -X POST $HOST:$PORT/reviews \
- * -H "Content-Type: application/json" --data \
- * '{"productId":123,"reviewId":456,"author":"me","subject":"yada, yada, yada",
- * "content":"yada, yada, yada"}'
- *
- * @param body review to be created.
- * @return just created review.
- * @since v3.0 codename Storm
- */
- @PostMapping(produces = APPLICATION_JSON_VALUE, consumes = APPLICATION_JSON_VALUE)
- Review createReview(@RequestBody Review body);
-
/**
* Sample usage:
*
@@ -45,16 +31,5 @@ public interface ReviewEndpoint extends ReviewService {
* @since v3.0 codename Storm
*/
@GetMapping(produces = APPLICATION_JSON_VALUE)
- List getReviews(@RequestParam("productId") int productId);
-
- /**
- * Sample usage:
- *
- * curl -X DELETE $HOST:$PORT/review?productId=1
- *
- * @param productId to delete its reviews.
- * @since v3.0 codename Storm
- */
- @DeleteMapping
- void deleteReviews(@RequestParam("productId") int productId);
+ Flux getReviews(@RequestParam("productId") int productId);
}
diff --git a/store-api/src/main/java/com/siriusxi/ms/store/api/core/review/ReviewService.java b/store-api/src/main/java/com/siriusxi/ms/store/api/core/review/ReviewService.java
index 7586ea26..20321f53 100644
--- a/store-api/src/main/java/com/siriusxi/ms/store/api/core/review/ReviewService.java
+++ b/store-api/src/main/java/com/siriusxi/ms/store/api/core/review/ReviewService.java
@@ -1,6 +1,7 @@
package com.siriusxi.ms.store.api.core.review;
import com.siriusxi.ms.store.api.core.review.dto.Review;
+import reactor.core.publisher.Flux;
import java.util.List;
@@ -13,18 +14,19 @@
*
*
* @author mohamed.taman
- * @version v0.2
+ * @version v4.0
* @since v0.1
*/
public interface ReviewService {
/**
* Get all reviews for specific product by product id.
+ * It is a Non-Blocking API.
*
* @param productId that you are looking for its reviews.
* @return list of reviews for this product, or empty list if there are no reviews.
*/
- List getReviews(int productId);
+ Flux getReviews(int productId);
/**
* Create a new review for a product.
@@ -32,12 +34,12 @@ public interface ReviewService {
* @param body review to be created.
* @return just created review.
*/
- Review createReview(Review body);
+ default Review createReview(Review body){return null;}
/**
* Delete all product reviews.
*
* @param productId to delete its reviews.
*/
- void deleteReviews(int productId);
+ default void deleteReviews(int productId){}
}
diff --git a/store-api/src/main/java/com/siriusxi/ms/store/api/event/Event.java b/store-api/src/main/java/com/siriusxi/ms/store/api/event/Event.java
new file mode 100644
index 00000000..462d379d
--- /dev/null
+++ b/store-api/src/main/java/com/siriusxi/ms/store/api/event/Event.java
@@ -0,0 +1,29 @@
+package com.siriusxi.ms.store.api.event;
+
+import lombok.*;
+
+import java.time.LocalDateTime;
+
+import static java.time.LocalDateTime.now;
+import static lombok.AccessLevel.NONE;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Setter(NONE)
+public class Event {
+
+ public enum Type {CREATE, DELETE}
+
+ private Event.Type eventType;
+ private K key;
+ private T data;
+ private LocalDateTime eventCreatedAt;
+
+ public Event(Type eventType, K key, T data) {
+ this.eventType = eventType;
+ this.key = key;
+ this.data = data;
+ this.eventCreatedAt = now();
+ }
+}
\ No newline at end of file
diff --git a/store-build-chassis/pom.xml b/store-build-chassis/pom.xml
index 989a9950..c0af229c 100644
--- a/store-build-chassis/pom.xml
+++ b/store-build-chassis/pom.xml
@@ -19,6 +19,7 @@
14
+ Hoxton.RELEASE
UTF-8
UTF-8
@@ -47,6 +48,15 @@
${springfox.swagger.version}
+
+
+ org.springframework.cloud
+ spring-cloud-dependencies
+ ${spring.cloud.version}
+ pom
+ import
+
+
@@ -66,6 +76,13 @@
spring-boot-starter-webflux
+
+
+ org.springframework.boot
+ spring-boot-properties-migrator
+ runtime
+
+
@@ -96,6 +113,10 @@
${maven.surefire.plugin.version}
--enable-preview
+
+ **/*Tests.java
+ **/*Test.java
+
diff --git a/store-service-chassis/pom.xml b/store-service-chassis/pom.xml
index 912d7bfc..bc0122db 100644
--- a/store-service-chassis/pom.xml
+++ b/store-service-chassis/pom.xml
@@ -23,7 +23,7 @@
-
+
org.springframework.boot
spring-boot-devtools
@@ -31,21 +31,35 @@
true
+
org.springframework.boot
spring-boot-configuration-processor
true
+
-
+
org.springframework.boot
spring-boot-starter-actuator
-
+
+
+ org.springframework.cloud
+ spring-cloud-starter-stream-rabbit
+
+
+
+
+ org.springframework.cloud
+ spring-cloud-starter-stream-kafka
+
+
+
@@ -65,11 +79,20 @@
test
+
io.projectreactor
reactor-test
test
+
+
+
+ org.springframework.cloud
+ spring-cloud-stream-test-support
+ test
+
+
diff --git a/store-service/src/main/java/com/siriusxi/ms/store/pcs/api/StoreController.java b/store-service/src/main/java/com/siriusxi/ms/store/pcs/api/StoreController.java
index 360eb8ff..e918b62d 100644
--- a/store-service/src/main/java/com/siriusxi/ms/store/pcs/api/StoreController.java
+++ b/store-service/src/main/java/com/siriusxi/ms/store/pcs/api/StoreController.java
@@ -7,6 +7,7 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.RestController;
+import reactor.core.publisher.Mono;
@RestController
@Log4j2
@@ -22,9 +23,10 @@ public StoreController(@Qualifier("StoreServiceImpl") StoreService storeService)
this.storeService = storeService;
}
- /** {@inheritDoc} */
+ /** {@inheritDoc}
+ * @return*/
@Override
- public ProductAggregate getProduct(int id) {
+ public Mono getProduct(int id) {
return storeService.getProduct(id);
}
diff --git a/store-service/src/main/java/com/siriusxi/ms/store/pcs/config/StoreConfiguration.java b/store-service/src/main/java/com/siriusxi/ms/store/pcs/config/StoreServiceConfiguration.java
similarity index 52%
rename from store-service/src/main/java/com/siriusxi/ms/store/pcs/config/StoreConfiguration.java
rename to store-service/src/main/java/com/siriusxi/ms/store/pcs/config/StoreServiceConfiguration.java
index d84591de..c683e55b 100644
--- a/store-service/src/main/java/com/siriusxi/ms/store/pcs/config/StoreConfiguration.java
+++ b/store-service/src/main/java/com/siriusxi/ms/store/pcs/config/StoreServiceConfiguration.java
@@ -1,6 +1,11 @@
package com.siriusxi.ms.store.pcs.config;
+import com.siriusxi.ms.store.pcs.integration.StoreIntegration;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.actuate.health.CompositeReactiveHealthContributor;
+import org.springframework.boot.actuate.health.ReactiveHealthContributor;
+import org.springframework.boot.actuate.health.ReactiveHealthIndicator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.client.RestTemplate;
@@ -9,39 +14,65 @@
import springfox.documentation.service.Contact;
import springfox.documentation.spring.web.plugins.Docket;
+import java.util.Map;
+
import static java.util.Collections.emptyList;
import static org.springframework.web.bind.annotation.RequestMethod.*;
import static springfox.documentation.builders.RequestHandlerSelectors.basePackage;
import static springfox.documentation.spi.DocumentationType.SWAGGER_2;
@Configuration
-public class StoreConfiguration {
+public class StoreServiceConfiguration {
+
+ private final StoreIntegration integration;
+
@Value("${api.common.version}")
- String apiVersion;
+ private String apiVersion;
@Value("${api.common.title}")
- String apiTitle;
+ private String apiTitle;
@Value("${api.common.description}")
- String apiDescription;
+ private String apiDescription;
@Value("${api.common.termsOfServiceUrl}")
- String apiTermsOfServiceUrl;
+ private String apiTermsOfServiceUrl;
@Value("${api.common.license}")
- String apiLicense;
+ private String apiLicense;
@Value("${api.common.licenseUrl}")
- String apiLicenseUrl;
+ private String apiLicenseUrl;
@Value("${api.common.contact.name}")
- String apiContactName;
+ private String apiContactName;
@Value("${api.common.contact.url}")
- String apiContactUrl;
+ private String apiContactUrl;
@Value("${api.common.contact.email}")
- String apiContactEmail;
+ private String apiContactEmail;
+
+ @Autowired
+ public StoreServiceConfiguration(StoreIntegration integration) {
+ this.integration = integration;
+ }
+
+ @Bean(name = "Core System Microservices")
+ ReactiveHealthContributor coreServices() {
+
+ ReactiveHealthIndicator productHealthIndicator = integration::getProductHealth;
+ ReactiveHealthIndicator recommendationHealthIndicator = integration::getRecommendationHealth;
+ ReactiveHealthIndicator reviewHealthIndicator = integration::getReviewHealth;
+
+ Map allIndicators =
+ Map.of(
+ "Product Service", productHealthIndicator,
+ "Recommendation Service", recommendationHealthIndicator,
+ "Review Service", reviewHealthIndicator);
+
+ return CompositeReactiveHealthContributor.fromMap(allIndicators);
+ }
@Bean
RestTemplate newRestClient() {
@@ -66,13 +97,17 @@ Using the apis() and paths() methods,
.paths(PathSelectors.any())
.build()
/*
- Using the globalResponseMessage() method, we ask SpringFox not to add any default HTTP response codes to the API documentation, such as 401 and 403, which we don't currently use.
+ Using the globalResponseMessage() method, we ask SpringFox not to add any default HTTP
+ response codes to the API documentation, such as 401 and 403,
+ which we don't currently use.
*/
.globalResponseMessage(POST, emptyList())
.globalResponseMessage(GET, emptyList())
.globalResponseMessage(DELETE, emptyList())
/*
- The api* variables that are used to configure the Docket bean with general information about the API are initialized from the property file using Spring @Value annotations.
+ The api* variables that are used to configure the Docket bean with general
+ information about the API are initialized from the property file using
+ Spring @Value annotations.
*/
.apiInfo(
new ApiInfo(
diff --git a/store-service/src/main/java/com/siriusxi/ms/store/pcs/integration/StoreIntegration.java b/store-service/src/main/java/com/siriusxi/ms/store/pcs/integration/StoreIntegration.java
index 58204b82..da4fc814 100644
--- a/store-service/src/main/java/com/siriusxi/ms/store/pcs/integration/StoreIntegration.java
+++ b/store-service/src/main/java/com/siriusxi/ms/store/pcs/integration/StoreIntegration.java
@@ -7,41 +7,49 @@
import com.siriusxi.ms.store.api.core.recommendation.dto.Recommendation;
import com.siriusxi.ms.store.api.core.review.ReviewService;
import com.siriusxi.ms.store.api.core.review.dto.Review;
+import com.siriusxi.ms.store.api.event.Event;
import com.siriusxi.ms.store.util.exceptions.InvalidInputException;
import com.siriusxi.ms.store.util.exceptions.NotFoundException;
import com.siriusxi.ms.store.util.http.HttpErrorInfo;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
-import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.boot.actuate.health.Health;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;
-import org.springframework.web.client.HttpClientErrorException;
-import org.springframework.web.client.RestTemplate;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClientResponseException;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
+import static com.siriusxi.ms.store.api.event.Event.Type.CREATE;
+import static com.siriusxi.ms.store.api.event.Event.Type.DELETE;
+import static com.siriusxi.ms.store.pcs.integration.StoreIntegration.MessageSources;
import static java.lang.String.valueOf;
-import static org.springframework.http.HttpMethod.GET;
+import static org.springframework.integration.support.MessageBuilder.withPayload;
+import static reactor.core.publisher.Flux.empty;
+@EnableBinding(MessageSources.class)
@Component
@Log4j2
public class StoreIntegration implements ProductService, RecommendationEndpoint, ReviewService {
public static final String PRODUCT_ID_QUERY_PARAM = "?productId=";
-
- private final RestTemplate restTemplate;
+ private final WebClient webClient;
private final ObjectMapper mapper;
-
+ private final MessageSources messageSources;
private final String productServiceUrl;
private final String recommendationServiceUrl;
private final String reviewServiceUrl;
-
@Autowired
public StoreIntegration(
- RestTemplate restTemplate,
+ WebClient.Builder webClient,
ObjectMapper mapper,
+ MessageSources messageSources,
@Value("${app.product-service.host}") String productServiceHost,
@Value("${app.product-service.port}") int productServicePort,
@Value("${app.recommendation-service.host}") String recommendationServiceHost,
@@ -49,204 +57,196 @@ public StoreIntegration(
@Value("${app.review-service.host}") String reviewServiceHost,
@Value("${app.review-service.port}") int reviewServicePort) {
- this.restTemplate = restTemplate;
+ this.webClient = webClient.build();
this.mapper = mapper;
+ this.messageSources = messageSources;
var http = "http://";
productServiceUrl =
http.concat(productServiceHost)
.concat(":")
- .concat(valueOf(productServicePort))
- .concat("/products/");
+ .concat(valueOf(productServicePort));
recommendationServiceUrl =
http.concat(recommendationServiceHost)
.concat(":")
- .concat(valueOf(recommendationServicePort))
- .concat("/recommendations");
+ .concat(valueOf(recommendationServicePort));
reviewServiceUrl =
http.concat(reviewServiceHost)
.concat(":")
- .concat(valueOf(reviewServicePort))
- .concat("/reviews");
+ .concat(valueOf(reviewServicePort));
}
@Override
public Product createProduct(Product body) {
-
- try {
- String url = productServiceUrl;
- log.debug("Will post a new product to URL: {}", url);
-
- Product product = restTemplate.postForObject(url, body, Product.class);
- log.debug("Created a product with id: {}", product != null ? product.getProductId() : -1);
-
- return product;
-
- } catch (HttpClientErrorException ex) {
- throw handleHttpClientException(ex);
- }
+ log.debug("Publishing a create event for a new product {}",body.toString());
+ messageSources
+ .outputProducts()
+ .send(withPayload(new Event<>(CREATE, body.getProductId(), body)).build());
+ return body;
}
@Override
- public Product getProduct(int productId) {
-
- try {
- String url = productServiceUrl + "/" + productId;
- log.debug("Will call the getProduct API on URL: {}", url);
+ public Mono getProduct(int productId) {
- Product product = restTemplate.getForObject(url, Product.class);
- log.debug("Found a product with id: {}", product != null ? product.getProductId() : -1);
+ var url = productServiceUrl
+ .concat("/products/")
+ .concat(valueOf(productId));
- return product;
+ log.debug("Will call the getProduct API on URL: {}", url);
- } catch (HttpClientErrorException ex) {
- throw handleHttpClientException(ex);
- }
+ return webClient
+ .get()
+ .uri(url)
+ .retrieve()
+ .bodyToMono(Product.class)
+ .log()
+ .onErrorMap(WebClientResponseException.class, this::handleException);
}
@Override
public void deleteProduct(int productId) {
- try {
- String url = productServiceUrl + "/" + productId;
- log.debug("Will call the deleteProduct API on URL: {}", url);
-
- restTemplate.delete(url);
-
- } catch (HttpClientErrorException ex) {
- throw handleHttpClientException(ex);
- }
+ log.debug("Publishing a delete event for product id {}", productId);
+ messageSources
+ .outputProducts()
+ .send(withPayload(new Event<>(DELETE, productId, null)).build());
}
@Override
public Recommendation createRecommendation(Recommendation body) {
+ log.debug("Publishing a create event for a new recommendation {}",body.toString());
- try {
- String url = recommendationServiceUrl;
- log.debug("Will post a new recommendation to URL: {}", url);
-
- Recommendation recommendation = restTemplate.postForObject(url, body, Recommendation.class);
- log.debug("Created a recommendation with id: {}",
- recommendation != null ? recommendation.getRecommendationId() : -1);
-
- return recommendation;
+ messageSources
+ .outputRecommendations()
+ .send(withPayload(new Event<>(CREATE, body.getProductId(), body)).build());
- } catch (HttpClientErrorException ex) {
- throw handleHttpClientException(ex);
- }
+ return body;
}
@Override
- public List getRecommendations(int productId) {
-
- try {
- String url = recommendationServiceUrl.concat(PRODUCT_ID_QUERY_PARAM).concat(valueOf(productId));
-
- log.debug("Will call the getRecommendations API on URL: {}", url);
- List recommendations =
- restTemplate
- .exchange(url, GET, null, new ParameterizedTypeReference>() {})
- .getBody();
-
- log.debug(
- "Found {} recommendations for a product with id: {}", recommendations != null ? recommendations.size() : 0, productId);
- return recommendations;
-
- } catch (Exception ex) {
- log.warn(
- "Got an exception while requesting recommendations, return zero recommendations: {}",
- ex.getMessage());
- return new ArrayList<>();
- }
+ public Flux getRecommendations(int productId) {
+
+ var url = recommendationServiceUrl
+ .concat("/recommendations")
+ .concat(PRODUCT_ID_QUERY_PARAM)
+ .concat(valueOf(productId));
+
+ log.debug("Will call the getRecommendations API on URL: {}", url);
+
+ /* Return an empty result if something goes wrong to make it possible
+ for the composite service to return partial responses
+ */
+ return webClient
+ .get()
+ .uri(url)
+ .retrieve()
+ .bodyToFlux(Recommendation.class)
+ .log()
+ .onErrorResume(error -> empty());
}
@Override
public void deleteRecommendations(int productId) {
- try {
- String url = recommendationServiceUrl
- .concat(PRODUCT_ID_QUERY_PARAM)
- .concat(valueOf(productId));
- log.debug("Will call the deleteRecommendations API on URL: {}", url);
-
- restTemplate.delete(url);
-
- } catch (HttpClientErrorException ex) {
- throw handleHttpClientException(ex);
- }
+ messageSources
+ .outputRecommendations()
+ .send(withPayload(new Event<>(DELETE, productId, null)).build());
}
@Override
public Review createReview(Review body) {
+ messageSources
+ .outputReviews()
+ .send(withPayload(new Event<>(CREATE, body.getProductId(), body)).build());
+ return body;
+ }
- try {
- String url = reviewServiceUrl;
- log.debug("Will post a new review to URL: {}", url);
+ @Override
+ public Flux getReviews(int productId) {
- var review = restTemplate.postForObject(url, body, Review.class);
- log.debug("Created a review with id: {}", review != null ? review.getProductId() : 0);
+ var url = reviewServiceUrl
+ .concat("/reviews")
+ .concat(PRODUCT_ID_QUERY_PARAM)
+ .concat(valueOf(productId));
- return review;
+ log.debug("Will call the getReviews API on URL: {}", url);
+
+ /* Return an empty result if something goes wrong to make it possible
+ for the composite service to return partial responses
+ */
+ return webClient
+ .get()
+ .uri(url)
+ .retrieve()
+ .bodyToFlux(Review.class).log()
+ .onErrorResume(error -> empty());
- } catch (HttpClientErrorException ex) {
- throw handleHttpClientException(ex);
- }
}
@Override
- public List getReviews(int productId) {
+ public void deleteReviews(int productId) {
+ messageSources
+ .outputReviews()
+ .send(withPayload(new Event<>(DELETE, productId, null)).build());
+ }
- try {
- String url = reviewServiceUrl
- .concat(PRODUCT_ID_QUERY_PARAM)
- .concat(valueOf(productId));
-
- log.debug("Will call the getReviews API on URL: {}", url);
- List reviews =
- restTemplate
- .exchange(url, GET, null, new ParameterizedTypeReference>() {})
- .getBody();
-
- log.debug("Found {} reviews for a product with id: {}", reviews != null ? reviews.size() : 0, productId);
- return reviews;
-
- } catch (Exception ex) {
- log.warn(
- "Got an exception while requesting reviews, return zero reviews: {}", ex.getMessage());
- return new ArrayList<>();
- }
+ public Mono getProductHealth() {
+ return getHealth(productServiceUrl);
}
- @Override
- public void deleteReviews(int productId) {
- try {
- String url = reviewServiceUrl
- .concat(PRODUCT_ID_QUERY_PARAM)
- .concat(valueOf(productId));
- log.debug("Will call the deleteReviews API on URL: {}", url);
+ public Mono getRecommendationHealth() {
+ return getHealth(recommendationServiceUrl);
+ }
+
+ public Mono getReviewHealth() {
+ return getHealth(reviewServiceUrl);
+ }
- restTemplate.delete(url);
+ private Mono getHealth(String url) {
+ url += "/actuator/health";
+ log.debug("Will call the Health API on URL: {}", url);
+ return webClient.get().uri(url).retrieve().bodyToMono(String.class)
+ .map(s -> new Health.Builder().up().build())
+ .onErrorResume(ex -> Mono.just(new Health.Builder().down(ex).build()))
+ .log();
+ }
- } catch (HttpClientErrorException ex) {
- throw handleHttpClientException(ex);
+ private Throwable handleException(Throwable ex) {
+ if (!(ex instanceof WebClientResponseException wcre)) {
+ log.warn("Got a unexpected error: {}, will rethrow it", ex.toString());
+ return ex;
}
- }
- private RuntimeException handleHttpClientException(HttpClientErrorException ex) {
- return switch (ex.getStatusCode()) {
- case NOT_FOUND -> new NotFoundException(getErrorMessage(ex));
- case UNPROCESSABLE_ENTITY -> new InvalidInputException(getErrorMessage(ex));
+ return switch (wcre.getStatusCode()) {
+ case NOT_FOUND -> new NotFoundException(getErrorMessage(wcre));
+ case UNPROCESSABLE_ENTITY -> new InvalidInputException(getErrorMessage(wcre));
default -> {
- log.warn("Got a unexpected HTTP error: {}, will rethrow it", ex.getStatusCode());
- log.warn("Error body: {}", ex.getResponseBodyAsString());
- throw ex;}
+ log.warn("Got a unexpected HTTP error: {}, will rethrow it", wcre.getStatusCode());
+ log.warn("Error body: {}", wcre.getResponseBodyAsString());
+ throw wcre;}
};
}
- private String getErrorMessage(HttpClientErrorException ex) {
+ private String getErrorMessage(WebClientResponseException ex) {
try {
return mapper.readValue(ex.getResponseBodyAsString(), HttpErrorInfo.class).message();
} catch (IOException ioException) {
return ex.getMessage();
}
}
+
+ public interface MessageSources {
+
+ String OUTPUT_PRODUCTS = "output-products";
+ String OUTPUT_RECOMMENDATIONS = "output-recommendations";
+ String OUTPUT_REVIEWS = "output-reviews";
+
+ @Output(OUTPUT_PRODUCTS)
+ MessageChannel outputProducts();
+
+ @Output(OUTPUT_RECOMMENDATIONS)
+ MessageChannel outputRecommendations();
+
+ @Output(OUTPUT_REVIEWS)
+ MessageChannel outputReviews();
+ }
}
diff --git a/store-service/src/main/java/com/siriusxi/ms/store/pcs/service/StoreServiceImpl.java b/store-service/src/main/java/com/siriusxi/ms/store/pcs/service/StoreServiceImpl.java
index 1a2a5f64..d8ea44d8 100644
--- a/store-service/src/main/java/com/siriusxi/ms/store/pcs/service/StoreServiceImpl.java
+++ b/store-service/src/main/java/com/siriusxi/ms/store/pcs/service/StoreServiceImpl.java
@@ -9,11 +9,11 @@
import com.siriusxi.ms.store.api.core.recommendation.dto.Recommendation;
import com.siriusxi.ms.store.api.core.review.dto.Review;
import com.siriusxi.ms.store.pcs.integration.StoreIntegration;
-import com.siriusxi.ms.store.util.exceptions.NotFoundException;
import com.siriusxi.ms.store.util.http.ServiceUtil;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
import java.util.List;
import java.util.stream.Collectors;
@@ -37,16 +37,17 @@ public void createProduct(ProductAggregate body) {
try {
log.debug(
- "createCompositeProduct: creates a new composite entity for id: {}", body.getProductId());
+ "createCompositeProduct: creates a new composite entity for productId: {}",
+ body.getProductId());
- Product product = new Product(body.getProductId(), body.getName(), body.getWeight(), null);
+ var product = new Product(body.getProductId(), body.getName(), body.getWeight(), null);
integration.createProduct(product);
if (body.getRecommendations() != null) {
body.getRecommendations()
.forEach(
r -> {
- Recommendation recommendation =
+ var recommendation =
new Recommendation(
body.getProductId(),
r.getRecommendationId(),
@@ -73,45 +74,49 @@ public void createProduct(ProductAggregate body) {
integration.createReview(review);
});
}
-
log.debug(
- "createCompositeProduct: composite entites created for id: {}", body.getProductId());
+ "createCompositeProduct: composite entities created for productId: {}",
+ body.getProductId());
} catch (RuntimeException re) {
- log.warn("createCompositeProduct failed", re);
+ log.warn("createCompositeProduct failed: {}", re.toString());
throw re;
}
}
@Override
- public ProductAggregate getProduct(int id) {
- log.debug("getCompositeProduct: lookup a product aggregate for id: {}", id);
-
- Product product = integration.getProduct(id);
- if (product == null) throw new NotFoundException("No product found for id: " + id);
-
- List recommendations = integration.getRecommendations(id);
-
- List reviews = integration.getReviews(id);
-
- log.debug("getCompositeProduct: aggregate entity found for id: {}", id);
-
- return createProductAggregate(
- product, recommendations, reviews, serviceUtil.getServiceAddress());
+ public Mono getProduct(int productId) {
+ return Mono.zip(
+ values ->
+ createProductAggregate(
+ (Product) values[0],
+ (List) values[1],
+ (List) values[2],
+ serviceUtil.getServiceAddress()),
+ integration.getProduct(productId),
+ integration.getRecommendations(productId).collectList(),
+ integration.getReviews(productId).collectList())
+ .doOnError(ex -> log.warn("getCompositeProduct failed: {}", ex.toString()))
+ .log();
}
@Override
- public void deleteProduct(int id) {
+ public void deleteProduct(int productId) {
- log.debug("deleteCompositeProduct: Deletes a product aggregate for id: {}", id);
+ try {
- integration.deleteProduct(id);
+ log.debug("deleteCompositeProduct: Deletes a product aggregate for productId: {}", productId);
- integration.deleteRecommendations(id);
+ integration.deleteProduct(productId);
+ integration.deleteRecommendations(productId);
+ integration.deleteReviews(productId);
- integration.deleteReviews(id);
+ log.debug("deleteCompositeProduct: aggregate entities deleted for productId: {}", productId);
- log.debug("getCompositeProduct: aggregate entities deleted for id: {}", id);
+ } catch (RuntimeException re) {
+ log.warn("deleteCompositeProduct failed: {}", re.toString());
+ throw re;
+ }
}
private ProductAggregate createProductAggregate(
@@ -121,7 +126,7 @@ private ProductAggregate createProductAggregate(
String serviceAddress) {
// 1. Setup product info
- int id = product.getProductId();
+ int productId = product.getProductId();
String name = product.getName();
int weight = product.getWeight();
@@ -150,15 +155,15 @@ private ProductAggregate createProductAggregate(
// 4. Create info regarding the involved microservices addresses
String productAddress = product.getServiceAddress();
String reviewAddress =
- (reviews != null && !reviews.isEmpty()) ? reviews.get(0).getServiceAddress() : "";
+ (reviews != null && reviews.size() > 0) ? reviews.get(0).getServiceAddress() : "";
String recommendationAddress =
- (recommendations != null && !recommendations.isEmpty())
+ (recommendations != null && recommendations.size() > 0)
? recommendations.get(0).getServiceAddress()
: "";
ServiceAddresses serviceAddresses =
new ServiceAddresses(serviceAddress, productAddress, reviewAddress, recommendationAddress);
return new ProductAggregate(
- id, name, weight, recommendationSummaries, reviewSummaries, serviceAddresses);
+ productId, name, weight, recommendationSummaries, reviewSummaries, serviceAddresses);
}
}
diff --git a/store-service/src/main/resources/application.yaml b/store-service/src/main/resources/application.yaml
index 2e7c1672..fba8a697 100644
--- a/store-service/src/main/resources/application.yaml
+++ b/store-service/src/main/resources/application.yaml
@@ -2,6 +2,35 @@ spring:
application:
name: store-service
+ cloud:
+ stream:
+ default-binder: rabbit
+ default:
+ contentType: application/json
+ bindings:
+ output-products:
+ destination: products
+ producer:
+ required-groups: auditGroup
+ output-recommendations:
+ destination: recommendations
+ producer:
+ required-groups: auditGroup
+ output-reviews:
+ destination: reviews
+ producer:
+ required-groups: auditGroup
+ kafka:
+ binder:
+ brokers: 127.0.0.1
+ defaultBrokerPort: 9092
+
+ rabbitmq:
+ host: 127.0.0.1
+ port: 5672
+ username: guest
+ password: guest
+
server:
port: 9080
@@ -12,6 +41,9 @@ logging:
com.siriusxi.ms.store: DEBUG
management:
+ info:
+ git:
+ enabled: true
endpoints:
web:
exposure:
@@ -19,6 +51,8 @@ management:
endpoint:
shutdown:
enabled: true
+ health:
+ show-details: "ALWAYS"
# Custom configurations
app:
@@ -35,8 +69,16 @@ app:
# Swagger properties
api:
common:
- version: 1.0.0
- title: Springy Store μServices
+ version: 4.0
+ title: "Springy Store μServices"
+ termsOfServiceUrl: https://mohamed-taman.github.io/Springy-Store-Microservices/
+ license: "MIT License"
+ licenseUrl: "https://github.com/mohamed-taman/Springy-Store-Microservices/blob/master/LICENSE"
+
+ contact:
+ name: "Mohamed Taman"
+ url: "https://twitter.com/_tamanm"
+ email: "mohamed.taman@mail.com"
description: |
**Springy Store** is a conceptual simple μServices-based project using the latest
cutting-edge technologies, to demonstrate how the store is created to be a
@@ -44,14 +86,6 @@ api:
This project μServices are developed based on Spring Boot & Cloud framework, that implement
**cloud-native** intuitive, **design patterns** and **best practices**.
- termsOfServiceUrl: https://mohamed-taman.github.io/Springy-Store-Microservices/
- license: MIT License
- licenseUrl: https://github.com/mohamed-taman/Springy-Store-Microservices/blob/master/LICENSE
-
- contact:
- name: Mohamed Taman
- url: https://twitter.com/_tamanm
- email: mohamed.taman@mail.com
product-composite:
get-composite-product:
@@ -96,6 +130,13 @@ spring:
profiles: docker
jmx:
enabled: false
+ rabbitmq:
+ host: rabbitmq
+ cloud:
+ stream:
+ kafka:
+ binder:
+ brokers: kafka
server:
port: 8080
diff --git a/store-service/src/test/java/com/siriusxi/ms/store/pcs/IsSameEvent.java b/store-service/src/test/java/com/siriusxi/ms/store/pcs/IsSameEvent.java
new file mode 100644
index 00000000..fb29bd2e
--- /dev/null
+++ b/store-service/src/test/java/com/siriusxi/ms/store/pcs/IsSameEvent.java
@@ -0,0 +1,81 @@
+package com.siriusxi.ms.store.pcs;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.siriusxi.ms.store.api.event.Event;
+import lombok.extern.log4j.Log4j2;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.hamcrest.TypeSafeMatcher;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+@Log4j2
+class IsSameEvent extends TypeSafeMatcher {
+
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ private Event expectedEvent;
+
+
+ private IsSameEvent(Event expectedEvent) {
+ this.expectedEvent = expectedEvent;
+ }
+
+ @Override
+ protected boolean matchesSafely(String eventAsJson) {
+
+ if (expectedEvent == null) return false;
+
+ log.trace("Convert the following json string to a map: {}", eventAsJson);
+ Map mapEvent = convertJsonStringToMap(eventAsJson);
+ mapEvent.remove("eventCreatedAt");
+
+ Map mapExpectedEvent = getMapWithoutCreatedAt(expectedEvent);
+
+ log.trace("Got the map: {}", mapEvent);
+ log.trace("Compare to the expected map: {}", mapExpectedEvent);
+ return mapEvent.equals(mapExpectedEvent);
+ }
+
+ @Override
+ public void describeTo(Description description) {
+ String expectedJson = convertObjectToJsonString(expectedEvent);
+ description.appendText("expected to look like " + expectedJson);
+ }
+
+ public static Matcher sameEventExceptCreatedAt(Event expectedEvent) {
+ return new IsSameEvent(expectedEvent);
+ }
+
+ private Map getMapWithoutCreatedAt(Event event) {
+ Map mapEvent = convertObjectToMap(event);
+ mapEvent.remove("eventCreatedAt");
+ return mapEvent;
+ }
+
+ private Map convertObjectToMap(Object object) {
+ JsonNode node = mapper.convertValue(object, JsonNode.class);
+ return mapper.convertValue(node, Map.class);
+ }
+
+ private String convertObjectToJsonString(Object object) {
+ try {
+ return mapper.writeValueAsString(object);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map convertJsonStringToMap(String eventAsJson) {
+ try {
+ return mapper.readValue(eventAsJson, new TypeReference(){});
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
\ No newline at end of file
diff --git a/store-service/src/test/java/com/siriusxi/ms/store/pcs/IsSameEventTests.java b/store-service/src/test/java/com/siriusxi/ms/store/pcs/IsSameEventTests.java
new file mode 100644
index 00000000..debdfe79
--- /dev/null
+++ b/store-service/src/test/java/com/siriusxi/ms/store/pcs/IsSameEventTests.java
@@ -0,0 +1,38 @@
+package com.siriusxi.ms.store.pcs;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.siriusxi.ms.store.api.core.product.dto.Product;
+import com.siriusxi.ms.store.api.event.Event;
+import org.junit.jupiter.api.Test;
+
+import static com.siriusxi.ms.store.api.event.Event.Type.CREATE;
+import static com.siriusxi.ms.store.api.event.Event.Type.DELETE;
+import static com.siriusxi.ms.store.pcs.IsSameEvent.sameEventExceptCreatedAt;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+
+class IsSameEventTests {
+
+ ObjectMapper mapper = new ObjectMapper();
+
+ @Test
+ public void testEventObjectCompare() throws JsonProcessingException {
+
+ /*
+ Event #1 and #2 are the same event, but occurs as different times
+ Event #3 and #4 are different events
+ */
+ Event event1 = new Event<>(CREATE, 1, new Product(1, "name", 1, null));
+ Event event2 = new Event<>(CREATE, 1, new Product(1, "name", 1, null));
+ Event event3 = new Event<>(DELETE, 1, null);
+ Event event4 = new Event<>(CREATE, 1, new Product(2, "name", 1, null));
+
+ String event1JSon = mapper.writeValueAsString(event1);
+
+ assertThat(event1JSon, is(sameEventExceptCreatedAt(event2)));
+ assertThat(event1JSon, not(sameEventExceptCreatedAt(event3)));
+ assertThat(event1JSon, not(sameEventExceptCreatedAt(event4)));
+ }
+}
diff --git a/store-service/src/test/java/com/siriusxi/ms/store/pcs/MessagingTests.java b/store-service/src/test/java/com/siriusxi/ms/store/pcs/MessagingTests.java
new file mode 100644
index 00000000..90a279f9
--- /dev/null
+++ b/store-service/src/test/java/com/siriusxi/ms/store/pcs/MessagingTests.java
@@ -0,0 +1,186 @@
+package com.siriusxi.ms.store.pcs;
+
+import com.siriusxi.ms.store.api.composite.dto.ProductAggregate;
+import com.siriusxi.ms.store.api.composite.dto.RecommendationSummary;
+import com.siriusxi.ms.store.api.composite.dto.ReviewSummary;
+import com.siriusxi.ms.store.api.core.product.dto.Product;
+import com.siriusxi.ms.store.api.core.recommendation.dto.Recommendation;
+import com.siriusxi.ms.store.api.core.review.dto.Review;
+import com.siriusxi.ms.store.api.event.Event;
+import com.siriusxi.ms.store.pcs.integration.StoreIntegration;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.cloud.stream.test.binder.MessageCollector;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.test.web.reactive.server.WebTestClient;
+import reactor.core.publisher.Mono;
+
+import java.util.concurrent.BlockingQueue;
+
+import static com.siriusxi.ms.store.api.event.Event.Type.CREATE;
+import static com.siriusxi.ms.store.api.event.Event.Type.DELETE;
+import static com.siriusxi.ms.store.pcs.IsSameEvent.sameEventExceptCreatedAt;
+import static java.lang.String.valueOf;
+import static java.util.Collections.singletonList;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
+import static org.springframework.cloud.stream.test.matcher.MessageQueueMatcher.receivesPayloadThat;
+import static org.springframework.http.HttpStatus.OK;
+
+@SpringBootTest(webEnvironment = RANDOM_PORT)
+class MessagingTests {
+
+ public static final String BASE_URL = "/store/api/v1/products/";
+
+ BlockingQueue> queueProducts = null;
+ BlockingQueue> queueRecommendations = null;
+ BlockingQueue> queueReviews = null;
+
+ @Autowired private WebTestClient client;
+ @Autowired private StoreIntegration.MessageSources channels;
+ @Autowired private MessageCollector collector;
+
+ @BeforeEach
+ public void setUp() {
+ queueProducts = getQueue(channels.outputProducts());
+ queueRecommendations = getQueue(channels.outputRecommendations());
+ queueReviews = getQueue(channels.outputReviews());
+ }
+
+ @Test
+ public void createCompositeProduct1() {
+
+ ProductAggregate composite = new ProductAggregate(1, "name", 1, null, null, null);
+ postAndVerifyProduct(composite);
+
+ // Assert one expected new product events queued up
+ assertEquals(1, queueProducts.size());
+
+ Event expectedEvent =
+ new Event<>(
+ CREATE,
+ composite.getProductId(),
+ new Product(
+ composite.getProductId(), composite.getName(), composite.getWeight(), null));
+ assertThat(queueProducts, is(receivesPayloadThat(sameEventExceptCreatedAt(expectedEvent))));
+
+ // Assert none recommendations and review events
+ assertEquals(0, queueRecommendations.size());
+ assertEquals(0, queueReviews.size());
+ }
+
+ @Test
+ public void createCompositeProduct2() {
+
+ ProductAggregate composite =
+ new ProductAggregate(
+ 1,
+ "name",
+ 1,
+ singletonList(new RecommendationSummary(1, "a", 1, "c")),
+ singletonList(new ReviewSummary(1, "a", "s", "c")),
+ null);
+
+ postAndVerifyProduct(composite);
+
+ // Assert one create product event queued up
+ assertEquals(1, queueProducts.size());
+
+ Event expectedProductEvent =
+ new Event<>(
+ CREATE,
+ composite.getProductId(),
+ new Product(
+ composite.getProductId(), composite.getName(), composite.getWeight(), null));
+ assertThat(queueProducts, receivesPayloadThat(sameEventExceptCreatedAt(expectedProductEvent)));
+
+ // Assert one create recommendation event queued up
+ assertEquals(1, queueRecommendations.size());
+
+ RecommendationSummary rec = composite.getRecommendations().get(0);
+ Event expectedRecommendationEvent =
+ new Event<>(
+ CREATE,
+ composite.getProductId(),
+ new Recommendation(
+ composite.getProductId(),
+ rec.getRecommendationId(),
+ rec.getAuthor(),
+ rec.getRate(),
+ rec.getContent(),
+ null));
+ assertThat(
+ queueRecommendations,
+ receivesPayloadThat(sameEventExceptCreatedAt(expectedRecommendationEvent)));
+
+ // Assert one create review event queued up
+ assertEquals(1, queueReviews.size());
+
+ ReviewSummary rev = composite.getReviews().get(0);
+ Event expectedReviewEvent =
+ new Event<>(
+ CREATE,
+ composite.getProductId(),
+ new Review(
+ composite.getProductId(),
+ rev.getReviewId(),
+ rev.getAuthor(),
+ rev.getSubject(),
+ rev.getContent(),
+ null));
+
+ assertThat(queueReviews, receivesPayloadThat(sameEventExceptCreatedAt(expectedReviewEvent)));
+ }
+
+ @Test
+ public void deleteCompositeProduct() {
+
+ deleteAndVerifyProduct(1);
+
+ // Assert one delete product event queued up
+ assertEquals(1, queueProducts.size());
+
+ Event expectedEvent = new Event<>(DELETE, 1, null);
+
+ assertThat(queueProducts, is(receivesPayloadThat(sameEventExceptCreatedAt(expectedEvent))));
+
+ // Assert one delete recommendation event queued up
+ assertEquals(1, queueRecommendations.size());
+
+ Event expectedRecommendationEvent = new Event<>(DELETE, 1, null);
+ assertThat(
+ queueRecommendations,
+ receivesPayloadThat(sameEventExceptCreatedAt(expectedRecommendationEvent)));
+
+ // Assert one delete review event queued up
+ assertEquals(1, queueReviews.size());
+
+ Event expectedReviewEvent = new Event<>(DELETE, 1, null);
+ assertThat(queueReviews, receivesPayloadThat(sameEventExceptCreatedAt(expectedReviewEvent)));
+ }
+
+ private BlockingQueue> getQueue(MessageChannel messageChannel) {
+ return collector.forChannel(messageChannel);
+ }
+
+ private void postAndVerifyProduct(ProductAggregate compositeProduct) {
+ client
+ .post()
+ .uri(BASE_URL)
+ .body(Mono.just(compositeProduct), ProductAggregate.class)
+ .exchange()
+ .expectStatus().isEqualTo(OK);
+ }
+
+ private void deleteAndVerifyProduct(int productId) {
+ client.delete()
+ .uri(BASE_URL.concat(valueOf(productId)))
+ .exchange()
+ .expectStatus().isEqualTo(OK);
+ }
+}
diff --git a/store-service/src/test/java/com/siriusxi/ms/store/pcs/ReactorTests.java b/store-service/src/test/java/com/siriusxi/ms/store/pcs/ReactorTests.java
new file mode 100644
index 00000000..65d3ceb2
--- /dev/null
+++ b/store-service/src/test/java/com/siriusxi/ms/store/pcs/ReactorTests.java
@@ -0,0 +1,39 @@
+package com.siriusxi.ms.store.pcs;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Flux;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ReactorTests {
+
+ @Test
+ public void TestFlux() {
+
+ List list = new ArrayList<>();
+
+ Flux.just(1, 2, 3, 4)
+ .filter(n -> n % 2 == 0)
+ .map(n -> n * 2)
+ .log()
+ .subscribe(list::add);
+
+ assertThat(list).containsExactly(4, 8);
+ }
+
+ @Test
+ public void TestFluxBlocking() {
+
+ List list = Flux.just(1, 2, 3, 4)
+ .filter(n -> n % 2 == 0)
+ .map(n -> n * 2)
+ .log()
+ .collectList().block();
+
+ assertThat(list).containsExactly(4, 8);
+ }
+
+}
diff --git a/store-service/src/test/java/com/siriusxi/ms/store/pcs/StoreServiceApplicationTests.java b/store-service/src/test/java/com/siriusxi/ms/store/pcs/StoreServiceApplicationTests.java
index 4eec94b6..ed55d7bb 100644
--- a/store-service/src/test/java/com/siriusxi/ms/store/pcs/StoreServiceApplicationTests.java
+++ b/store-service/src/test/java/com/siriusxi/ms/store/pcs/StoreServiceApplicationTests.java
@@ -1,8 +1,5 @@
package com.siriusxi.ms.store.pcs;
-import com.siriusxi.ms.store.api.composite.dto.ProductAggregate;
-import com.siriusxi.ms.store.api.composite.dto.RecommendationSummary;
-import com.siriusxi.ms.store.api.composite.dto.ReviewSummary;
import com.siriusxi.ms.store.api.core.product.dto.Product;
import com.siriusxi.ms.store.api.core.recommendation.dto.Recommendation;
import com.siriusxi.ms.store.api.core.review.dto.Review;
@@ -17,6 +14,7 @@
import org.springframework.http.HttpStatus;
import org.springframework.test.web.reactive.server.WebTestClient;
import org.springframework.test.web.reactive.server.WebTestClient.BodyContentSpec;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import static java.util.Collections.singletonList;
@@ -33,27 +31,27 @@ class StoreServiceApplicationTests {
private static final int PRODUCT_ID_NOT_FOUND = 2;
private static final int PRODUCT_ID_INVALID = 3;
- @Autowired
- private WebTestClient client;
+ @Autowired private WebTestClient client;
- @MockBean
- private StoreIntegration storeIntegration;
+ @MockBean private StoreIntegration storeIntegration;
@BeforeEach
void setUp() {
when(storeIntegration.getProduct(PRODUCT_ID_OK))
- .thenReturn(new Product(PRODUCT_ID_OK, "name", 1, "mock-address"));
+ .thenReturn(Mono.just(new Product(PRODUCT_ID_OK, "name", 1, "mock-address")));
when(storeIntegration.getRecommendations(PRODUCT_ID_OK))
.thenReturn(
- singletonList(
- new Recommendation(PRODUCT_ID_OK, 1, "author", 1, "content", "mock address")));
+ Flux.fromIterable(
+ singletonList(
+ new Recommendation(PRODUCT_ID_OK, 1, "author", 1, "content", "mock address"))));
when(storeIntegration.getReviews(PRODUCT_ID_OK))
.thenReturn(
- singletonList(
- new Review(PRODUCT_ID_OK, 1, "author", "subject", "content", "mock address")));
+ Flux.fromIterable(
+ singletonList(
+ new Review(PRODUCT_ID_OK, 1, "author", "subject", "content", "mock address"))));
when(storeIntegration.getProduct(PRODUCT_ID_NOT_FOUND))
.thenThrow(new NotFoundException("NOT FOUND: " + PRODUCT_ID_NOT_FOUND));
@@ -62,45 +60,6 @@ void setUp() {
.thenThrow(new InvalidInputException("INVALID: " + PRODUCT_ID_INVALID));
}
- @Test
- public void createCompositeProduct1() {
-
- var compositeProduct = new ProductAggregate(1, "name", 1, null, null, null);
-
- postAndVerifyProductIsCreated(compositeProduct);
- }
-
- @Test
- public void createCompositeProduct2() {
- var compositeProduct =
- new ProductAggregate(
- 1,
- "name",
- 1,
- singletonList(new RecommendationSummary(1, "a", 1, "c")),
- singletonList(new ReviewSummary(1, "a", "s", "c")),
- null);
-
- postAndVerifyProductIsCreated(compositeProduct);
- }
-
- @Test
- public void deleteCompositeProduct() {
- var compositeProduct =
- new ProductAggregate(
- 1,
- "name",
- 1,
- singletonList(new RecommendationSummary(1, "a", 1, "c")),
- singletonList(new ReviewSummary(1, "a", "s", "c")),
- null);
-
- postAndVerifyProductIsCreated(compositeProduct);
-
- deleteAndVerifyProductIsDeleted(compositeProduct.getProductId());
- deleteAndVerifyProductIsDeleted(compositeProduct.getProductId());
- }
-
@Test
public void getProductById() {
@@ -145,18 +104,4 @@ private BodyContentSpec getAndVerifyProduct(int productId, HttpStatus expectedSt
.contentType(APPLICATION_JSON)
.expectBody();
}
-
- private void postAndVerifyProductIsCreated(ProductAggregate compositeProduct) {
- client
- .post()
- .uri(BASE_URL)
- .body(Mono.just(compositeProduct), ProductAggregate.class)
- .exchange()
- .expectStatus()
- .isEqualTo(OK);
- }
-
- private void deleteAndVerifyProductIsDeleted(int productId) {
- client.delete().uri(BASE_URL + productId).exchange().expectStatus().isEqualTo(OK);
- }
}
diff --git a/store-utils/src/main/java/com/siriusxi/ms/store/util/exceptions/EventProcessingException.java b/store-utils/src/main/java/com/siriusxi/ms/store/util/exceptions/EventProcessingException.java
new file mode 100644
index 00000000..a958ba95
--- /dev/null
+++ b/store-utils/src/main/java/com/siriusxi/ms/store/util/exceptions/EventProcessingException.java
@@ -0,0 +1,18 @@
+package com.siriusxi.ms.store.util.exceptions;
+
+public class EventProcessingException extends RuntimeException {
+ public EventProcessingException() {
+ }
+
+ public EventProcessingException(String message) {
+ super(message);
+ }
+
+ public EventProcessingException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public EventProcessingException(Throwable cause) {
+ super(cause);
+ }
+}
\ No newline at end of file
diff --git a/test-em-all.sh b/test-em-all.sh
index d064668f..9e02cb3a 100644
--- a/test-em-all.sh
+++ b/test-em-all.sh
@@ -1,6 +1,6 @@
#!/usr/bin/env bash
## Author: Mohamed Taman
-## version: v3.0
+## version: v4.0
### Sample usage:
#
# for local run
@@ -8,12 +8,17 @@
# with docker compose
# HOST=localhost PORT=8080 ./test-em-all.bash start stop
#
-echo -e "Starting [Springy Store] full functionality testing....\n"
+echo -e "Starting [Springy Store] full functionality [Blackbox] testing....\n"
: ${HOST=localhost}
: ${PORT=8080}
+: ${PROD_ID_REVS_RECS=2}
+: ${PROD_ID_NOT_FOUND=14}
+: ${PROD_ID_NO_RECS=114}
+: ${PROD_ID_NO_REVS=214}
BASE_URL="/store/api/v1/products"
+
function assertCurl() {
local expectedHttpCode=$1
@@ -30,11 +35,12 @@ function assertCurl() {
else
echo "Test OK (HTTP Code: $httpCode, $RESPONSE)"
fi
+ return 0
else
echo "Test FAILED, EXPECTED HTTP Code: $expectedHttpCode, GOT: $httpCode, WILL ABORT!"
echo "- Failing command: $curlCmd"
echo "- Response Body: $RESPONSE"
- exit 1
+ return 1
fi
}
@@ -46,9 +52,10 @@ function assertEqual() {
if [[ "$actual" = "$expected" ]]
then
echo "Test OK (actual value: $actual)"
+ return 0
else
echo "Test FAILED, EXPECTED VALUE: $expected, ACTUAL VALUE: $actual, WILL ABORT"
- exit 1
+ return 1
fi
}
@@ -82,7 +89,52 @@ function waitForService() {
done
}
-function createProduct() {
+function testCompositeCreated() {
+
+ # Expect that the Product Composite for productId $PROD_ID_REVS_RECS
+ # has been created with three recommendations and three reviews
+ if ! assertCurl 200 "curl http://${HOST}:${PORT}${BASE_URL}/${PROD_ID_REVS_RECS} -s"
+ then
+ echo -n "FAIL"
+ return 1
+ fi
+
+ set +e
+ assertEqual "$PROD_ID_REVS_RECS" $(echo ${RESPONSE} | jq .productId)
+ if [[ "$?" -eq "1" ]] ; then return 1; fi
+
+ assertEqual 3 $(echo ${RESPONSE} | jq ".recommendations | length")
+ if [[ "$?" -eq "1" ]] ; then return 1; fi
+
+ assertEqual 3 $(echo ${RESPONSE} | jq ".reviews | length")
+ if [[ "$?" -eq "1" ]] ; then return 1; fi
+
+ set -e
+}
+
+function waitForMessageProcessing() {
+ echo "Wait for messages to be processed... "
+
+ # Give background processing some time to complete...
+ sleep 1
+
+ n=0
+ until testCompositeCreated
+ do
+ n=$((n + 1))
+ if [[ ${n} == 40 ]]
+ then
+ echo " Give up"
+ exit 1
+ else
+ sleep 6
+ echo -n ", retry #$n "
+ fi
+ done
+ echo "All messages are now processed!"
+}
+
+function recreateComposite() {
local productId=$1
local composite=$2
@@ -92,38 +144,43 @@ function createProduct() {
function setupTestData() {
- body=\
-'{"productId":1,"name":"product 1","weight":1, "recommendations":[
- {"recommendationId":1,"author":"author 1","rate":1,"content":"content 1"},
- {"recommendationId":2,"author":"author 2","rate":2,"content":"content 2"},
- {"recommendationId":3,"author":"author 3","rate":3,"content":"content 3"}
- ], "reviews":[
- {"reviewId":1,"author":"author 1","subject":"subject 1","content":"content 1"},
- {"reviewId":2,"author":"author 2","subject":"subject 2","content":"content 2"},
- {"reviewId":3,"author":"author 3","subject":"subject 3","content":"content 3"}
- ]}'
- createProduct 1 "$body"
-
- body=\
-'{"productId":113,"name":"product 113","weight":113, "reviews":[
+ body="{\"productId\":$PROD_ID_NO_RECS"
+ body+=\
+',"name":"product name A","weight":100, "reviews":[
{"reviewId":1,"author":"author 1","subject":"subject 1","content":"content 1"},
{"reviewId":2,"author":"author 2","subject":"subject 2","content":"content 2"},
{"reviewId":3,"author":"author 3","subject":"subject 3","content":"content 3"}
]}'
- createProduct 113 "$body"
+ recreateComposite "$PROD_ID_NO_RECS" "$body"
- body=\
-'{"productId":213,"name":"product 213","weight":213, "recommendations":[
+ body="{\"productId\":$PROD_ID_NO_REVS"
+ body+=\
+',"name":"product name B","weight":200, "recommendations":[
{"recommendationId":1,"author":"author 1","rate":1,"content":"content 1"},
{"recommendationId":2,"author":"author 2","rate":2,"content":"content 2"},
{"recommendationId":3,"author":"author 3","rate":3,"content":"content 3"}
]}'
- createProduct 213 "$body"
+ recreateComposite "$PROD_ID_NO_REVS" "$body"
+
+
+ body="{\"productId\":$PROD_ID_REVS_RECS"
+ body+=\
+',"name":"product name C","weight":300, "recommendations":[
+ {"recommendationId":1,"author":"author 1","rate":1,"content":"content 1"},
+ {"recommendationId":2,"author":"author 2","rate":2,"content":"content 2"},
+ {"recommendationId":3,"author":"author 3","rate":3,"content":"content 3"}
+ ], "reviews":[
+ {"reviewId":1,"author":"author 1","subject":"subject 1","content":"content 1"},
+ {"reviewId":2,"author":"author 2","subject":"subject 2","content":"content 2"},
+ {"reviewId":3,"author":"author 3","subject":"subject 3","content":"content 3"}
+ ]}'
+
+ recreateComposite 1 "$body"
}
set -e
-echo "Start:" `date`
+echo "Start Tests:" `date`
echo "HOST=${HOST}"
echo "PORT=${PORT}"
@@ -131,34 +188,36 @@ echo "PORT=${PORT}"
if [[ $@ == *"start"* ]]
then
echo "Restarting the test environment..."
- echo "$ docker-compose down"
- docker-compose down
+ echo "$ docker-compose -p ssm down --remove-orphans"
+ docker-compose -p ssm down --remove-orphans
echo "$ docker-compose -p ssm up -d"
docker-compose -p ssm up -d
fi
-waitForService curl -X DELETE http://${HOST}:${PORT}${BASE_URL}/13
+waitForService curl http://${HOST}:${PORT}/actuator/health
setupTestData
+waitForMessageProcessing
+
# Verify that a normal request works, expect three recommendations and three reviews
-assertCurl 200 "curl http://$HOST:$PORT${BASE_URL}/1 -s"
-assertEqual 1 $(echo ${RESPONSE} | jq .productId)
+assertCurl 200 "curl http://$HOST:$PORT${BASE_URL}/$PROD_ID_REVS_RECS -s"
+assertEqual ${PROD_ID_REVS_RECS} $(echo ${RESPONSE} | jq .productId)
assertEqual 3 $(echo ${RESPONSE} | jq ".recommendations | length")
assertEqual 3 $(echo ${RESPONSE} | jq ".reviews | length")
# Verify that a 404 (Not Found) error is returned for a non existing productId (13)
-assertCurl 404 "curl http://$HOST:$PORT${BASE_URL}/13 -s"
+assertCurl 404 "curl http://$HOST:$PORT${BASE_URL}/$PROD_ID_NOT_FOUND -s"
# Verify that no recommendations are returned for productId 113
-assertCurl 200 "curl http://$HOST:$PORT${BASE_URL}/113 -s"
-assertEqual 113 $(echo ${RESPONSE} | jq .productId)
+assertCurl 200 "curl http://$HOST:$PORT${BASE_URL}/$PROD_ID_NO_RECS -s"
+assertEqual ${PROD_ID_NO_RECS} $(echo ${RESPONSE} | jq .productId)
assertEqual 0 $(echo ${RESPONSE} | jq ".recommendations | length")
assertEqual 3 $(echo ${RESPONSE} | jq ".reviews | length")
# Verify that no reviews are returned for productId 213
-assertCurl 200 "curl http://$HOST:$PORT${BASE_URL}/213 -s"
-assertEqual 213 $(echo ${RESPONSE} | jq .productId)
+assertCurl 200 "curl http://$HOST:$PORT${BASE_URL}/$PROD_ID_NO_REVS -s"
+assertEqual ${PROD_ID_NO_REVS} $(echo ${RESPONSE} | jq .productId)
assertEqual 3 $(echo ${RESPONSE} | jq ".recommendations | length")
assertEqual 0 $(echo ${RESPONSE} | jq ".reviews | length")
@@ -170,11 +229,11 @@ assertEqual "\"Invalid productId: -1\"" "$(echo ${RESPONSE} | jq .message)"
assertCurl 400 "curl http://$HOST:$PORT${BASE_URL}/invalidProductId -s"
assertEqual "\"Type mismatch.\"" "$(echo ${RESPONSE} | jq .message)"
+echo "End, all tests OK:" `date`
+
if [[ $@ == *"stop"* ]]
then
echo "We are done, stopping the test environment..."
- echo "$ docker-compose down"
- docker-compose -p ssm down
-fi
-
-echo "End:" `date`
\ No newline at end of file
+ echo "$ docker-compose down --remove-orphans"
+ docker-compose -p ssm down --remove-orphans
+fi
\ No newline at end of file