2025-11 spring-30-endpoints-flow-components

2025-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/

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/pom.xml

@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>ru.otus</groupId>
+    <artifactId>endpoints-flow-components-exercise</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <parent>
+        <groupId>ru.otus</groupId>
+        <artifactId>endpoints-flow-components</artifactId>
+        <version>1.0</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/App.java

@@ -0,0 +1,12 @@
+package ru.otus.spring.integration;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+
+@SpringBootApplication
+public class App {
+    public static void main(String[] args) {
+        SpringApplication.run(App.class, args);
+    }
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/config/AppRunner.java

@@ -0,0 +1,19 @@
+package ru.otus.spring.integration.config;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.stereotype.Component;
+import ru.otus.spring.integration.services.OrderService;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class AppRunner implements CommandLineRunner {
+    final OrderService orderService;
+
+    @Override
+    public void run(String... args) {
+        orderService.startGenerateOrdersLoop();
+    }
+}

2025-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();
+    }
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/domain/Food.java

@@ -0,0 +1,6 @@
+package ru.otus.spring.integration.domain;
+
+
+public record Food(String name) {
+
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/domain/OrderItem.java

@@ -0,0 +1,5 @@
+package ru.otus.spring.integration.domain;
+
+public record OrderItem(String itemName) {
+
+}

2025-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);
+}

2025-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);
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/KitchenServiceImpl.java

@@ -0,0 +1,28 @@
+package ru.otus.spring.integration.services;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+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.itemName());
+        delay();
+        log.info("Cooking {} done", orderItem.itemName());
+
+        return new Food(orderItem.itemName());
+    }
+
+    private static void delay() {
+        try {
+            Thread.sleep(3000);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

2025-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();
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-exercise/src/main/java/ru/otus/spring/integration/services/OrderServiceImpl.java

@@ -0,0 +1,43 @@
+package ru.otus.spring.integration.services;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.RandomUtils;
+import org.springframework.stereotype.Service;
+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.itemName());
+            Food food = cafe.process(orderItem);
+            log.info("{}, Ready food: {}", num, food.name());
+            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);
+        }
+    }
+}

2025-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

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/pom.xml

@@ -0,0 +1,72 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>ru.otus</groupId>
+    <artifactId>endpoints-flow-components-solution</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <parent>
+        <groupId>ru.otus</groupId>
+        <artifactId>endpoints-flow-components</artifactId>
+        <version>1.0</version>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <configuration>
+                    <mainClass>ru.otus.spring.integration.App</mainClass>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-checkstyle-plugin</artifactId>
+                <version>${checkstyle-plugin.version}</version>
+                <dependencies>
+                    <dependency>
+                        <groupId>com.puppycrawl.tools</groupId>
+                        <artifactId>checkstyle</artifactId>
+                        <version>${checkstyle.version}</version>
+                    </dependency>
+                </dependencies>
+                <configuration>
+                    <configLocation>${checkstyle.config.url}</configLocation>
+                </configuration>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>check</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+</project>

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/App.java

@@ -0,0 +1,14 @@
+package ru.otus.spring.integration;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@Slf4j
+@SpringBootApplication
+public class App {
+
+    public static void main(String[] args) {
+        SpringApplication.run(App.class, args);
+    }
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/config/AppRunner.java

@@ -0,0 +1,19 @@
+package ru.otus.spring.integration.config;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.stereotype.Component;
+import ru.otus.spring.integration.services.OrderService;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class AppRunner implements CommandLineRunner {
+    final OrderService orderService;
+
+    @Override
+    public void run(String... args) {
+        orderService.startGenerateOrdersLoop();
+    }
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/config/IntegrationConfig.java

@@ -0,0 +1,38 @@
+package ru.otus.spring.integration.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.dsl.*;
+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")
+                .<Food, Food>transform(f -> new Food(f.name().toUpperCase()))
+                .aggregate()
+                .channel(foodChannel())
+                .get();
+    }
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/domain/Food.java

@@ -0,0 +1,6 @@
+package ru.otus.spring.integration.domain;
+
+
+public record Food(String name) {
+
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/domain/OrderItem.java

@@ -0,0 +1,5 @@
+package ru.otus.spring.integration.domain;
+
+public record OrderItem(String itemName) {
+
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/CafeGateway.java

@@ -0,0 +1,16 @@
+package ru.otus.spring.integration.services;
+
+
+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;
+
+import java.util.Collection;
+
+@MessagingGateway
+public interface CafeGateway {
+
+    @Gateway(requestChannel = "itemsChannel", replyChannel = "foodChannel")
+    Collection<Food> process(Collection<OrderItem> orderItem);
+}

2025-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);
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/KitchenServiceImpl.java

@@ -0,0 +1,27 @@
+package ru.otus.spring.integration.services;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+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.itemName());
+        delay();
+        log.info("Cooking {} done", orderItem.itemName());
+        return new Food(orderItem.itemName());
+    }
+
+    private static void delay() {
+        try {
+            Thread.sleep(3000);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

2025-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();
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/integration/services/OrderServiceImpl.java

@@ -0,0 +1,65 @@
+package ru.otus.spring.integration.services;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.RandomUtils;
+import org.springframework.stereotype.Service;
+import ru.otus.spring.integration.domain.Food;
+import ru.otus.spring.integration.domain.OrderItem;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.ForkJoinPool;
+import java.util.stream.Collectors;
+
+@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<OrderItem> items = generateOrderItems();
+                log.info("{}, New orderItems: {}", num,
+                        items.stream().map(OrderItem::itemName)
+                                .collect(Collectors.joining(",")));
+                Collection<Food> food = cafe.process(items);
+                log.info("{}, Ready food: {}", num, food.stream()
+                        .map(Food::name)
+                        .collect(Collectors.joining(",")));
+            });
+            delay();
+        }
+    }
+
+    private static OrderItem generateOrderItem() {
+        return new OrderItem(MENU[RandomUtils.nextInt(0, MENU.length)]);
+    }
+
+    private static Collection<OrderItem> generateOrderItems() {
+        List<OrderItem> 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);
+        }
+
+    }
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/bridge/BridgeApp.java

@@ -0,0 +1,58 @@
+package ru.otus.spring.test.bridge;
+
+import lombok.extern.slf4j.Slf4j;
+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 java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+@SuppressWarnings("unused")
+@SpringBootApplication
+@IntegrationComponentScan
+@Slf4j
+public class BridgeApp {
+    public static void main(String[] args) {
+        ConfigurableApplicationContext ctx = SpringApplication.run(BridgeApp.class, args);
+        Map<String, MessageChannel> channels = ctx.getBeansOfType(MessageChannel.class);
+        log.warn("CHANNELS:");
+        int i = 0;
+        for (Map.Entry<String, MessageChannel> entry : channels.entrySet()) {
+            log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue());
+        }
+        log.warn("HANDLERS:");
+        i = 0;
+        Map<String, MessageHandler> endpoints = ctx.getBeansOfType(MessageHandler.class);
+        for (Map.Entry<String, MessageHandler> entry : endpoints.entrySet()) {
+            log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue());
+        }
+
+        Bridge bean = ctx.getBean(Bridge.class);
+        List<String> strings = List.of("TEST1", "end");
+        Collection<String> result = bean.send(strings);
+        log.warn("Bridge send: {}, receive: {}", strings, result);
+    }
+
+
+    @MessagingGateway
+    public interface Bridge {
+        @Gateway(requestChannel = "flow.input"/*, replyChannel = "p2pChannel"*/)
+        Collection<String> send(Collection<String> strings);
+    }
+
+    @Bean
+    public IntegrationFlow flow() {
+        return f -> f
+                .channel(MessageChannels.queue("p2pChannel", 10));
+    }
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/gateway/GatewayApp.java

@@ -0,0 +1,76 @@
+package ru.otus.spring.test.gateway;
+
+import lombok.extern.slf4j.Slf4j;
+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 java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+
+@SuppressWarnings("unused")
+@SpringBootApplication
+@IntegrationComponentScan
+@Slf4j
+public class GatewayApp {
+    public static void main(String[] args) {
+        ConfigurableApplicationContext ctx = SpringApplication.run(GatewayApp.class, args);
+        Map<String, MessageChannel> channels = ctx.getBeansOfType(MessageChannel.class);
+        log.warn("CHANNELS:");
+        int i = 0;
+        for (Map.Entry<String, MessageChannel> entry : channels.entrySet()) {
+            log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue());
+        }
+        log.warn("HANDLERS:");
+        i = 0;
+        Map<String, MessageHandler> endpoints = ctx.getBeansOfType(MessageHandler.class);
+        for (Map.Entry<String, MessageHandler> entry : endpoints.entrySet()) {
+            log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue());
+        }
+        Upcase upcase = ctx.getBean(Upcase.class);
+        Collection<String> result = upcase.upcase(Arrays.asList("test", "new", "last"));
+        log.warn("Upcase result: {}", result);
+
+    }
+
+    @MessagingGateway
+    public interface Upcase {
+        @Gateway(requestChannel = "upcase.input")
+        Collection<String> upcase(Collection<String> 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")
+                .<String, String>transform(String::toUpperCase)
+                .channel("from-transformer-to-aggregate")
+                .aggregate();
+//				.<Collection<String>>filter(source -> source.stream().anyMatch(s -> s.startsWith("a")))
+    }
+
+
+//	@Bean
+//	CustomSplitter getCustomSplitter() {
+//		return new CustomSplitter();
+//	}
+//
+//	public static class CustomSplitter {
+//		public Collection<String> split(Message<Collection<String>> message) {
+//			return message.getPayload().stream().skip(1).collect(Collectors.toList());
+//		}
+//	}
+
+
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/polling/PollingApp.java

@@ -0,0 +1,92 @@
+package ru.otus.spring.test.polling;
+
+import lombok.extern.slf4j.Slf4j;
+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 java.util.Map;
+
+import static java.lang.Thread.sleep;
+
+@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<String, PollingConsumer> pollers = ctx.getBeansOfType(PollingConsumer.class);
+        int i = 0;
+        for (Map.Entry<String, PollingConsumer> entry : pollers.entrySet()) {
+            log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue());
+        }
+        log.warn("CHANNELS:");
+        Map<String, MessageChannel> channels = ctx.getBeansOfType(MessageChannel.class);
+        i = 0;
+        for (Map.Entry<String, MessageChannel> entry : channels.entrySet()) {
+            log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue());
+        }
+        log.warn("HANDLERS:");
+        i = 0;
+        Map<String, MessageHandler> endpoints = ctx.getBeansOfType(MessageHandler.class);
+        for (Map.Entry<String, MessageHandler> 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");
+    }
+}

2025-11/spring-30-endpoints-flow-components/endpoints-flow-components-solution/src/main/java/ru/otus/spring/test/transform/TransformApp.java

@@ -0,0 +1,91 @@
+package ru.otus.spring.test.transform;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+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 java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+
+@SuppressWarnings("unused")
+@SpringBootApplication
+@IntegrationComponentScan
+@Slf4j
+public class TransformApp {
+    public static void main(String[] args) {
+        ConfigurableApplicationContext ctx = SpringApplication.run(TransformApp.class, args);
+        Map<String, MessageChannel> channels = ctx.getBeansOfType(MessageChannel.class);
+        log.warn("CHANNELS:");
+        int i = 0;
+        for (Map.Entry<String, MessageChannel> entry : channels.entrySet()) {
+            log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue());
+        }
+        log.warn("HANDLERS:");
+        i = 0;
+        Map<String, MessageHandler> endpoints = ctx.getBeansOfType(MessageHandler.class);
+        for (Map.Entry<String, MessageHandler> entry : endpoints.entrySet()) {
+            log.warn("{}. {}/{} -> {}", ++i, entry.getKey(), entry.getValue().getClass().getSimpleName(), entry.getValue());
+        }
+        Upcase upcase = ctx.getBean(Upcase.class);
+        Collection<Item> result = upcase.upcase(Arrays.asList(new Item("test"), new Item("new"), new Item("last")));
+//		Collection<String> 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<Item> upcase(Collection<Item> strings);
+    }
+
+    @Bean
+    public IntegrationFlow upcase() {
+        return f -> f
+                .split()
+                .transform(Transformers.toMap())
+                .<Map<String, String>, Map<String, String>>transform(map -> {
+                    map.replaceAll((k, v) -> v != null ? v.toUpperCase() : v);
+                    return map;
+                })
+                .transform(Transformers.fromMap(Item.class))
+                .aggregate();
+    }
+
+}

2025-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

2025-11/spring-30-endpoints-flow-components/pom.xml

@@ -0,0 +1,34 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>3.5.2</version>
+    </parent>
+
+    <groupId>ru.otus</groupId>
+    <artifactId>endpoints-flow-components</artifactId>
+    <version>1.0</version>
+
+    <packaging>pom</packaging>
+
+    <modules>
+        <module>endpoints-flow-components-exercise</module>
+        <module>endpoints-flow-components-solution</module>
+    </modules>
+
+    <properties>
+        <maven.compiler.target>17</maven.compiler.target>
+        <maven.compiler.source>17</maven.compiler.source>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <checkstyle-plugin.version>3.1.2</checkstyle-plugin.version>
+        <checkstyle.version>10.9.1</checkstyle.version>
+        <checkstyle.config.url>https://raw.githubusercontent.com/OtusTeam/Spring/master/checkstyle.xml
+        </checkstyle.config.url>
+    </properties>
+</project>