Apache верблюд (?, Который EIP использовать здесь) агрегатный/обогащают все строки с повторяющимися данными заголовка

Question

У меня есть CSV я обработка отформатирован следующим образом:

01,H2,H3 
02,B2,B3,B4,B5 
02,B2,B3,B4,B5 
02,B2,B3,B4,B5 
02,B2,B3,B4,B5 
01,H2,H3 
02,B2,B3,B4,B5 
02,B2,B3,B4,B5 
01,H2,H3 
02,B2,B3,B4,B5 
02,B2,B3,B4,B5 
02,B2,B3,B4,B5 

01 указывает строку заголовка и 02 указывает строку тела ,

мне нужно взять данные заголовка и добавить его к сообщениям тела, так что я в конечном итоге отправки сообщений, как это:

H2,H3,B2,B3,B4,B5 
H2,H3,B2,B3,B4,B5 
H2,H3,B2,B3,B4,B5 

Я попытался объединить, но это не кажется, право EIP в этот сценарий, поскольку я просто объединяю одно и то же сообщение снова и снова, а не несколько сообщений в один ... На фундаментальном уровне мне нужен доступ к данным заголовка для обработки тела (по правде говоря, это всего лишь одно поле). Я просто не знаю, как установить переменную, поскольку заголовки и свойства очищаются на каждом обмене. Какие-нибудь советы? Заранее спасибо. Дайте мне знать, если это поможет увидеть верблюжий маршрут в его нынешнем виде.

Вот верблюд маршрут, который может помочь:

from("direct:inventory") 
    .split(body().tokenize("\n")).streaming() 
    .throttle(100) 
    .choice() 
     .when(property("CamelSplitComplete").isEqualTo(true)) 
     .log("Processed ${property.CamelSplitSize} updates") 
     .end() 
    .unmarshal(csv) 
      .log("${body}") 
      .aggregate(header("CamelFileLastModified"), new InventoryAggregationStrategy()) 
      .completionPredicate(header("aggregationComplete").isEqualTo(true)) 
      .to("freemarker://templates/inventory.ftl") 
      .unmarshal().string("UTF-8") 
      .unmarshal().json(JsonLibrary.Jackson) 
      .convertBodyTo(JsonObject.class) 
      .to("endpoint"); 

Here is the spec for the data

Answer 1

Вы всегда можете просто пойти с простым подходом и использовать компонент.

public class CamelHeadersAndRows { 
    public static class HeaderBean { 
     String header = null; 
     public void setHeader(String body) { 
      header = body.substring("01,".length()); 
     } 
     public String useHeader(String body) { 
      return header + "," + body.substring("02,".length()); 
     } 
    } 

    public static void main(String[] args) throws Exception { 
     final HeaderBean headerBean = new HeaderBean(); 

     Main camelMain = new Main(); 
     camelMain.addRouteBuilder(new RouteBuilder() { 
      @Override 
      public void configure() throws Exception { 
       from("timer:foo?period=1s&repeatCount=1") 
         .setBody(constant(
           "01,H2,H3\n" + 
           "02,B2,B3,B4,B5\n" + 
           "02,B2,B3,B4,B5\n" + 
           "02,B2,B3,B4,B5\n" + 
           "02,B2,B3,B4,B5\n" + 
           "01,H2,H3\n" + 
           "02,B2,B3,B4,B5\n" + 
           "02,B2,B3,B4,B5\n" + 
           "01,H2,H3\n" + 
           "02,B2,B3,B4,B5\n" + 
           "02,B2,B3,B4,B5\n" + 
           "02,B2,B3,B4,B5" 
         )) 
         .to("direct:inventory"); 

       from("direct:inventory") 
         .split(body().tokenize("\n")).streaming() 
         .choice() 
          .when().simple("${body} regex '^01.*'") 
           .bean(headerBean, "setHeader") 
           .stop() 
          .otherwise() 
           .bean(headerBean, "useHeader") 
         .end() 
         .log("message: ${body}") 
       ; 
      } 
     }); 
     camelMain.run(); 
    } 

}