Merge pull request #13 from BOURGUI07/test

BOURGUI07 · web-flow · commit 90b7a34651b8 · 2024-12-15T19:54:30.000+01:00
Redesign the product event publisher implementation using the Listene…
diff --git a/catalog-service/src/main/java/com/example/catalog_service/config/ProductEventPublisherConfig.java b/catalog-service/src/main/java/com/example/catalog_service/config/ProductEventPublisherConfig.java
@@ -27,11 +27,4 @@ public Supplier<Flux<Message<ProductEvent>>> producer(){
                 .doOnNext(msg -> log.info("Catalog-Service Produced Product Event: {}", Util.write(msg.getPayload())));
     }
 
-
-    @Bean
-    public Supplier<Flux<Message<ProductEvent>>> productViewProducer(){
-        return () -> publisher.publishViewedProducts()
-                .map(MessageConverter.toProductEventMessage())
-                .doOnNext(msg -> log.info("Catalog-Service Produced Product Viewed Event: {}", Util.write(msg.getPayload())));
-    }
 }
diff --git a/catalog-service/src/main/java/com/example/catalog_service/domain/ProductAction.java b/catalog-service/src/main/java/com/example/catalog_service/domain/ProductAction.java
@@ -0,0 +1,8 @@
+package com.example.catalog_service.domain;
+
+public enum ProductAction {
+    VIEWED,
+    CREATED,
+    DELETED,
+    UPDATED
+}
diff --git a/catalog-service/src/main/java/com/example/catalog_service/dto/ProductActionDTO.java b/catalog-service/src/main/java/com/example/catalog_service/dto/ProductActionDTO.java
@@ -0,0 +1,11 @@
+package com.example.catalog_service.dto;
+
+import com.example.catalog_service.domain.ProductAction;
+import lombok.Builder;
+
+@Builder
+public record ProductActionDTO(
+        ProductAction action,
+        ProductResponse response
+) {
+}
diff --git a/catalog-service/src/main/java/com/example/catalog_service/listener/ProductEventListener.java b/catalog-service/src/main/java/com/example/catalog_service/listener/ProductEventListener.java
@@ -0,0 +1,28 @@
+package com.example.catalog_service.listener;
+
+import com.example.catalog_service.dto.ProductActionDTO;
+import com.example.catalog_service.events.ProductEvent;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public interface ProductEventListener {
+
+    Flux<ProductEvent> productEvents();
+
+    default void listen(ProductActionDTO dto){
+        switch (dto.action()) {
+            case VIEWED -> handleViewedProduct(dto);
+            case CREATED -> handleCreatedProduct(dto);
+            case UPDATED -> handleUpdatedProduct(dto);
+            case DELETED -> handleDeletedProduct(dto);
+        }
+    }
+
+    void handleDeletedProduct(ProductActionDTO dto);
+
+    void handleUpdatedProduct(ProductActionDTO dto);
+
+    void handleCreatedProduct(ProductActionDTO dto);
+
+    void handleViewedProduct(ProductActionDTO dto);
+}
diff --git a/catalog-service/src/main/java/com/example/catalog_service/listener/ProductEventListenerImpl.java b/catalog-service/src/main/java/com/example/catalog_service/listener/ProductEventListenerImpl.java
@@ -0,0 +1,43 @@
+package com.example.catalog_service.listener;
+
+import com.example.catalog_service.dto.ProductActionDTO;
+import com.example.catalog_service.events.ProductEvent;
+import com.example.catalog_service.mapper.Mapper;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+@Service
+public class ProductEventListenerImpl implements ProductEventListener {
+    private final Sinks.Many<ProductEvent> sink = Sinks.many().unicast().onBackpressureBuffer();
+
+    @Override
+    public Flux<ProductEvent> productEvents(){
+        return sink.asFlux();
+    }
+
+    @Override
+    public void handleDeletedProduct(ProductActionDTO dto) {
+        var deletedEvent = Mapper.toDeletedProductEvent().apply(dto);
+        sink.tryEmitNext(deletedEvent);
+    }
+
+    @Override
+    public void handleUpdatedProduct(ProductActionDTO dto) {
+        var updatedEvent = Mapper.toUpdatedProductEvent().apply(dto);
+        sink.tryEmitNext(updatedEvent);
+    }
+
+    @Override
+    public void handleCreatedProduct(ProductActionDTO dto) {
+        var createdEvent = Mapper.toCreatedProductEvent().apply(dto);
+        sink.tryEmitNext(createdEvent);
+    }
+
+    @Override
+    public void handleViewedProduct(ProductActionDTO dto) {
+        var viewedEvent = Mapper.toViewedProductEvent().apply(dto);
+        sink.tryEmitNext(viewedEvent);
+    }
+}
diff --git a/catalog-service/src/main/java/com/example/catalog_service/mapper/Mapper.java b/catalog-service/src/main/java/com/example/catalog_service/mapper/Mapper.java
@@ -1,6 +1,7 @@
 package com.example.catalog_service.mapper;
 
 import com.example.catalog_service.domain.Product;
+import com.example.catalog_service.domain.ProductAction;
 import com.example.catalog_service.domain.ProductInventory;
 import com.example.catalog_service.dto.*;
 import com.example.catalog_service.events.InventoryEvent;
@@ -114,31 +115,51 @@ public static BiFunction<Throwable,OrderEvent.Created,Mono<InventoryEvent>> toDe
                 .build());
     }
 
+    public static BiFunction<ProductAction,ProductResponse,ProductActionDTO> toProductActionDTO(){
+        return (action, dto) -> ProductActionDTO.builder()
+                .action(action)
+                .response(dto)
+                .build();
+    }
 
-    public static Function<ProductResponse, ProductEvent> toCreatedProductEvent(){
+    public static ProductActionDTO productActionDTO(ProductAction action,String code){
+        return ProductActionDTO.builder()
+                .action(action)
+                .response(ProductResponse.builder().code(code).build())
+                .build();
+    }
+
+    public static Function<ProductActionDTO,ProductEvent> toViewedProductEvent(){
+        return dto -> ProductEvent.View.builder()
+                .code(dto.response().code())
+                .build();
+    }
+
+    public static Function<ProductActionDTO, ProductEvent> toCreatedProductEvent(){
         return dto -> ProductEvent.Created.builder()
-                .code(dto.code())
-                .price(dto.price())
-                .productId(dto.id())
+                .code(dto.response().code())
+                .productId(dto.response().id())
+                .price(dto.response().price())
                 .build();
     }
 
-    public static Function<ProductResponse, ProductEvent> toUpdatedProductEvent(){
+    public static Function<ProductActionDTO, ProductEvent> toUpdatedProductEvent(){
         return dto -> ProductEvent.Updated.builder()
-                .price(dto.price())
-                .code(dto.code())
+                .code(dto.response().code())
+                .price(dto.response().price())
                 .build();
     }
 
-
-    public static Function<String, ProductEvent> toDeletedProductEvent(){
-        return code -> ProductEvent.Deleted.builder()
-                .code(code)
+    public static Function<ProductActionDTO, ProductEvent> toDeletedProductEvent(){
+        return dto -> ProductEvent.Deleted.builder()
+                .code(dto.response().code())
                 .build();
     }
 
 
 
 
 
+
+
 }
diff --git a/catalog-service/src/main/java/com/example/catalog_service/publisher/ProductEventPublisher.java b/catalog-service/src/main/java/com/example/catalog_service/publisher/ProductEventPublisher.java
@@ -1,6 +1,7 @@
 package com.example.catalog_service.publisher;
 
 import com.example.catalog_service.events.ProductEvent;
+import com.example.catalog_service.listener.ProductEventListener;
 import com.example.catalog_service.service.ProductService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -18,18 +19,14 @@
 @RequiredArgsConstructor
 @Slf4j
 public class ProductEventPublisher {
-    private final ProductService service;
+    private final ProductEventListener listener;
 
 
     public Flux<ProductEvent> publish() {
-        return service.products();
+        return listener.productEvents();
     }
 
 
-    public Flux<ProductEvent> publishViewedProducts() {
-        return service.viewedProducts();
-    }
-
 
 
 
diff --git a/catalog-service/src/main/java/com/example/catalog_service/service/ProductService.java b/catalog-service/src/main/java/com/example/catalog_service/service/ProductService.java
@@ -1,9 +1,11 @@
 package com.example.catalog_service.service;
 
 import com.example.catalog_service.domain.Product;
+import com.example.catalog_service.domain.ProductAction;
 import com.example.catalog_service.dto.*;
 import com.example.catalog_service.events.ProductEvent;
 import com.example.catalog_service.exceptions.ApplicationsExceptions;
+import com.example.catalog_service.listener.ProductEventListener;
 import com.example.catalog_service.mapper.Mapper;
 import com.example.catalog_service.repo.ProductRepo;
 import java.math.BigDecimal;
@@ -35,25 +37,9 @@
 public class ProductService {
     private final ProductRepo repo;
     private final CatalogServiceProperties properties;
-
-    private final Sinks.Many<ProductEvent> sink = Sinks.many().unicast().onBackpressureBuffer();
-    private final Sinks.Many<ProductEvent> viewedProductsSink = Sinks.many().unicast().onBackpressureBuffer();
-
     private final CacheTemplate<String,Product> cacheTemplate;
+    private final ProductEventListener listener;
 
-    /**
-     * This is gonna be used by the CreatedProductEventPublisher
-     */
-    public Flux<ProductEvent> products(){
-        return sink.asFlux();
-    }
-
-    /**
-     * This is gonna be used by the ViewedProductEventPublisher
-     */
-    public Flux<ProductEvent> viewedProducts(){
-        return viewedProductsSink.asFlux();
-    }
 
 
     /**
@@ -84,7 +70,7 @@ public Mono<PagedResult<ProductResponse>> getProducts(int page) {
     public Mono<ProductResponse> findByCode(String code) {
         return cacheTemplate.findByKey(code)
                 .map(Mapper.toDto())
-                .doOnNext(dto -> viewedProductsSink.tryEmitNext(ProductEvent.View.builder().code(code).build()))
+                .doOnNext(dto -> listener.listen(Mapper.toProductActionDTO().apply(ProductAction.VIEWED,dto)))
                 .doOnNext(response -> log.info("Product with code: {} is: {}", code, Util.write(response)));
     }
 
@@ -104,7 +90,7 @@ public Mono<ProductResponse> createProduct(Mono<ProductCreationRequest> request)
                 .map(Mapper.toEntity())
                 .flatMap(repo::save)
                 .map(Mapper.toDto())
-                .doOnNext(dto -> sink.tryEmitNext(Mapper.toCreatedProductEvent().apply(dto)))
+                .doOnNext(dto -> listener.listen(Mapper.toProductActionDTO().apply(ProductAction.CREATED,dto)))
                 .doOnNext(response -> log.info("A New Product is Created: {}", Util.write(response)));
 
     }
@@ -135,7 +121,7 @@ public Mono<Void> deleteByCode(String code){
         return  cacheTemplate.findByKey(code)
                 .flatMap(product -> repo.delete(product).then(cacheTemplate.doOnChanged(product)).then())
                 .doOnSuccess(x->{
-                    sink.tryEmitNext(Mapper.toDeletedProductEvent().apply(code));
+                    listener.handleDeletedProduct(Mapper.productActionDTO(ProductAction.DELETED,code));
                     log.info("Product Deleted Successfully");
                 });
     }
@@ -152,7 +138,7 @@ public Mono<ProductResponse> update(String code, Mono<ProductUpdateRequest> requ
                 .flatMap(Function.identity())
                 .doOnNext(response -> {
                     log.info("Product Updated Successfully: {}", Util.write(response));
-                    sink.tryEmitNext(Mapper.toUpdatedProductEvent().apply(response));
+                    listener.handleUpdatedProduct(Mapper.toProductActionDTO().apply(ProductAction.UPDATED,response));
                 });
     }
 
diff --git a/catalog-service/src/main/resources/application.yml b/catalog-service/src/main/resources/application.yml
@@ -62,7 +62,7 @@ spring:
 ############# SPRING CLOUD STREAM WITH KAFKA CONFIGURATION ###############
   cloud:
     function:
-      definition: processor;producer;productViewProducer
+      definition: processor;producer
     stream:
       default:
         producer:
diff --git a/catalog-service/src/test/java/com/example/catalog_service/integration_tests/MESSAGING_abstract_tests/AbstractProductViewPublisherTest.java b/catalog-service/src/test/java/com/example/catalog_service/integration_tests/MESSAGING_abstract_tests/AbstractProductViewPublisherTest.java
@@ -17,7 +17,7 @@
 
 @Import(AbstractProductViewPublisherTest.testConfig.class)
 @TestPropertySource(properties = {
-        "spring.cloud.function.definition=productViewProducer;consumer",
+        "spring.cloud.function.definition=producer;consumer",
         "spring.cloud.stream.bindings.consumer-in-0.destination=catalog-events"
 })
 public class AbstractProductViewPublisherTest extends AbstractIntegrationTest {
diff --git a/catalog-service/src/test/java/com/example/catalog_service/integration_tests/MESSAGING_tests/ProductViewPublisherTests.java b/catalog-service/src/test/java/com/example/catalog_service/integration_tests/MESSAGING_tests/ProductViewPublisherTests.java
@@ -7,8 +7,8 @@ public class ProductViewPublisherTests extends AbstractProductViewPublisherTest
 
     @Test
     void test(){
-        viewProduct("p100");
-        viewProduct("p114");
-        viewProduct("p102");
+        viewProduct("P100");
+        viewProduct("P114");
+        viewProduct("P102");
     }
 }
diff --git a/catalog-service/src/test/java/com/example/catalog_service/unit_tests/CacheServiceTests.java b/catalog-service/src/test/java/com/example/catalog_service/unit_tests/CacheServiceTests.java
diff --git a/catalog-service/src/test/java/com/example/catalog_service/unit_tests/InventoryServiceTests.java b/catalog-service/src/test/java/com/example/catalog_service/unit_tests/InventoryServiceTests.java
diff --git a/catalog-service/src/test/java/com/example/catalog_service/unit_tests/ProductServiceTests.java b/catalog-service/src/test/java/com/example/catalog_service/unit_tests/ProductServiceTests.java

Original file line number	Diff line number	Diff line change
`@@ -27,11 +27,4 @@ public Supplier<Flux<Message<ProductEvent>>> producer(){`
`27`	`27`	`.doOnNext(msg -> log.info("Catalog-Service Produced Product Event: {}", Util.write(msg.getPayload())));`
`28`	`28`	`}`
`29`	`29`
`30`		`-`
`31`		`- @Bean`
`32`		`- public Supplier<Flux<Message<ProductEvent>>> productViewProducer(){`
`33`		`- return () -> publisher.publishViewedProducts()`
`34`		`- .map(MessageConverter.toProductEventMessage())`
`35`		`- .doOnNext(msg -> log.info("Catalog-Service Produced Product Viewed Event: {}", Util.write(msg.getPayload())));`
`36`		`- }`
`37`	`30`	`}`