Camel - 当消费目录不存在时停止路由答案

【问题标题】：Camel - Stop route when the consuming directory not existsCamel - 当消费目录不存在时停止路由
【发布时间】：2016-09-17 16:47:17
【问题描述】：

我有一个 SFTP 路由（在 Spring XML 中），它的起始路径以每日更改的目录（即 /yyyyMMdd）结束，当autoCreate=true 或路由开始时目录存在时一切正常。但是如果不存在，我不允许创建目录！

当目录存在时，路由获取文件并自行终止。

当目录不存在时，路由将永久轮询并发出警告（即org.apache.camel.component.file.GenericFileOperationFailedException: Cannot change directory to: 20160917）并且永不停止。

如何避免这种行为（例如，将警告转换为空消息或异常或...）？我已经对startingDirectoryMustExist、consumer.bridgeErrorHandler 和许多其他人进行了实验，但没有任何成功。

简化路线（开始前，将实际日期填入elmu.sftp.importDir属性）：

    <from
        uri="sftp://{{elmu.sftp.host}}:{{elmu.sftp.port}}{{elmu.sftp.importDir}}?username={{elmu.sftp.userName}}&amp;password={{elmu.sftp.password}}&amp;
        autoCreate=false&amp;preferredAuthentications=password&amp;binary=true&amp;include={{elmu.importMask}}&amp;initialDelay=100&amp;
        noop=true&amp;sortBy=file:name&amp;sendEmptyMessageWhenIdle=true"/>
    <choice>
        <when>
            <simple>${body} != null</simple>
            ... a lot of stuff ...
            <to uri="bean:shutdownRoute" />
        </when>
        <otherwise>
            <to uri="bean:shutdownRoute" />
        </otherwise>
    </choice>

directoryMustExist=true and startingDirectoryMustExist=true 的结果是一个无限循环（轮询），并带有以下警告：

08:30:14,658 WARN  SftpConsumer - Consumer Consumer[sftp://xxx.xxx.xx:22/DBHtest/ELMUteszt/Kiadott_adatok/20160918?autoCreate=false&binary=true&directoryMustExist=true&include=%5E.*%24&initialDelay=100&noop=true&password=xxxxxx&preferredAuthentications=password&sendEmptyMessageWhenIdle=true&sortBy=file%3Aname&startingDirectoryMustExist=true&username=xxx] failed polling endpoint: Endpoint[sftp://xxx:22/DBHtest/ELMUteszt/Kiadott_adatok/20160918?autoCreate=false&binary=true&directoryMustExist=true&include=%5E.*%24&initialDelay=100&noop=true&password=xxxxxx&preferredAuthentications=password&sendEmptyMessageWhenIdle=true&sortBy=file%3Aname&startingDirectoryMustExist=true&username=xxx]. Will try again at next poll. Caused by: [org.apache.camel.component.file.GenericFileOperationFailedException - Cannot change directory to: 20160918] 
org.apache.camel.component.file.GenericFileOperationFailedException: Cannot change directory to: 20160918
    at org.apache.camel.component.file.remote.SftpOperations.doChangeDirectory(SftpOperations.java:576)
    at org.apache.camel.component.file.remote.SftpOperations.changeCurrentDirectory(SftpOperations.java:564)
    at org.apache.camel.component.file.remote.SftpConsumer.doPollDirectory(SftpConsumer.java:107)
    at org.apache.camel.component.file.remote.SftpConsumer.pollDirectory(SftpConsumer.java:79)
    at org.apache.camel.component.file.GenericFileConsumer.poll(GenericFileConsumer.java:131)
    at org.apache.camel.impl.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:175)
    at org.apache.camel.impl.ScheduledPollConsumer.run(ScheduledPollConsumer.java:102)
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
    at java.util.concurrent.FutureTask.runAndReset(Unknown Source)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(Unknown Source)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
Caused by: 2: No such file
    at com.jcraft.jsch.ChannelSftp.throwStatusError(ChannelSftp.java:2846)
    at com.jcraft.jsch.ChannelSftp._realpath(ChannelSftp.java:2340)
    at com.jcraft.jsch.ChannelSftp.cd(ChannelSftp.java:342)
    at org.apache.camel.component.file.remote.SftpOperations.doChangeDirectory(SftpOperations.java:574)
    ... 13 more

它不适用于stepwise=false：

11:52:19,210 WARN  SftpConsumer - Consumer Consumer[sftp://xxx:22/DBHtest/ELMUteszt/Kiadott_adatok/20160918?autoCreate=false&binary=true&directoryMustExist=true&include=%5E.*%24&initialDelay=100&noop=true&password=xxxxxx&preferredAuthentications=password&sendEmptyMessageWhenIdle=true&sortBy=file%3Aname&startingDirectoryMustExist=true&stepwise=false&username=xxx] failed polling endpoint: Endpoint[sftp://xxx:22/DBHtest/ELMUteszt/Kiadott_adatok/20160918?autoCreate=false&binary=true&directoryMustExist=true&include=%5E.*%24&initialDelay=100&noop=true&password=xxxxxx&preferredAuthentications=password&sendEmptyMessageWhenIdle=true&sortBy=file%3Aname&startingDirectoryMustExist=true&stepwise=false&username=xxx]. Will try again at next poll. Caused by: [org.apache.camel.component.file.GenericFileOperationFailedException - Cannot list directory: DBHtest/ELMUteszt/Kiadott_adatok/20160918] 
org.apache.camel.component.file.GenericFileOperationFailedException: Cannot list directory: DBHtest/ELMUteszt/Kiadott_adatok/20160918

更新（根据@ruffp 的回答）：

我尝试设置自定义 PollingConsumerPollStrategy，但我无法从中停止路由。只有第三行（注释行）停止了路线，但我有几条路线，我不知道实际路线的名称。我怎样才能得到它？

    @Override
    public boolean rollback(Consumer consumer, Endpoint endpoint, int retryCounter, Exception cause) throws Exception {
        consumer.getEndpoint().stop(); // 1
        consumer.stop();  // 2
        consumer.getEndpoint().getCamelContext().stopRoute(route???);  // 3
        return false;
    }

【问题讨论】：

您能按原样分享您的路线吗？
我已经分享了路线。
您是否尝试过添加 directoryMustExist=true 和 startingDirectoryMustExist=true？添加之后，能不能显示一下得到的日志输出？
我也会添加 stepwise=false 。我似乎正在尝试对该文件夹执行 cd，但由于该文件夹不存在，因此会发出警告。您可以添加此参数并再次发布日志输出吗？我怀疑这是主要问题。如果可行，我可以将其添加为答案。
您确定您的用户具有执行基本文件操作的正确权限吗？您的用户应该能够列出目录。

标签： apache-camel

【解决方案1】：

最后我用consumer.exceptionHandler 解决了这个问题。但似乎，此选项不在可用选项列表中（http://camel.apache.org/file2.html，我一遍又一遍地阅读）只是在大页面底部的示例中提到过一次。不幸的是，它是如此“隐藏”，以至于我到现在都错过了。

我建立了一个新类并实现了handleExceptions方法：

public class DirNotExistsExHandler implements ExceptionHandler

获取异常并决定做什么。在上下文中，我做了一个 bean 定义：

<bean id="dirNotExistsExHandler" class="hu.dbit.eleo.DirNotExistsExHandler" />

在消费者中，将 bean 传递给处理程序：

consumer.exceptionHandler=#dirNotExistsExHandler

非常感谢您的帮助！

【讨论】：

【解决方案2】：

在我看来，最好的方法是：

在sftp 端点上启用选项throwExceptionOnConnectFailed，如下所示：

sftp://{{elmu.sftp.host}}:{{elmu.sftp.port}}{{elmu.sftp.importDir}}?username={{elmu.sftp.userName}}&amp;password={{elmu.sftp.password}}&amp;
autoCreate=false&amp;preferredAuthentications=password&amp;binary=true&amp;include={{elmu.importMask}}&amp;initialDelay=100&amp;
noop=true&amp;sortBy=file:name&amp;sendEmptyMessageWhenIdle=true&amp;throwExceptionOnConnectFailed=true

这可以帮助管理异常并通过骆驼路线 onException 捕获它。但是我不确定你的情况是否真的有必要。

对你的连接异常进行特殊处理

// 2a) Solution to redirect an empty body to a destination
onException(GenericFileOperationFailedException.class)
            .handled(true)
            .log(LoggingLevel.INFO, "Source directory not present: send empty body to shutdown route...")
            .setBody(null)
            .to("bean:shutdownRoute");

或者其他方式：

// 2b)Solution to just stop the processing without log in warn or error
onException(GenericFileOperationFailedException.class)
    .handled(true)
    .log(LoggingLevel.INFO, "Source directory not present: stop the process and wait until next time...")
    .stop();

更新：我找到了this post，显然没有其他方法可以实现自己的PollingConsumerPollStrategy，因为GenericFileOperationFailedException 显然是在默认实现中处理的。

【讨论】：

谢谢，可惜不行，因为没有异常（在路由层面）。我认为，即使throwExceptionOnConnectFailed=true，骆驼也会“捕捉并翻译”警告异常。
我已经阅读了您编辑的帖子，看来这可能是实现我的目标的方法。但不幸的是，这似乎非常困难（并且不知所措，如果我可以说的话）而且我不知道如何开始在 xml（spring）中实现它。但我正在努力管理它。谢谢！
我已经更新了我原来的问题，请看一下。谢谢。

【解决方案3】：

使用 doCatch 块处理异常。如果没有抛出异常，则使用您的代码检查文件是否存在并手动抛出异常。

http://camel.apache.org/try-catch-finally.html

【讨论】：

对不起，我不明白你的意思。你可以再详细一点吗？如何检查（以及什么文件）在 sftp 上是否存在？我的问题又来了：我必须下载所有文件，然后终止。但是当目录（以每日目录结尾的完整路径）不存在时，就终止。
我想当文件不在目的地时，会抛出一些异常，如 FileNotFoundException。因此，只需捕获该异常，路由就会终止
在你的情况下，使用：GenericFileOperationFailedException
我不知道要下载的确切文件名，我只有一个掩码。正如我所提到的，根本没有例外，所以没有什么可捕捉的。只有一个无法捕捉到的警告。

【解决方案4】：

Otto Csatari：要在回滚方法中获取失败路由的 Route Id，请使用以下代码片段

    CamelContext context = endpoint.getCamelContext();
    List<Route> routes = context.getRoutes();
    SftpEndpoint sftpEndpoint = (SftpEndpoint)endpoint;
    Route failedRoute =  routes.stream().filter(route -> ((SftpEndpoint) route.getEndpoint())
            .getConfiguration().getDirectoryName().equals(sftpEndpoint.getConfiguration().getDirectoryName()))
            .findAny().orElse(null);

【讨论】：