如何拆分聚合器创建的附件

Question

我在输入文件夹中有一堆相关的 PDF 和 XML 文件：

1.PDF - 1.XML
2.PDF - 2.XML
3.PDF - 3.XML
etc.

PDF文件是扫描文档，相关的XML文件包含与PDF文件相关的索引信息。

我使用 Apache Camel 轮询文件夹中的 XML 和 PDF 文件，然后在我有相关对时将其发布到输出文件夹。理想情况下，我想最后发布 XML 文件。

以下代码似乎可以正确提取这对文件，但是当我再次尝试拆分文件时收到错误消息。

@Grab('org.apache.camel:camel-core:2.13.0')
@Grab('org.slf4j:slf4j-simple:1.6.6')
import org.apache.camel.*
import org.apache.camel.impl.*
import org.apache.camel.builder.*
import org.apache.camel.util.jndi.*

def jndiContext = new JndiContext();

def dataDir = "/${System.properties['user.home']}/test/file-poller-demo"
def camelContext = new DefaultCamelContext(jndiContext)
camelContext.addRoutes(new RouteBuilder() {
    def void configure() {
    from("file://${dataDir}/in")
        .aggregate(simple('${file:onlyname.noext}')).completionSize(2).groupExchanges()
        .split(body())
        .to("file://${dataDir}/out")
}
})
camelContext.start()
addShutdownHook{ camelContext.stop() }
synchronized(this){ this.wait() }

错误信息：

没有可用的类型：java.io.InputStream 但有值：Exchange[1.xml] 类型：org.apache.camel.impl.DefaultExchange on：1.txt。原因：没有类型转换器可用于从类型 org.apache.camel.impl.DefaultExchange 转换为所需类型：java.io.InputStream，值为 Exchange[1.xml]。交换[1.txt]。

Answer 1

以下是 Java DSL 中的工作示例。

您可以添加一个额外的处理器以在交换聚合后设置正确的输出正文，但在文件使用者中没有 noop=true 选项，第一个文件将在聚合后移动到 .camel 并且文件生产者将抛出一个表达式当它尝试将其写入发件箱文件夹时。

from("file:data/inbox?noop=true")
        .aggregate(simple("${file:onlyname.noext}")).completionSize(2).groupExchanges()
        .split(body())
        .process(new Processor() {
         @Override
         public void process(Exchange exchange) throws Exception {
            Exchange exchangeIn = exchange.getIn().getBody(Exchange.class);
            exchange.getIn().setBody(exchangeIn.getIn().getBody(GenericFile.class));
            exchange.getIn().setHeaders(exchangeIn.getIn().getHeaders());
         }
         })
         .to("file:data/outbox/");

我认为将文件加载到内存数据存储中并在之后聚合会更好。

在以下路线中，文件被使用后，它被添加到内存数据存储中。聚合完成后，将从拆分器中的数据存储中检索文件，并在最后一次交换中清除数据存储。文件使用者中的删除选项是可选的。如果未添加，文件将在完成后移动到 .camel。

from("file:data/inbox?delete=true")
.setProperty("correlationId", simple("${file:onlyname.noext}"))
.beanRef("dataStoreBean","addToStore")
.aggregate(property("correlationId")).completionSize(2).groupExchanges()
.split(body())
.beanRef("dataStoreBean", "getFromStore")
.to("file:data/outbox/");

数据存储 bean 在哪里：

public class DataStoreBean {

    Map<String,byte[]> dataStore = new HashMap<String,byte[]>();

    public void addToStore(Exchange exchange) {
        dataStore.put(exchange.getIn().getHeader(Exchange.FILE_NAME,StrinTheg.class),exchange.getIn().getBody(byte[].class));
        exchange.getOut().setBody(exchange.getIn().getHeader(Exchange.FILE_NAME,String.class));
    }

     public void getFromStore(Exchange exchange) {
    Exchange fileExchange = exchange.getIn().getBody(Exchange.class);
    String fileName = fileExchange.getIn().getBody(String.class);
    exchange.getOut().setHeader(Exchange.FILE_NAME, fileName);
    exchange.getOut().setBody(dataStore.get(fileName));
    if(exchange.getProperty("CamelSplitComplete", Boolean.class)) {
        String correlationId = fileExchange.getProperty("correlationId", String.class);
        dataStore.remove(correlationId+".pdf");
        dataStore.remove(correlationId+".xml");
    }
}
}