Проверка доступности файлов на основе времени в Apache Camel - Spring Boot ⇐ JAVA

[Anonymous]

Я реализую услугу, в которой мой исходный каталог постоянно контролируется, а файлы будут вытянуты и отправлены в пункт назначения. Я хочу достичь одной функции, где доступ к файлам, основанную на времени. Если файл также не появился и в окне буфера, запустите электронное письмо. Всякий раз, когда происходит время, он приостанавливает файлы, чтобы извлечь из источника, то есть, если вы копируете какой -либо файл, чтобы источник, он не тянет и не подталкивает к месту назначения. Это своего рода замораживание. Никакое действие не выполняет. Но через 5 минут все нормально, т.е. Все файлы будут вытянуты и отправлены в пункт назначения.@Component
public class LocalToLocalRoute implements FileTransferRouteResolver {

private final FileTransferProperties properties;
private final LocalCacheService localCacheService;
private final List filesToMonitor;

public LocalToLocalRoute(FileTransferProperties properties, LocalCacheService localCacheService) {
this.properties = properties;
this.localCacheService = localCacheService;
this.filesToMonitor = getFilesToMonitor(properties);
}

@Override
public void configureRoute(RouteBuilder routeBuilder) throws Exception {
LocalToLocal localtolocalProps = properties.getLocaltolocal();
List allowedExtensions = localtolocalProps.getFilter();
Long retryDelay = properties.getRetryDelay();
Integer maxRetryLimit = properties.getMaxRetryLimit();

AftUri fromUri = AftUri.builder("file")
.path(localtolocalProps.getSource())
.addParams(localtolocalProps.getFromCamelOptions())
.build();

AftUri toUri = AftUri.builder("file")
.path(localtolocalProps.getDestination())
.addParams(localtolocalProps.getToCamelOptions())
.build();

AftUri errorUri = AftUri.builder("file")
.path(localtolocalProps.getSource() + File.separator + ".error")
.build();

transfer(routeBuilder, fromUri, toUri, errorUri, this.getClass().getSimpleName(), retryDelay, maxRetryLimit, allowedExtensions, localCacheService, filesToMonitor);

if (localtolocalProps.getFilechecks() == null) {
return;
}

for (LocalToLocal.FileCheck fileCheck : localtolocalProps.getFilechecks()) {
String fileName = fileCheck.getFilename();
String checkTime = fileCheck.getCheckTime();
int bufferMinutes = fileCheck.getBufferMinutes();
long fileExpectedTime = convertTimeToMilliseconds(checkTime);
String cron = toCron(checkTime);
String quartzId = "quartz://fileCheckGroup/" + fileName + "?cron=" + cron;
invokeFileCheck(routeBuilder, fileName, checkTime, fileExpectedTime, quartzId);
invokeBufferedCheck(routeBuilder, fileName, bufferMinutes, fileExpectedTime);
}

}

private void invokeFileCheck(RouteBuilder routeBuilder, String fileName, String checkTime, long fileExpectedTime, String quartzId) {
routeBuilder.from(quartzId)
.routeId("FileCheckRoute-" + fileName + "-" + UUID.randomUUID())
.log("Checking for file: " + fileName + " at scheduled time: " + checkTime)
.process(exchange -> {
boolean containsKey = localCacheService.containsKey(fileName);
if (containsKey) {
Long fileProcessedTime = localCacheService.get(fileName);
if (fileProcessedTime >= fileExpectedTime) {
exchange.setProperty("fileName", fileName);
exchange.setProperty("fileExists", true);
}
} else {
exchange.setProperty("fileName", fileName);
exchange.setProperty("fileExists", false);
}
})
.choice()
.when(routeBuilder.simple("${exchangeProperty.fileExists} == false"))
.log("File ${exchangeProperty.fileName} is missing. Waiting for buffer window...")
.to("seda:bufferDelay-" + fileName)
.otherwise()
.log("File ${exchangeProperty.fileName} found. No need to wait.")
.process(e -> localCacheService.remove(fileName))
.end();
}

private void invokeBufferedCheck(RouteBuilder routeBuilder, String fileName, int bufferMinutes, long fileExpectedTime) {
routeBuilder.from("seda:bufferDelay-" + fileName)
.routeId("BufferedCheckRoute-" + fileName + "-" + UUID.randomUUID())
.log("Buffer check started for file: " + fileName + ", waiting for {} minutes"+ bufferMinutes)
.delayer(bufferMinutes * 60 * 1000L)
.log("Buffer delay expired for file: " + fileName)
.process(exchange -> {
Long fileProcessedTime = localCacheService.get(fileName);
boolean stillMissing = (fileProcessedTime == null) || (fileProcessedTime < fileExpectedTime);
exchange.setProperty("stillMissing", stillMissing);
exchange.setProperty("fileName", fileName);
})
.choice()
.when(routeBuilder.simple("${exchangeProperty.stillMissing} == true"))
.log("ALERT: File ${exchangeProperty.fileName} still missing after buffer wait.")
.otherwise()
.log("File ${exchangeProperty.fileName} arrived within buffer window.")
.process(e -> localCacheService.remove(fileName))
.end();
}

public static long convertTimeToMilliseconds(String time) {
String[] parts = time.split(":");
int hours = Integer.parseInt(parts[0]);
int minutes = Integer.parseInt(parts[1]);
return (hours * 60L + minutes) * 60 * 1000;
}

private String toCron(String time) {
String[] parts = time.split(":");
int hour = Integer.parseInt(parts[0]);
int minute = Integer.parseInt(parts[1]);
return String.format("0 %d %d * * ?", minute, hour); // Quartz cron format
}
}
< /code>


import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public interface FileTransferRouteResolver {

final Logger LOGGER = LoggerFactory.getLogger(FileTransferRouteResolver.class);
void configureRoute(RouteBuilder routeBuilder) throws Exception;

default void transfer(RouteBuilder routeBuilder, AftUri fromUri, AftUri toUri,AftUri errorUri, String routeName, int retryDelay, int maxDeliveries, List allowedExtensions) {
String filterExpression = buildExtensionFilter(allowedExtensions);
routeBuilder.onException(Exception.class)
.redeliveryDelay(retryDelay)
.maximumRedeliveries(maxDeliveries)
.retryAttemptedLogLevel(org.apache.camel.LoggingLevel.INFO)
.handled(true)
.log("Exception occurred while moving file: ${file:name}")
.to(errorUri.build());

routeBuilder.from(fromUri.build())
.routeId(routeName + "-" + UUID.randomUUID())
.log("Detected new file: ${file:name}")
.filter().simple(filterExpression)
.log("Accepted file: ${file:name}")
.to(toUri.build())
.log("File: ${file:name} moved to destination")
.end();
}

default void transfer(RouteBuilder routeBuilder, AftUri fromUri, AftUri toUri,AftUri errorUri, String routeName, Long retryDelay, Integer maxDeliveries, List allowedExtensions, LocalCacheService localCacheService, List filesToMonitor) {
String filterExpression = buildExtensionFilter(allowedExtensions);

routeBuilder.onException(Exception.class)
.redeliveryDelay(retryDelay)
.maximumRedeliveries(maxDeliveries)
.retryAttemptedLogLevel(org.apache.camel.LoggingLevel.INFO)
.log("Exception occurred while moving file: ${file:name}")
.to(errorUri.build());

routeBuilder.from(fromUri.build())
.routeId(routeName + "-" + UUID.randomUUID())
.log("Detected new file: ${file:name}")
.filter().simple(filterExpression)
.log("Accepted file: ${file:name}")
.process(exchange -> {
String fileName = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
if(filesToMonitor.contains(fileName)) {
localCacheService.put(fileName, System.currentTimeMillis());
LOGGER.info("File '{}' detected and recorded in cache at {}.", fileName, System.currentTimeMillis());
}
})
.to(toUri.build())
.log("File: ${file:name} moved to destination")
.end();
}

default String buildExtensionFilter(List extensions) {
if (extensions == null || extensions.isEmpty() || extensions.get(0).equals("*")) {
return "true"; // Allow all files if no filter is defined
}

return extensions.stream()
.filter(ext -> ext != null && !ext.isBlank())
.map(ext -> ext.startsWith(".") ? ext.substring(1) : ext) // Remove leading dot
.flatMap(ext -> Stream.of(
"${file:ext} == '" + ext.toLowerCase() + "'",
"${file:ext} == '" + ext.toUpperCase() + "'"
))
.reduce((a, b) -> a + " || " + b)
.orElse("true");
}

default List getFilesToMonitor(FileTransferProperties properties) {
List filechecks = properties.getLocaltolocal().getFilechecks();
if(filechecks.isEmpty()) {
return null;
} else {
return filechecks.stream().map(e -> e.getFilename()).collect(Collectors.toList());
}
}

}



Подробнее здесь: https://stackoverflow.com/questions/796 ... pring-boot