Верблюд для обработки нескольких файлов

Question

Я новичок на верблюде. У меня будет обработка файла с верблюдом, но я не нашел готового решения для моего случая. Я должен обработать несколько файлов вместе, если они существуют. Эти файлы загружаются в определенную папку с некоторыми задержками (пример: у нас есть два файла A.csv и B.csv, и A.csv загружается на 10 секунд позже, чем B.csv и наоборот). Также, если один файл отсутствует более определенного времени, мне нужно обработать только один файл. Может ли кто-нибудь помочь мне с выбором шаблона? Как я понимаю, я могу использовать верблюжий фильтр, чтобы быть уверенным, что у нас уже есть эти два файла A.csv и B.csv, и только после этого начинать обработку, но это не решает мою проблему.

Answer 1

Это Агрегатор EIP .

from("file:inputFolder")
        .aggregate(constant(true), AggregationStrategies.groupedExchange())
        .completionSize(2) //Wait for two files
        .completionTimeout(60000) //Or process single file, if completionSize was not fulfilled within one minute
            .to("log:do_something") //Here you can access List<Exchange> from message body

---

Для группировки сообщений вы можете использовать корреляцию Expression.Для вашего примера (группировать сообщения по префиксу имени файла перед _) это может быть что-то вроде этого:

private final Expression CORRELATION_EXPRESSION = new Expression() {
    @Override
    public <T> T evaluate(Exchange exchange, Class<T> type) {
        final String fileName = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
        final String correlationExpression = fileName.substring(0, fileName.indexOf('_'));
        return exchange.getContext().getTypeConverter().convertTo(
                type,
                correlationExpression
        );
    }
};

И передать его Aggregator:

from("file:inputDirectory")
    .aggregate(CORRELATION_EXPRESSION, AggregationStrategies.groupedExchange())
    ...

См. Эту сутьдля полного примера https://gist.github.com/bedlaj/a2a56aa9291bced8c0a8edebacaf22b0