Spring Integration Cheatsheet

package com.example.integration.order;

import org.springframework.integration.annotation.*;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import java.util.List;

// ── Gateway ──────────────────────────────────────────────────
@MessagingGateway
public interface OrderGateway {
    @Gateway(requestChannel = "orderInput", replyChannel = "orderComplete")
    OrderResult submit(Order order);
}

// ── Filter: Validate ─────────────────────────────────────────
@Component
class OrderValidator {
    @Filter(inputChannel = "orderInput", outputChannel = "validOrders",
            discardChannel = "invalidOrders")
    public boolean validate(Order order) {
        return order.getTotal() > 0
            && order.getEmail() != null
            && order.getItems() != null && !order.getItems().isEmpty();
    }
}

// ── Transformer: Enrich ──────────────────────────────────────
@Component
class OrderEnricher {
    @Transformer(inputChannel = "validOrders", outputChannel = "enrichedOrders")
    public Message<Order> enrich(Order order) {
        order.setTax(order.getTotal() * 0.077);
        order.setTimestamp(System.currentTimeMillis());
        String region = order.getCountry().equals("CH") ? "DOMESTIC" : "INTERNATIONAL";
        return MessageBuilder.withPayload(order)
            .setHeader("region", region)
            .build();
    }
}

// ── Router: Domestic vs International ────────────────────────
@Component
class RegionRouter {
    @Router(inputChannel = "enrichedOrders")
    public String route(@Header("region") String region) {
        return region.equals("DOMESTIC") ? "domesticChannel" : "internationalChannel";
    }
}

// ── ServiceActivator: Process ────────────────────────────────
@Component
class DomesticProcessor {
    @ServiceActivator(inputChannel = "domesticChannel", outputChannel = "notifyChannel")
    public Order process(Order order) {
        order.setStatus("PROCESSED");
        order.setTrackingNumber("CH-" + System.currentTimeMillis());
        return order;
    }
}

@Component
class InternationalProcessor {
    @ServiceActivator(inputChannel = "internationalChannel", outputChannel = "notifyChannel")
    public Order processIntl(Order order) {
        order.setCustomsCleared(true);
        order.setStatus("PROCESSED");
        order.setTrackingNumber("INTL-" + System.currentTimeMillis());
        return order;
    }
}

// ── ServiceActivator: Notify ─────────────────────────────────
@Component
class OrderNotifier {
    @ServiceActivator(inputChannel = "notifyChannel", outputChannel = "orderComplete")
    public OrderResult notify(Order order) {
        // emailService.send(order.getEmail(), "Order " + order.getId() + " confirmed");
        return new OrderResult(order.getId(), order.getStatus(), order.getTrackingNumber());
    }
}

package com.example.integration.etl;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.*;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.filters.SimplePatternFileListFilter;
import org.springframework.stereotype.Component;
import java.io.File;
import java.nio.file.Files;
import java.util.List;
import java.util.stream.Collectors;

@Configuration
class EtlFlowConfig {

    @Bean
    public FileReadingMessageSource csvSource() {
        FileReadingMessageSource source = new FileReadingMessageSource();
        source.setDirectory(new File("/data/incoming"));
        source.setFilter(new SimplePatternFileListFilter("*.csv"));
        return source;
    }

    @Bean
    public IntegrationFlow etlFlow(FileReadingMessageSource csvSource) {
        return IntegrationFlow
            .from(csvSource, c -> c.poller(Pollers.fixedDelay(10_000)))
            .channel("csvFiles")
            .get();
    }
}

@Component
class CsvLineSplitter {
    @Splitter(inputChannel = "csvFiles", outputChannel = "csvLines")
    public List<String> split(File file) throws Exception {
        return Files.readAllLines(file.toPath()).stream()
            .skip(1).filter(l -> !l.isBlank()).collect(Collectors.toList());
    }
}

@Component
class CsvToProductTransformer {
    @Transformer(inputChannel = "csvLines", outputChannel = "products")
    public Product transform(String line) {
        String[] p = line.split(",");
        return new Product(p[0].trim(), p[1].trim(),
            Double.parseDouble(p[2].trim()), Integer.parseInt(p[3].trim()));
    }
}

@Component
class ProductValidator {
    @Filter(inputChannel = "products", outputChannel = "validProducts",
            discardChannel = "invalidProducts")
    public boolean validate(Product p) {
        return p.getPrice() > 0 && p.getQuantity() >= 0 && !p.getSku().isBlank();
    }
}

@Configuration
class BatchAggregatorConfig {
    @Bean
    public IntegrationFlow batchAndInsert() {
        return IntegrationFlow.from("validProducts")
            .aggregate(a -> a
                .correlationStrategy(m -> "BATCH")  // single batch group
                .releaseStrategy(g -> g.size() >= 100)
                .groupTimeout(5000)
                .sendPartialResultOnExpiry(true))
            .handle("jdbcBulkInsertService", "insertBatch")
            .get();
    }
}

package com.example.integration.scatter;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.stereotype.Component;
import java.util.Comparator;
import java.util.List;

@Configuration
class ScatterGatherFlow {

    @Bean
    public IntegrationFlow bestPriceFlow() {
        return IntegrationFlow.from("priceRequest")
            .scatterGather(
                s -> s.applySequence(true)
                    .recipientFlow(f -> f.handle("supplierA", "quote"))
                    .recipientFlow(f -> f.handle("supplierB", "quote"))
                    .recipientFlow(f -> f.handle("supplierC", "quote")),
                g -> g.releaseStrategy(group -> group.size() == 3)
                    .groupTimeout(5000)
                    .sendPartialResultOnExpiry(true)
            )
            .handle((payload, headers) -> {
                @SuppressWarnings("unchecked")
                List<PriceQuote> quotes = (List<PriceQuote>) payload;
                return quotes.stream()
                    .min(Comparator.comparing(PriceQuote::getAmount))
                    .orElse(null);
            })
            .get();
    }
}

@Component
class SupplierA {
    @ServiceActivator
    public PriceQuote quote(PriceRequest req) {
        return new PriceQuote("SupplierA", req.getProductId(), 29.99);
    }
}

@Component
class SupplierB {
    @ServiceActivator
    public PriceQuote quote(PriceRequest req) {
        return new PriceQuote("SupplierB", req.getProductId(), 24.50);
    }
}

@Component
class SupplierC {
    @ServiceActivator
    public PriceQuote quote(PriceRequest req) {
        return new PriceQuote("SupplierC", req.getProductId(), 31.00);
    }
}

package com.example.integration.pubsub;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.stereotype.Component;

@Configuration
class EventBusConfig {

    @Bean
    public PublishSubscribeChannel eventBus() {
        return new PublishSubscribeChannel();
    }
}

@Component
class AuditLogger {
    @ServiceActivator(inputChannel = "eventBus")
    public void audit(DomainEvent event) {
        // auditRepository.save(new AuditEntry(event));
        System.out.printf("[AUDIT] %s: %s at %s%n",
            event.getType(), event.getEntityId(), event.getTimestamp());
    }
}

@Component
class EmailNotifier {
    @ServiceActivator(inputChannel = "eventBus")
    public void notify(DomainEvent event) {
        if ("ORDER_PLACED".equals(event.getType())) {
            // emailService.send(event.getUserEmail(), "Order confirmed!");
        }
    }
}

@Component
class MetricsCollector {
    @ServiceActivator(inputChannel = "eventBus")
    public void collect(DomainEvent event) {
        // meterRegistry.counter("events", "type", event.getType()).increment();
    }
}

@Component
class CacheInvalidator {
    @ServiceActivator(inputChannel = "eventBus")
    public void invalidate(DomainEvent event) {
        if ("PRODUCT_UPDATED".equals(event.getType())) {
            // cacheManager.getCache("products").evict(event.getEntityId());
        }
    }
}