Enterprise Integration Patterns

@Component
public class OrderRoutes extends RouteBuilder {
 
    @Override
    public void configure() {
 
        // Errors are part of the design, not an afterthought.
        errorHandler(deadLetterChannel("kafka:orders.dlq")
                .maximumRedeliveries(3)
                .redeliveryDelay(1000)
                .backOffMultiplier(2)
                .retryAttemptedLogLevel(LoggingLevel.WARN));
 
        // Content-based router
        from("kafka:incoming-orders")
            .routeId("order-router")
            .unmarshal().json(OrderMessage.class)
            .choice()
                .when(simple("${body.totalMinorUnits} > 100000"))
                    .to("direct:high-value")
                .when(simple("${body.region} == 'EU'"))
                    .to("direct:eu-processing")
                .otherwise()
                    .to("direct:standard");
 
        // Splitter with an aggregation strategy: one message per line, then
        // recombine the per-line results into one summary.
        from("direct:high-value")
            .split(simple("${body.lines}"), new OrderLineAggregationStrategy())
                .parallelProcessing()
                .to("direct:process-line")
            .end()
            .to("kafka:orders-processed");
 
        // Content enricher: the message lacks customer detail, so fetch it.
        from("direct:process-line")
            .enrich("direct:fetch-customer", (original, resource) -> {
                OrderLine line = original.getIn().getBody(OrderLine.class);
                Customer customer = resource.getIn().getBody(Customer.class);
                original.getIn().setBody(line.enrichedWith(customer));
                return original;
            })
            .to("bean:pricingService?method=calculate");
    }
}

from("kafka:line-results")
    .aggregate(header("orderId"), new OrderResultAggregationStrategy())
        .completionSize(header("expectedLineCount"))
        // Without this, a lost message means this group is held forever.
        .completionTimeout(30_000)
        .aggregationRepository(persistentRepository)   // survives a restart
    .to("direct:finalise-order");

@Configuration
public class OrderIntegrationConfig {
 
    @Bean
    public IntegrationFlow orderProcessingFlow() {
        return IntegrationFlow
                .from(Kafka.messageDrivenChannelAdapter(consumerFactory, "incoming-orders"))
                .transform(Transformers.fromJson(OrderMessage.class))
                .filter((OrderMessage order) -> !order.lines().isEmpty())
                .route(OrderMessage::region, mapping -> mapping
                        .subFlowMapping("EU", sub -> sub.channel("euChannel"))
                        .subFlowMapping("US", sub -> sub.channel("usChannel"))
                        .defaultOutputChannel("standardChannel"))
                .get();
    }
 
    @Bean
    public IntegrationFlow claimCheckFlow(ObjectStorage storage) {
        return IntegrationFlow.from("largePayloads")
                // Brokers are optimised for small messages. Store the payload,
                // pass a reference, and let the consumer fetch it.
                .handle((payload, headers) -> {
                    String key = storage.put((byte[]) payload);
                    return MessageBuilder.withPayload(new PayloadReference(key))
                            .copyHeaders(headers).build();
                })
                .channel("processedPayloads")
                .get();
    }
}

@CamelSpringBootTest
@MockEndpoints("kafka:*")
class OrderRoutesTest {
 
    @Autowired ProducerTemplate producer;
    @EndpointInject("mock:kafka:orders-processed") MockEndpoint processed;
 
    @Test
    void routesHighValueOrdersSeparately() throws Exception {
        processed.expectedMessageCount(1);
        processed.expectedHeaderReceived("priority", "HIGH");
 
        producer.sendBody("direct:incoming", highValueOrder());
 
        processed.assertIsSatisfied(5_000);
    }
}