Spring Integration

Full Reference: See adapters.md for File, HTTP, Kafka adapters, Error Handling, and Testing patterns.

Overview

┌─────────────────────────────────────────────────────────────────────┐
│                     Spring Integration Flow                         │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│   [Inbound]     [Channel]    [Transformer]    [Channel]   [Outbound] │
│   Adapter  ──▶  ════════ ──▶ ┌─────────┐ ──▶ ════════ ──▶ Adapter   │
│   (File,        (Queue/      │ Convert │     (Direct/    (DB,       │
│    HTTP,         Direct)     │ Enrich  │      PubSub)     Kafka,    │
│    Kafka)                    └─────────┘                   HTTP)    │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘

Quick Start

<!-- pom.xml -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-file</artifactId>
</dependency>

@Configuration
@EnableIntegration
public class IntegrationConfig {

    @Bean
    public IntegrationFlow fileProcessingFlow() {
        return IntegrationFlow
            .from(Files.inboundAdapter(new File("/input"))
                    .patternFilter("*.csv"),
                e -> e.poller(Pollers.fixedDelay(1000)))
            .transform(Files.toStringTransformer())
            .handle((payload, headers) -> {
                System.out.println("Processing: " + payload);
                return payload;
            })
            .get();
    }
}

---

Message & Channels

// Message structure
Message<String> message = MessageBuilder
    .withPayload("Hello Integration")
    .setHeader("contentType", "text/plain")
    .setHeader("priority", 1)
    .setCorrelationId(UUID.randomUUID())
    .build();

// Channel types
@Configuration
public class ChannelConfig {

    // Direct Channel (point-to-point, synchronous)
    @Bean
    public DirectChannel orderChannel() {
        return new DirectChannel();
    }

    // Queue Channel (point-to-point, async with buffer)
    @Bean
    public QueueChannel processingQueue() {
        return new QueueChannel(100);
    }

    // PublishSubscribe Channel (broadcast to all subscribers)
    @Bean
    public PublishSubscribeChannel notificationChannel() {
        return new PublishSubscribeChannel();
    }

    // Executor Channel (async with thread pool)
    @Bean
    public ExecutorChannel asyncChannel() {
        return new ExecutorChannel(Executors.newFixedThreadPool(10));
    }
}

---

Gateway (Entry Point)

@MessagingGateway
public interface OrderGateway {

    @Gateway(requestChannel = "orderChannel")
    void submitOrder(Order order);

    @Gateway(requestChannel = "orderChannel", replyChannel = "orderResponseChannel")
    OrderConfirmation submitOrderAndWait(Order order);

    @Gateway(requestChannel = "orderChannel", replyTimeout = 5000)
    @Async
    CompletableFuture<OrderConfirmation> submitOrderAsync(Order order);
}

@Service
@RequiredArgsConstructor
public class OrderService {
    private final OrderGateway orderGateway;

    public OrderConfirmation createOrder(CreateOrderRequest request) {
        Order order = mapToOrder(request);
        return orderGateway.submitOrderAndWait(order);
    }
}

---

Integration Flow DSL

@Bean
public IntegrationFlow orderFlow() {
    return IntegrationFlow
        .from("orderChannel")
        // Validation
        .filter(Order.class, order -> order.getTotal().compareTo(BigDecimal.ZERO) > 0,
            f -> f.discardChannel("invalidOrderChannel"))
        // Enrichment
        .enrich(e -> e
            .requestChannel("customerLookupChannel")
            .propertyExpression("customer", "payload"))
        // Transformation
        .transform(Order.class, order -> {
            order.setStatus(OrderStatus.VALIDATED);
            return order;
        })
        // Routing
        .<Order, String>route(order ->
                order.getTotal().compareTo(new BigDecimal("1000")) > 0
                    ? "highValueOrder" : "standardOrder",
            r -> r
                .subFlowMapping("highValueOrder", sf -> sf
                    .handle("priorityOrderHandler", "process"))
                .subFlowMapping("standardOrder", sf -> sf
                    .handle("standardOrderHandler", "process")))
        .handle("orderRepository", "save")
        .get();
}

---

Splitter & Aggregator

@Bean
public IntegrationFlow batchOrderFlow() {
    return IntegrationFlow
        .from("batchOrderChannel")
        // Split batch into individual orders
        .split(BatchOrder.class, BatchOrder::getOrders)
        .channel(c -> c.executor(Executors.newFixedThreadPool(5)))
        .handle("orderProcessor", "process")
        // Aggregate results
        .aggregate(a -> a
            .correlationStrategy(m -> m.getHeaders().get("correlationId"))
            .releaseStrategy(g -> g.size() == g.getSequenceSize())
            .outputProcessor(g -> new BatchResult(
                g.getMessages().stream()
                    .map(m -> (OrderResult) m.getPayload())
                    .toList()
            ))
            .expireGroupsUponCompletion(true)
            .groupTimeout(30000))
        .get();
}

---

Transformers

@Bean
public IntegrationFlow transformFlow() {
    return IntegrationFlow
        .from("inputChannel")
        .transform(String.class, String::toUpperCase)
        .transform(Transformers.toJson())
        .transform(Transformers.fromJson(Order.class))
        .enrichHeaders(h -> h
            .header("timestamp", Instant.now())
            .headerExpression("orderValue", "payload.total"))
        .channel("outputChannel")
        .get();
}

---

Routers

// Header-based router
@Bean
public IntegrationFlow headerRouterFlow() {
    return IntegrationFlow
        .from("inboundChannel")
        .<Message<?>, String>route(m -> m.getHeaders().get("type", String.class),
            r -> r
                .subFlowMapping("ORDER", sf -> sf.channel("orderChannel"))
                .subFlowMapping("PAYMENT", sf -> sf.channel("paymentChannel"))
                .defaultOutputChannel("unknownChannel"))
        .get();
}

// Payload-based router
@Bean
public IntegrationFlow payloadRouterFlow() {
    return IntegrationFlow
        .from("orderChannel")
        .<Order, OrderType>route(Order::getType,
            r -> r
                .subFlowMapping(OrderType.STANDARD, sf -> sf
                    .handle("standardProcessor", "process"))
                .subFlowMapping(OrderType.EXPRESS, sf -> sf
                    .handle("expressProcessor", "process")))
        .get();
}

---

Service Activator

@Component
public class OrderHandler {

    @ServiceActivator(inputChannel = "orderChannel", outputChannel = "resultChannel")
    public OrderResult processOrder(Order order,
                                    @Header("priority") int priority) {
        validateOrder(order);
        calculateTotals(order);
        return new OrderResult(order.getId(), "PROCESSED");
    }
}

// DSL equivalent
@Bean
public IntegrationFlow serviceActivatorFlow() {
    return IntegrationFlow
        .from("orderChannel")
        .handle(Order.class, (order, headers) -> {
            return new OrderResult(order.getId(), "PROCESSED");
        })
        .channel("resultChannel")
        .get();
}

---

Best Practices

Do	Don't
Use DSL for readable flows	Build flows with XML only
Configure error channels	Ignore errors silently
Implement retry with backoff	Fail on first error
Use queue channels for decoupling	Block in synchronous handlers
Monitor channel metrics	Deploy without observability

Production Checklist

• Error channels configured
• Retry policies implemented
• Queue channel capacities set
• Idempotent receivers where needed
• Proper timeout configuration
• Dead letter channels for failures

Anti-Patterns

Anti-Pattern	Problem	Solution
No error channel	Silent failures	Configure global errorChannel
Blocking in DirectChannel	Thread exhaustion	Use QueueChannel or async
Unbounded queue	Memory leak	Set queue capacity
Missing idempotency	Duplicate processing	Use IdempotentReceiverInterceptor

Quick Troubleshooting

Problem	Fix
No channel found	Define @Bean for channel
Dispatcher has no subscribers	Add handler to flow
Reply timeout	Use nullChannel or return value
Duplicate messages	Add IdempotentReceiverInterceptor

Reference Documentation

• Spring Integration Reference
• Enterprise Integration Patterns