diff --git a/2023-11/spring-30-endpoints-flow-components/.gitignore b/2023-11/spring-30-endpoints-flow-components/.gitignore new file mode 100644 index 00000000..4ea52072 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/.gitignore @@ -0,0 +1,24 @@ +target/ + +### STS ### +.apt_generated +.classpath +.factorypath +.project +.settings +.springBeans +.sts4-cache + +### IntelliJ IDEA ### +.idea +*.iws +*.iml +*.ipr + +### NetBeans ### +/nbproject/private/ +/build/ +/nbbuild/ +/dist/ +/nbdist/ +/.nb-gradle/ diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/pom.xml b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/pom.xml new file mode 100644 index 00000000..6c20857a --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/pom.xml @@ -0,0 +1,48 @@ + + + 4.0.0 + + ru.otus + endpoints-flow-components-exercise + 1.0-SNAPSHOT + + + ru.otus + endpoints-flow-components + 1.0 + + + + + org.springframework.boot + spring-boot-starter-integration + + + + org.apache.commons + commons-lang3 + + + + + org.projectlombok + lombok + provided + + + com.fasterxml.jackson.core + jackson-databind + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/App.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/App.java new file mode 100644 index 00000000..994b4482 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/App.java @@ -0,0 +1,17 @@ +package ru.otus.spring.integration; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; + +import ru.otus.spring.integration.services.OrderService; + + +@SpringBootApplication +public class App { + public static void main(String[] args) { + ConfigurableApplicationContext ctx = SpringApplication.run(App.class, args); + OrderService orderService = ctx.getBean(OrderService.class); + orderService.startGenerateOrdersLoop(); + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/config/IntegrationConfig.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/config/IntegrationConfig.java new file mode 100644 index 00000000..3d9183a7 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/config/IntegrationConfig.java @@ -0,0 +1,32 @@ +package ru.otus.spring.integration.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.MessageChannelSpec; +import org.springframework.integration.dsl.MessageChannels; + +@SuppressWarnings("unused") +@Configuration +public class IntegrationConfig { + @Bean + public MessageChannelSpec itemsChannel() { + return MessageChannels.queue(10); + } + + @Bean + public MessageChannelSpec foodChannel() { + return MessageChannels.publishSubscribe(); + } + + // TODO: create default poller + + @Bean + public IntegrationFlow cafeFlow() { + return IntegrationFlow.from(itemsChannel()) + // TODO: cook OrderItem in the kitchen + // TODO*: add splitter and aggregator + // TODO: forward it to the publish subscriber channel + .get(); + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/domain/Food.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/domain/Food.java new file mode 100644 index 00000000..23f7a1b9 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/domain/Food.java @@ -0,0 +1,15 @@ +package ru.otus.spring.integration.domain; + + +public class Food { + + private final String name; + + public Food(String name) { + this.name = name; + } + + public String getName() { + return name; + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/domain/OrderItem.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/domain/OrderItem.java new file mode 100644 index 00000000..992bf044 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/domain/OrderItem.java @@ -0,0 +1,14 @@ +package ru.otus.spring.integration.domain; + +public class OrderItem { + + private final String itemName; + + public OrderItem(String itemName) { + this.itemName = itemName; + } + + public String getItemName() { + return itemName; + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/Cafe.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/Cafe.java new file mode 100644 index 00000000..b0af720f --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/Cafe.java @@ -0,0 +1,12 @@ +package ru.otus.spring.integration.services; + + +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +// TODO: add messaging gateway annotation +public interface Cafe { + + // TODO: add gateway annotation with required channels + Food process(OrderItem orderItem); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/CafeGateway.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/CafeGateway.java new file mode 100644 index 00000000..5b11b23d --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/CafeGateway.java @@ -0,0 +1,12 @@ +package ru.otus.spring.integration.services; + + +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +// TODO: add messaging gateway annotation +public interface CafeGateway { + + // TODO: add gateway annotation with required channels + Food process(OrderItem orderItem); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/CafeService.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/CafeService.java new file mode 100644 index 00000000..6b3b8629 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/CafeService.java @@ -0,0 +1,17 @@ +package ru.otus.spring.integration.services; + + +import java.util.List; +import java.util.Map; + +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.MessagingGateway; + +import ru.otus.spring.integration.domain.OrderItem; + +@MessagingGateway +public interface CafeService { + + @Gateway(requestChannel = "itemsChannel", replyChannel = "foodChannel") + List> process(List orderItem); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/KitchenService.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/KitchenService.java new file mode 100644 index 00000000..cc9998ae --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/KitchenService.java @@ -0,0 +1,9 @@ +package ru.otus.spring.integration.services; + +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +public interface KitchenService { + + Food cook(OrderItem orderItem); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/KitchenServiceImpl.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/KitchenServiceImpl.java new file mode 100644 index 00000000..be4f9ae2 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/KitchenServiceImpl.java @@ -0,0 +1,29 @@ +package ru.otus.spring.integration.services; + +import org.springframework.stereotype.Service; + +import lombok.extern.slf4j.Slf4j; +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +@Service +@Slf4j +public class KitchenServiceImpl implements KitchenService { + + @Override + public Food cook(OrderItem orderItem) { + log.info("Cooking {}", orderItem.getItemName()); + delay(); + log.info("Cooking {} done", orderItem.getItemName()); + + return new Food(orderItem.getItemName()); + } + + private static void delay() { + try { + Thread.sleep(3000); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/OrderService.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/OrderService.java new file mode 100644 index 00000000..3ea65c11 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/OrderService.java @@ -0,0 +1,5 @@ +package ru.otus.spring.integration.services; + +public interface OrderService { + void startGenerateOrdersLoop(); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/OrderServiceImpl.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/OrderServiceImpl.java new file mode 100644 index 00000000..32a8af06 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/OrderServiceImpl.java @@ -0,0 +1,44 @@ +package ru.otus.spring.integration.services; + +import org.apache.commons.lang3.RandomUtils; +import org.springframework.stereotype.Service; + +import lombok.extern.slf4j.Slf4j; +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +@Service +@Slf4j +public class OrderServiceImpl implements OrderService { + private static final String[] MENU = {"coffee", "tea", "smoothie", "whiskey", "beer", "cola", "water"}; + + private final CafeGateway cafe; + + public OrderServiceImpl(CafeGateway cafe) { + this.cafe = cafe; + } + + @Override + public void startGenerateOrdersLoop() { + for (int i = 0; i < 10; i++) { + OrderItem orderItem = generateOrderItem(); + int num = i + 1; + log.info("{}, New orderItem: {}", num, orderItem.getItemName()); + Food food = cafe.process(orderItem); + log.info("{}, Ready food: {}", num, food.getName()); + delay(); + } + } + + private static OrderItem generateOrderItem() { + return new OrderItem(MENU[RandomUtils.nextInt(0, MENU.length)]); + } + + private void delay() { + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/resources/application.yml b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/resources/application.yml new file mode 100644 index 00000000..1a9aeb48 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/resources/application.yml @@ -0,0 +1,5 @@ +logging: + level: + root: info + + org.springframework.integration: debug \ No newline at end of file diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/pom.xml b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/pom.xml new file mode 100644 index 00000000..fd184690 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/pom.xml @@ -0,0 +1,72 @@ + + + 4.0.0 + + ru.otus + endpoints-flow-components-solution + 1.0-SNAPSHOT + + + ru.otus + endpoints-flow-components + 1.0 + + + + + org.springframework.boot + spring-boot-starter-integration + + + + org.apache.commons + commons-lang3 + + + + org.projectlombok + lombok + provided + + + com.fasterxml.jackson.core + jackson-databind + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + ru.otus.spring.integration.App + + + + org.apache.maven.plugins + maven-checkstyle-plugin + ${checkstyle-plugin.version} + + + com.puppycrawl.tools + checkstyle + ${checkstyle.version} + + + + ${checkstyle.config.url} + + + + + check + + + + + + + diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/App.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/App.java new file mode 100644 index 00000000..6fba7d7c --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/App.java @@ -0,0 +1,20 @@ +package ru.otus.spring.integration; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; + +import lombok.extern.slf4j.Slf4j; +import ru.otus.spring.integration.services.OrderService; + +@Slf4j +@SpringBootApplication +public class App { + + public static void main(String[] args) { + ConfigurableApplicationContext ctx = SpringApplication.run(App.class, args); + + OrderService orderService = ctx.getBean(OrderService.class); + orderService.startGenerateOrdersLoop(); + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/config/IntegrationConfig.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/config/IntegrationConfig.java new file mode 100644 index 00000000..99454510 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/config/IntegrationConfig.java @@ -0,0 +1,43 @@ +package ru.otus.spring.integration.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.MessageChannelSpec; +import org.springframework.integration.dsl.MessageChannels; +import org.springframework.integration.dsl.PollerSpec; +import org.springframework.integration.dsl.Pollers; +import org.springframework.integration.scheduling.PollerMetadata; + +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.services.KitchenService; + +@Configuration +public class IntegrationConfig { + + @Bean + public MessageChannelSpec itemsChannel() { + return MessageChannels.queue(10); + } + + @Bean + public MessageChannelSpec foodChannel() { + return MessageChannels.publishSubscribe(); + } + + @Bean(name = PollerMetadata.DEFAULT_POLLER) + public PollerSpec poller() { + return Pollers.fixedRate(100).maxMessagesPerPoll(2); + } + + @Bean + public IntegrationFlow cafeFlow(KitchenService kitchenService) { + return IntegrationFlow.from(itemsChannel()) + .split() + .handle(kitchenService, "cook") + .transform(f -> new Food(f.getName().toUpperCase())) + .aggregate() + .channel(foodChannel()) + .get(); + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/domain/Food.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/domain/Food.java new file mode 100644 index 00000000..23f7a1b9 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/domain/Food.java @@ -0,0 +1,15 @@ +package ru.otus.spring.integration.domain; + + +public class Food { + + private final String name; + + public Food(String name) { + this.name = name; + } + + public String getName() { + return name; + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/domain/OrderItem.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/domain/OrderItem.java new file mode 100644 index 00000000..992bf044 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/domain/OrderItem.java @@ -0,0 +1,14 @@ +package ru.otus.spring.integration.domain; + +public class OrderItem { + + private final String itemName; + + public OrderItem(String itemName) { + this.itemName = itemName; + } + + public String getItemName() { + return itemName; + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/Cafe.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/Cafe.java new file mode 100644 index 00000000..aaaf753f --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/Cafe.java @@ -0,0 +1,17 @@ +package ru.otus.spring.integration.services; + + +import java.util.Collection; + +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.MessagingGateway; + +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +@MessagingGateway +public interface Cafe { + + @Gateway(requestChannel = "itemsChannel", replyChannel = "foodChannel") + Collection process(Collection orderItem); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/CafeGateway.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/CafeGateway.java new file mode 100644 index 00000000..fb8e4972 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/CafeGateway.java @@ -0,0 +1,17 @@ +package ru.otus.spring.integration.services; + + +import java.util.Collection; + +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.MessagingGateway; + +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +@MessagingGateway +public interface CafeGateway { + + @Gateway(requestChannel = "itemsChannel", replyChannel = "foodChannel") + Collection process(Collection orderItem); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/KitchenService.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/KitchenService.java new file mode 100644 index 00000000..cc9998ae --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/KitchenService.java @@ -0,0 +1,9 @@ +package ru.otus.spring.integration.services; + +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +public interface KitchenService { + + Food cook(OrderItem orderItem); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/KitchenServiceImpl.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/KitchenServiceImpl.java new file mode 100644 index 00000000..66e7b240 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/KitchenServiceImpl.java @@ -0,0 +1,28 @@ +package ru.otus.spring.integration.services; + +import org.springframework.stereotype.Service; + +import lombok.extern.slf4j.Slf4j; +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +@Service +@Slf4j +public class KitchenServiceImpl implements KitchenService { + + @Override + public Food cook(OrderItem orderItem) { + log.info("Cooking {}", orderItem.getItemName()); + delay(); + log.info("Cooking {} done", orderItem.getItemName()); + return new Food(orderItem.getItemName()); + } + + private static void delay() { + try { + Thread.sleep(3000); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/OrderService.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/OrderService.java new file mode 100644 index 00000000..3ea65c11 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/OrderService.java @@ -0,0 +1,5 @@ +package ru.otus.spring.integration.services; + +public interface OrderService { + void startGenerateOrdersLoop(); +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/OrderServiceImpl.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/OrderServiceImpl.java new file mode 100644 index 00000000..5dac1c25 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/OrderServiceImpl.java @@ -0,0 +1,66 @@ +package ru.otus.spring.integration.services; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.ForkJoinPool; +import java.util.stream.Collectors; + +import org.apache.commons.lang3.RandomUtils; +import org.springframework.stereotype.Service; + +import lombok.extern.slf4j.Slf4j; +import ru.otus.spring.integration.domain.Food; +import ru.otus.spring.integration.domain.OrderItem; + +@Service +@Slf4j +public class OrderServiceImpl implements OrderService { + private static final String[] MENU = {"coffee", "tea", "smoothie", "whiskey", "beer", "cola", "water"}; + + private final CafeGateway cafe; + + public OrderServiceImpl(CafeGateway cafe) { + this.cafe = cafe; + } + + @Override + public void startGenerateOrdersLoop() { + ForkJoinPool pool = ForkJoinPool.commonPool(); + for (int i = 0; i < 10; i++) { + int num = i + 1; + pool.execute(() -> { + Collection items = generateOrderItems(); + log.info("{}, New orderItems: {}", num, + items.stream().map(OrderItem::getItemName) + .collect(Collectors.joining(","))); + Collection food = cafe.process(items); + log.info("{}, Ready food: {}", num, food.stream() + .map(Food::getName) + .collect(Collectors.joining(","))); + }); + delay(); + } + } + + private static OrderItem generateOrderItem() { + return new OrderItem(MENU[RandomUtils.nextInt(0, MENU.length)]); + } + + private static Collection generateOrderItems() { + List items = new ArrayList<>(); + for (int i = 0; i < RandomUtils.nextInt(1, 5); ++i) { + items.add(generateOrderItem()); + } + return items; + } + + private void delay() { + try { + Thread.sleep(7000); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/bridge/BridgeApp.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/bridge/BridgeApp.java new file mode 100644 index 00000000..49070065 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/bridge/BridgeApp.java @@ -0,0 +1,59 @@ +package ru.otus.spring.test.bridge; + +import java.util.Collection; +import java.util.List; +import java.util.Map; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.MessageChannels; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; + +import lombok.extern.slf4j.Slf4j; + +@SuppressWarnings("unused") +@SpringBootApplication +@IntegrationComponentScan +@Slf4j +public class BridgeApp { + public static void main(String[] args) { + ConfigurableApplicationContext ctx = SpringApplication.run(BridgeApp.class, args); + Map channels = ctx.getBeansOfType(MessageChannel.class); + log.warn("CHANNELS:"); + int i = 0; + for (Map.Entry entry : channels.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + log.warn("HANDLERS:"); + i = 0; + Map endpoints = ctx.getBeansOfType(MessageHandler.class); + for (Map.Entry entry : endpoints.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + + Bridge bean = ctx.getBean(Bridge.class); + List strings = List.of("TEST1", "end"); + Collection result = bean.send(strings); + log.warn("Bridge send: {}, receive: {}", strings, result); + } + + + @MessagingGateway + public interface Bridge { + @Gateway(requestChannel = "flow.input"/*, replyChannel = "p2pChannel"*/) + Collection send(Collection strings); + } + + @Bean + public IntegrationFlow flow() { + return f -> f + .channel(MessageChannels.queue("p2pChannel", 10)); + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/gateway/GatewayApp.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/gateway/GatewayApp.java new file mode 100644 index 00000000..34866b9f --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/gateway/GatewayApp.java @@ -0,0 +1,77 @@ +package ru.otus.spring.test.gateway; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Map; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; + +import lombok.extern.slf4j.Slf4j; + +@SuppressWarnings("unused") +@SpringBootApplication +@IntegrationComponentScan +@Slf4j +public class GatewayApp { + public static void main(String[] args) { + ConfigurableApplicationContext ctx = SpringApplication.run(GatewayApp.class, args); + Map channels = ctx.getBeansOfType(MessageChannel.class); + log.warn("CHANNELS:"); + int i = 0; + for (Map.Entry entry : channels.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + log.warn("HANDLERS:"); + i = 0; + Map endpoints = ctx.getBeansOfType(MessageHandler.class); + for (Map.Entry entry : endpoints.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + Upcase upcase = ctx.getBean(Upcase.class); + Collection result = upcase.upcase(Arrays.asList("test", "new", "last")); + log.warn("Upcase result: {}", result); + + } + + @MessagingGateway + public interface Upcase { + @Gateway(requestChannel = "upcase.input") + Collection upcase(Collection strings); + } + + @Bean + public IntegrationFlow upcase() { + return f -> f//.channel("from-input-to-split") + .split() +// .split(list -> list.getObject().spliterator()) +// .split(getCustomSplitter(), "split") + .channel("from-split-to-transformer") + .transform(String::toUpperCase) + .channel("from-transformer-to-aggregate") + .aggregate(); +// .>filter(source -> source.stream().anyMatch(s -> s.startsWith("a"))) + } + + +// @Bean +// CustomSplitter getCustomSplitter() { +// return new CustomSplitter(); +// } +// +// public static class CustomSplitter { +// public Collection split(Message> message) { +// return message.getPayload().stream().skip(1).collect(Collectors.toList()); +// } +// } + + +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/polling/PollingApp.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/polling/PollingApp.java new file mode 100644 index 00000000..ba9e3b79 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/polling/PollingApp.java @@ -0,0 +1,93 @@ +package ru.otus.spring.test.polling; + +import static java.lang.Thread.sleep; + +import java.util.Map; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.MessageChannelSpec; +import org.springframework.integration.dsl.MessageChannels; +import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; + +import lombok.extern.slf4j.Slf4j; + +@SuppressWarnings("unused") +@SpringBootApplication +@IntegrationComponentScan +@Slf4j +public class PollingApp { + public static void main(String[] args) throws InterruptedException { + ConfigurableApplicationContext ctx = SpringApplication.run(PollingApp.class, args); + log.warn("POLLER:"); + Map pollers = ctx.getBeansOfType(PollingConsumer.class); + int i = 0; + for (Map.Entry entry : pollers.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + log.warn("CHANNELS:"); + Map channels = ctx.getBeansOfType(MessageChannel.class); + i = 0; + for (Map.Entry entry : channels.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + log.warn("HANDLERS:"); + i = 0; + Map endpoints = ctx.getBeansOfType(MessageHandler.class); + for (Map.Entry entry : endpoints.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + + Polling polling = ctx.getBean(Polling.class); + String result = polling.send("test"); + log.warn("Polling result: {}", result); + + sleep(5000); + ctx.close(); + } + + @MessagingGateway + public interface Polling { + @Gateway(requestChannel = "flow.input", replyChannel = "pubSub") + String send(String value); + } + +// @Bean(name = PollerMetadata.DEFAULT_POLLER) +// public PollerMetadata defaultPoller() { +//// return Pollers.fixedRate(10_000).get(); +// PollerMetadata pollerMetadata = new PollerMetadata(); +// pollerMetadata.setMaxMessagesPerPoll(5); +// pollerMetadata.setTrigger(new PeriodicTrigger(Duration.ofSeconds(3))); +// return pollerMetadata; +// } + + @Bean + public IntegrationFlow flow() { + return f -> f + .channel("pubSub") + .channel("p2p"); + } + + @Bean + public MessageChannelSpec p2p() { + return MessageChannels.queue("p2p", 10); + } + + @Bean + public MessageChannelSpec p2p2() { + return MessageChannels.priority("p2p2").capacity(10); + } + + @Bean + public MessageChannelSpec pubSub() { + return MessageChannels.publishSubscribe("pubSub"); + } +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/transform/TransformApp.java b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/transform/TransformApp.java new file mode 100644 index 00000000..4577c215 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/transform/TransformApp.java @@ -0,0 +1,92 @@ +package ru.otus.spring.test.transform; + +import java.util.Arrays; +import java.util.Collection; +import java.util.Map; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.annotation.Gateway; +import org.springframework.integration.annotation.IntegrationComponentScan; +import org.springframework.integration.annotation.MessagingGateway; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.Transformers; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; + +import lombok.Getter; +import lombok.extern.slf4j.Slf4j; + +@SuppressWarnings("unused") +@SpringBootApplication +@IntegrationComponentScan +@Slf4j +public class TransformApp { + public static void main(String[] args) { + ConfigurableApplicationContext ctx = SpringApplication.run(TransformApp.class, args); + Map channels = ctx.getBeansOfType(MessageChannel.class); + log.warn("CHANNELS:"); + int i = 0; + for (Map.Entry entry : channels.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + log.warn("HANDLERS:"); + i = 0; + Map endpoints = ctx.getBeansOfType(MessageHandler.class); + for (Map.Entry entry : endpoints.entrySet()) { + log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue()); + } + Upcase upcase = ctx.getBean(Upcase.class); + Collection result = upcase.upcase(Arrays.asList(new Item("test"), new Item("new"), new Item("last"))); +// Collection result = upcase.upcase(Arrays.asList("test", "new", "last")); + log.warn("Upcase result: {}", result); + + } + + @Getter + public static class Item { + public Item() { + } + + public Item(String value) { + this.value = value; + } + + String value; + String name; + + public void setValue(String value) { + this.value = value; + } + + @Override + public String toString() { + return "Item{" + + "value='" + value + '\'' + + ", name='" + name + '\'' + + '}'; + } + } + + @MessagingGateway + public interface Upcase { + @Gateway(requestChannel = "upcase.input") + Collection upcase(Collection strings); + } + + @Bean + public IntegrationFlow upcase() { + return f -> f + .split() + .transform(Transformers.toMap()) + ., Map>transform(map -> { + map.replaceAll((k, v) -> v != null ? v.toUpperCase() : v); + return map; + }) + .transform(Transformers.fromMap(Item.class)) + .aggregate(); + } + +} diff --git a/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/resources/application.yml b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/resources/application.yml new file mode 100644 index 00000000..1a9aeb48 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/resources/application.yml @@ -0,0 +1,5 @@ +logging: + level: + root: info + + org.springframework.integration: debug \ No newline at end of file diff --git a/2023-11/spring-30-endpoints-flow-components/pom.xml b/2023-11/spring-30-endpoints-flow-components/pom.xml new file mode 100644 index 00000000..2ae9fba6 --- /dev/null +++ b/2023-11/spring-30-endpoints-flow-components/pom.xml @@ -0,0 +1,34 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 3.2.4 + + + ru.otus + endpoints-flow-components + 1.0 + + pom + + + endpoints-flow-components-exercise + endpoints-flow-components-solution + + + + 17 + 17 + UTF-8 + UTF-8 + 3.1.2 + 10.9.1 + https://raw.githubusercontent.com/OtusTeam/Spring/master/checkstyle.xml + + +