ftp使用者流下载seda队列在拆分器tariterator期间失败

Question

首先介绍一下我的要求：

• 从多个动态定义的只读FTP/SFTP站点下载大型tar.gz文件。
• 根据条目名的扩展名处理。tar.gz中的文件。
• 使用Camel 2.19.3

from("ftp://user@localhost/path?download=false&inProgressRepository=#inProgressRepo&idempotentRepository=#idemRepo&noop=true&readLock=changed&readLockMarkerFile=false&autoCreate=false&stepwise=false").to("seda:download?size=3&concurrentConsumers=3&blockWhenFull=true&purgeWhenStopping=true")

from("seda:download?size=3&concurrentConsumers=3&blockWhenFull=true&purgeWhenStopping=true")
.process({
    String fileName = exchange.getIn().getHeader(Exchange.FILE_NAME_ONLY, String.class);
    CamelContext context = exchange.getContext();
    ConsumerTemplate downloadConsumer = context.createConsumerTemplate();
    Producer unpackProducer = context.getRoute("unpack").getEndpoint().createProducer();
    Map<String,Object> parms = new HashMap<>();
    parms.put("fileName", fileName);
    parms.put("runLoggingLevel", "INFO");
    parms.put("consumer.bridgeErrorHandler", "true");
    parms.put("idempotentRepository", "#idemRepo");
    parms.put("noop", "true");
    parms.put("readLock", "changed");
    parms.put("readLockLoggingLevel", "INFO");
    parms.put("readLockMarkerFile", "false");
    parms.put("initialDelay", "0");
    parms.put("autoCreate", "false");
    parms.put("maximumReconnectAttempts", "0");
    parms.put("streamDownload", "true");
    parms.put("stepwise", "false");
    parms.put("throwExceptionOnConnectFailed", "true");
    parms.put("useList", "false");
    downloadConsumer.start();
    Exchange downloadExchange = downloadConsumer.receive(URISupport.normalizeUri(URISupport.appendParametersToURI("ftp://user@localhost/path", parms));
    unpackProducer.process(downloadExchange);
    if (downloadExchange.isFailed()) {
      LOGGER.error("unpack failed", downloadExchange.getException());
      exchange.setException(downloadExchange.getException());
    }
    downloadConsumer.doneUoW(downloadExchange);
    downloadConsumer.stop();
}

from("direct:unpack").routeId("unpack")
.convertBodyTo(InputStream.class, null)
  .split(new TarSplitter()).streaming()
    .choice()
      .when(header(FILE_NAME).regex(XML_FILTER))
        .unmarshal().jacksonxml(POJO.class)
        .endChoice()
      .when(header(FILE_NAME).regex(XML2_FILTER))
        .unmarshal().jacksonxml(POJO2.class)
        .endChoice()
      .end()
    .end()
  .to("file://...")

提前感谢您的帮助。

Answer 1

我的错误，需要添加binary=true。