如何在Apache Camel的from端点调用前设置Header？

问题背景 #

用户需调用Azure Storage Blob API，希望在发送请求前设置自定义Header。最初代码中setHeader放在from端点之后，无法在Azure调用前生效：

Stream.of("a", "b").forEach(id -> {
    from("azure-storage-blob://devstoreaccount1/hotfolder?serviceClient=#blobServiceClient&operation=listBlobs")
    .routeId(id)
    .autoStartup(false)
    .tracing()
     // 此操作无法在Azure调用前执行
    .setHeader("MyHeader", () -> id)
    .process(exchange -> {
        ArrayList<BlobItem> items = exchange.getIn().getBody(ArrayList.class);
        for (BlobItem blobItem : items) {
            System.out.println(blobItem.getName());
        }
    });
});

上下文启动后，用户通过以下方式启动路由：

final Route route = camelContext.getRoute("a");
ServiceHelper.startService(route.getConsumer());

后续尝试新增中间路由提前设置Header，但未产生效果：

from("direct:intermediate")
  .noAutoStartup()
  .routeId("intermediate")
  .to("direct:hf");

from("direct:hf")
  .setHeader("MyHeader", () -> id)
  .to("azure-storage-blob://devstoreaccount1/hotfolder?serviceClient=#blobServiceClient&operation=listBlobs")
  .process(exchange -> {
      ArrayList<BlobItem> items = exchange.getIn().getBody(ArrayList.class);
      for (BlobItem blobItem : items) {
          System.out.println(blobItem.getName());
      }
  });

final Route route = camelContext.getRoute("intermediate");
ServiceHelper.startService(route.getConsumer());

可行解决方案 #

1. 使用ProducerTemplate主动触发并传递Header #

当from绑定Azure Storage Blob这类消费者端点时，请求由端点主动发起，无法直接在调用前设置Header。改用ProducerTemplate主动发送请求，可在发送前配置Header：

// 初始化ProducerTemplate
ProducerTemplate template = camelContext.createProducerTemplate();

// 针对每个id发送请求，提前设置Header
Stream.of("a", "b").forEach(id -> {
    Map<String, Object> headers = new HashMap<>();
    headers.put("MyHeader", id);
    // 发送请求到Azure Storage Blob端点
    ArrayList<BlobItem> items = template.requestBodyAndHeaders(
        "azure-storage-blob://devstoreaccount1/hotfolder?serviceClient=#blobServiceClient&operation=listBlobs",
        null,
        headers,
        ArrayList.class
    );
    // 处理返回结果
    for (BlobItem blobItem : items) {
        System.out.println(blobItem.getName());
    }
});

2. 全局拦截Azure请求并注入Header #

通过interceptSendToEndpoint拦截所有发送到Azure Storage Blob的请求，在拦截逻辑中设置Header，无需修改原有路由结构：

// 全局拦截Azure Storage Blob请求，注入Header
interceptSendToEndpoint("azure-storage-blob:*")
    .process(exchange -> {
        // 从路由ID获取对应的标识设置Header
        String routeId = exchange.getFromRouteId();
        exchange.getIn().setHeader("MyHeader", routeId);
    })
    .continueRouting();

// 保留原有路由定义
Stream.of("a", "b").forEach(id -> {
    from("azure-storage-blob://devstoreaccount1/hotfolder?serviceClient=#blobServiceClient&operation=listBlobs")
    .routeId(id)
    .autoStartup(false)
    .tracing()
    .process(exchange -> {
        ArrayList<BlobItem> items = exchange.getIn().getBody(ArrayList.class);
        for (BlobItem blobItem : items) {
            System.out.println(blobItem.getName());
        }
    });
});

3. 修正中间路由的触发方式 #

之前中间路由无效是因为仅启动了路由但未触发请求。启动路由后，通过ProducerTemplate发送消息到direct端点，即可触发Header设置与Azure调用流程：

// 定义路由
from("direct:intermediate")
  .routeId("intermediate")
  .setHeader("MyHeader", () -> "a") // 可根据实际参数动态设置Header值
  .to("azure-storage-blob://devstoreaccount1/hotfolder?serviceClient=#blobServiceClient&operation=listBlobs")
  .process(exchange -> {
      ArrayList<BlobItem> items = exchange.getIn().getBody(ArrayList.class);
      for (BlobItem blobItem : items) {
          System.out.println(blobItem.getName());
      }
  });

// 启动路由并触发请求
final Route route = camelContext.getRoute("intermediate");
ServiceHelper.startService(route.getConsumer());

ProducerTemplate template = camelContext.createProducerTemplate();
template.sendBody("direct:intermediate", null);

内容的提问来源于stack exchange，提问作者kfaria