Apache Camel高级路由与重试策略：实现动态消息分发与配置

// 假设这是您的RemappedMessage类
public class RemappedMessage {
    private String agentId;
    private String messageContent;
    // ... 其他字段
    // Getters and Setters
}

// 假设这是您的CustomerConfig类
public class CustomerConfig {
    private String customerId;
    private String targetUrl;
    private String oauthUrl;
    private String credentials;
    private String filterCriteria;
    // ... 其他字段
    // Getters and Setters
}

// 中间Bean：将RemappedMessage和List<CustomerConfig>组合成List<ImmutablePair>
public class PrepareCustomerMessagesBean {
    public List<ImmutablePair<RemappedMessage, CustomerConfig>> prepare(
            @Body RemappedMessage remappedMessage,
            @ExchangeProperty("customerConfigs") List<CustomerConfig> customerConfigs) { // 假设customerConfigs已作为Exchange属性设置

        List<ImmutablePair<RemappedMessage, CustomerConfig>> preparedMessages = new ArrayList<>();
        for (CustomerConfig config : customerConfigs) {
            // 注意：这里可以根据需要复制或修改 remappedMessage
            // 如果每个客户需要一个独立的、修改过的 RemappedMessage 副本，则在此处创建副本
            preparedMessages.add(ImmutablePair.of(remappedMessage, config));
        }
        return preparedMessages;
    }
}

// 假设 customerConfig 已作为 exchangeProperty.customerConfig 存在
// 并且 OAuthTokenRetrieverBean 已将令牌设置在 header.oauthToken 中

// 设置动态目标URL
.setHeader(Exchange.HTTP_URI, simple("${exchangeProperty.customerConfig.targetUrl}"))

// 设置OAuth认证头
.setHeader("Authorization", simple("Bearer ${header.oauthToken}"))

// 对于Basic Auth，假设您有一个 bean 来生成 base64 编码的凭据
// .bean(BasicAuthHeaderGenerator.class) // 此 bean 会将 "Basic <base64_encoded_credentials>" 放入一个 header
// .setHeader("Authorization", simple("${header.basicAuthValue}"))

// 1. 定义发送部分的错误处理策略
// 当 HTTP 操作失败时（例如 5xx 错误），进行重试
onException(org.apache.camel.component.http.HttpOperationFailedException.class)
    .handled(true) // 标记异常已处理，避免路由停止
    .maximumRedeliveries(3) // 最多重试3次
    .redeliveryDelay(2000L) // 每次重试间隔2秒
    .backOffMultiplier(2) // 每次重试延迟翻倍
    .log("发送消息到客户失败，正在重试。客户ID: ${exchangeProperty.customerConfig.customerId}, URL: ${header.CamelHttpUri}")
    .end();

from("activemq:queue:" + appConfig.getQueueName())
    .routeId("mainMessageProcessor")
    .bean(IncomingMessageConverter.class) // 转换为 RemappedMessage
    .bean(UserIdValidator.class) // 验证用户ID，如果失败，路由可能在此结束
    .bean(CustomerConfigRetrieverBean.class) // 根据 RemappedMessage 获取 List<CustomerConfig>
                                            // 确保此 Bean 将 List<CustomerConfig> 存储到 Exchange 属性中
                                            // 例如：exchange.setProperty("customerConfigs", customerConfigsList);

    // 准备要拆分的数据：将 RemappedMessage 与每个 CustomerConfig 组合
    .bean(PrepareCustomerMessagesBean.class) // 返回 List<ImmutablePair<RemappedMessage, CustomerConfig>>

    // 2. 拆分消息，为每个客户独立处理
    .split(body())
        .parallelProcessing() // 可选：并行处理每个客户消息
        .id("customerMessageSplitter")
        .process(exchange -> {
            // 从 ImmutablePair 中提取 RemappedMessage 和 CustomerConfig
            ImmutablePair<RemappedMessage, CustomerConfig> pair = exchange.getIn().getBody(ImmutablePair.class);
            exchange.getIn().setBody(pair.getLeft()); // 将 RemappedMessage 设置为消息体
            exchange.setProperty("customerConfig", pair.getRight()); // 将 CustomerConfig 设置为 Exchange 属性
        })

        // 3. 字段裁剪：根据客户配置移除不感兴趣的字段
        .bean(EndpointFieldsTailor.class) // 此 Bean 将使用消息体 (RemappedMessage) 和 Exchange 属性 (CustomerConfig)

        // 4. 条件过滤：检查消息内容是否满足客户标准
        .filter(simple("${body.messageContent} contains ${exchangeProperty.customerConfig.filterCriteria}")) // 示例过滤条件
            .id("customerMessageFilter")

            // 5. OAuth 认证与发送
            .bean(OAuthTokenRetrieverBean.class) // 获取 OAuth 令牌，并将其存储在消息头中（例如：header.oauthToken）

            // 动态设置目标 URL 和认证头
            .setHeader(Exchange.HTTP_URI, simple("${exchangeProperty.customerConfig.targetUrl}"))
            .setHeader("Authorization", simple("Bearer ${header.oauthToken}"))

            // 6. 发送消息到客户（此步骤将受 onException 策略保护，实现重试）
            // 使用 to("http://dummy")，实际 URL 由 CamelHttpUri 头决定
            .to("http://dummyhost/api")
            .log("成功发送消息到客户。客户ID: ${exchangeProperty.customerConfig.customerId}, URL: ${header.CamelHttpUri}")

        .end() // 结束 filter
    .end(); // 结束 split