Camel не удаляет файл camelLock после завершения теста маршрута

Question

Я изучаю верблюд Apache из книги «Camel in Action», и в настоящее время я перехожу к преобразованию данных. В частности, Content Enricher EIP. Я заметил, что когда я запускаю приведенный ниже код из книги, Camel создает файл fileName + .camelLock, но не удаляет его после завершения маршрута.

Есть что-то не так со стороны кода? Или он должен так работать?

import java.io.File; 

import org.apache.camel.Exchange; 
import org.apache.camel.builder.RouteBuilder; 
import org.apache.camel.processor.aggregate.AggregationStrategy; 
import org.apache.camel.test.junit4.CamelTestSupport; 
import org.junit.Test; 

public class OrderToCsvProcessorTest extends CamelTestSupport { 

    @Test 
    public void testOrderToCsvProcessor() throws Exception { 
     // this is the inhouse format we want to transform to CSV 
     String inhouse = "0000004444000001212320091208 1217@1478@2132"; 
     template.sendBodyAndHeader("direct:start", inhouse, "Date", "20091208"); 

     File file = new File("target/orders/received/report-20091208.csv"); 
     assertTrue("File should exist", file.exists()); 

     // compare the expected file content 
     String body = context.getTypeConverter().convertTo(String.class, file); 
     assertEquals("000000444,20091208,000001212,1217,1478,2132\nthis,is,sample,string", body); 
    } 

    @Override 
    protected RouteBuilder createRouteBuilder() throws Exception { 
     return new RouteBuilder() { 
      @Override 
      public void configure() throws Exception { 
       from("direct:start") 
         .process(new OrderToCsvProcessor()) 
         .pollEnrich("file://target/input?noop=true", 
           new AggregationStrategy() { 
            @Override 
            public Exchange aggregate(Exchange oldExchange, Exchange newExchange) { 
             if (newExchange == null) { 
              return oldExchange; 
             } 

             String http = oldExchange.getIn().getBody(String.class); 
             String ftp = newExchange.getIn().getBody(String.class); 

             String body = http + "\n" + ftp; 

             oldExchange.getIn().setBody(body); 

             return oldExchange; 
            } 
           }) 
         .to("file://target/orders/received?fileName=report-${header.Date}.csv"); 
      } 
     }; 
    } 
} 

процессора, который используется в коде:

import org.apache.camel.Exchange; 
import org.apache.camel.Processor; 

public class OrderToCsvProcessor implements Processor { 

    public void process(Exchange exchange) throws Exception { 
     String custom = exchange.getIn().getBody(String.class); 

     String id = custom.substring(0, 9); 
     String customerId = custom.substring(10, 19); 
     String date = custom.substring(20, 29); 
     String items = custom.substring(30); 
     String[] itemIds = items.split("@"); 

     StringBuilder csv = new StringBuilder(); 
     csv.append(id.trim()); 
     csv.append(",").append(date.trim()); 
     csv.append(",").append(customerId.trim()); 
     for (String item : itemIds) { 
      csv.append(",").append(item.trim()); 
     } 

     exchange.getIn().setBody(csv.toString()); 
    } 

} 

Answer 1

GenericFileOnCompletion отвечает за удаление файла блокировки. Вам нужно передать команды в AggregationStrategy именно так.

new AggregationStrategy() { 
    @Override 
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) { if (newExchange == null) { 
     return oldExchange; } 

    String http = oldExchange.getIn().getBody(String.class); String ftp = newExchange.getIn().getBody(String.class); 

    String body = http + "\n" + ftp; 

    oldExchange.getIn().setBody(body); 
    newExchange.handoverCompletions(oldExchange); 

    return oldExchange; } })