Camel - 当消费目录不存在时停止路由 [英] Camel - Stop route when the consuming directory not exists

问题描述

我有一个 SFTP 路由(在 Spring XML 中)，它的 from 路径以每日更改的目录(即/yyyyMMdd)结尾，并且当 autoCreate=true 或路径开始时目录存在.但是如果目录不存在，我不允许创建！

I've an SFTP route (in Spring XML), and its from path ends in a daily changing directory (ie. /yyyyMMdd), and everything is working well when autoCreate=true or the directory exists when the route starts. But it is not permitted to me to create the directory if not exists!

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

When the dir exists, the route get files and terminates itself.

当目录不存在时，路由将永久轮询并发出警告(即org.apache.camel.component.file.GenericFileOperationFailedException:无法将目录更改为:20160917)并且永不停止.

When the dir not exists, the route is polling permanently with a warning (ie. org.apache.camel.component.file.GenericFileOperationFailedException: Cannot change directory to: 20160917) and never stops.

如何避免这种行为(例如，将警告转换为空消息或异常或...)?我已经用startingDirectoryMustExist、consumer.bridgeErrorHandler 和许多其他方法做过实验，但都没有成功.

How can I avoid this behaviour (eg. convert the warning to an empty message or exception or ...)? I've done experiments with startingDirectoryMustExist, consumer.bridgeErrorHandler and many others without any success.

简化路线(开始前，用实际日期填写elmu.sftp.importDir属性):

The simplified route (before start, fill the elmu.sftp.importDir property with the actual date):

    <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 和startingDirectoryMustExist=true，结果是一个无限循环(轮询)，并带有此警告:

With directoryMustExist=true and startingDirectoryMustExist=true the result is an endless loop (poll) with this warning:

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 的回答):

UPDATE (according to @ruffp's answer):

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

I've tried to set up a custom PollingConsumerPollStrategy, but I cannot stop the route from it. Only the third (commented row) stops the route, but I've several routes and I cannot know the name of the actual route. How can I get it?

    @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;
    }

推荐答案

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

Finally I've solved it with consumer.exceptionHandler. But it seems, that this option is not in list of the usable options (http://camel.apache.org/file2.html which I've been reading again and again) just mentioned once with an example at the bottom of the huge page. Unfortunately, it was so "hidden" that I missed it up to now.

我设置了一个新类并实现了 handleExceptions 方法:

I set up a new class and implemented the handleExceptions methods:

public class DirNotExistsExHandler implements ExceptionHandler

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

which get the exceptions and decide what to do. In the context, I did a bean definition:

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

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

And in the consumer, passed the bean to the handler:

consumer.exceptionHandler=#dirNotExistsExHandler

非常感谢您的帮助！

这篇关于Camel - 当消费目录不存在时停止路由的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！