Apache Camel：使用SFTP组件一次下载多个文件

Question

40_20200313_0cd6963f-bf5b-4eb0-b310-255a23ed778e_p.dat
123_20200313_0cd6963f-bf5b-4eb0-b310-255a23ed778e_p.dat
etc.

    private static String regex() {
        return "(22|23|24|25|26|28|29|32|35|40|41|46|52|70|85|88|123)_(?:.*)_p.dat";
    }

    private static String sftpComponent() {
        return "sftp://transit.ergogroup.no/Eyeshare/From_Eyeshare_Test"
                + "?username=Eyeshare_test"
                + "&password=epw3ePOugG" // Stored on wildfly server
                + "&download=true" //Shall be read chunk by chunk to avoid heap space issues. Earlier download=true was used: Harpreet
                + "&useList=true"
                + "&stepwise=false"
                + "&disconnect=true"
                + "&passiveMode=true"
                + "&reconnectDelay=10000"
//              + "&bridgeErrorHandler=true"
                + "&delay=300000"
                //+ "&fileName=" + sftpFileName
//              + "&include=kiki\\.txt"
//              + "&include=40_*_p\\.dat"sss
                + "&include="+regex()
                + "&preMove=$simple{file:onlyname}.$simple{date:now:yyyy-MM-dd'T'hh-mm-ss}.processing"
                + "&move=$simple{file:onlyname.noext}.$simple{date:now:yyyy-MM-dd'T'hh-mm-ss}.success"
                + "&moveFailed=$simple{file:onlyname.noext}.$simple{date:now:yyyy-MM-dd'T'hh-mm-ss}.failed";
//              + "&idempotentRepository=#infinispan"
//              + "&readLockRemoveOnCommit=true";
    }

   from(sftpComponent()).log("CHU").to(archiveReceivedFile())

代码看起来很好，但输出不是。有人建议吗

Answer 1

下面是聚合器的一些示例：

from("file:///somePath/consume/?maxMessagesPerPoll=2&delay=5000")
            .aggregate(constant(true), new ZipAggregationStrategy()).completion(exchange -> exchange.getProperty("CamelBatchComplete", Boolean.class))
            .to("file:///somePath/produce/")

这里的maxMessagesPerPoll定义了要存档的文件数量。但是，如果文件夹中的文件数低于maxMessagesPerPoll值，它将等待丢失的文件进行完整归档。下面是ZipAggregationStrategy的示例：

private static class ZipAggregationStrategy implements AggregationStrategy {
    private ZipOutputStream zipOutputStream;
    private ByteArrayOutputStream out;
    @Override
    public Exchange aggregate(final Exchange oldExchange, final Exchange newExchange) {
        try {
            if (oldExchange == null) {
                out = new ByteArrayOutputStream();
                zipOutputStream = new ZipOutputStream(out);
            }
            createEntry(newExchange);
            return newExchange;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    private void createEntry(final Exchange exchange) throws Exception {
        final ZipEntry zipEntry = new ZipEntry(exchange.getIn().getHeader(Exchange.FILE_NAME, String.class));
        zipOutputStream.putNextEntry(zipEntry);
        byte[] bytes = new byte[1024];
        int length;
        try (InputStream body = exchange.getIn().getBody(InputStream.class)) {
            while ((length = body.read(bytes)) >= 0) {
                zipOutputStream.write(bytes, 0, length);
            }
        }
    }
    @Override
    public void onCompletion(final Exchange exchange) {
        try {
            zipOutputStream.close();
            exchange.getIn().setBody(new ByteArrayInputStream(out.toByteArray()));
            exchange.getIn().setHeader(Exchange.FILE_NAME, "someArchive.zip");
        }catch (Exception e){
            throw new RuntimeException(e);
        } finally {
            try {
                out.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

这是内存中的例子。您可以改进它，例如使用临时文件。您可以根据自己的逻辑创建自己的完成谓词。