apache-camel - Apache camel sftp org.apache.camel.NoTypeConversionAvailableException：没有可用于转换 java.io.InputStream 的类型转换器
问题描述
目前，由于类型转换，我一直无法从 ftp 服务器下载文件。
因此，下面提到的路线是从已经将我的自定义 POJO 作为主体集的管道调用的。自定义pojo如下：
public class DirectoryLocationListing {
private String domainName;
private String countryCode;
private String productName;
private String profileId;
private String directoryLocation;
sftp 的路径如下，我从特定服务器的目录中获取所有文件并将它们保存在本地文件路径中。
<route id="route5">
<from uri="direct:sftpGetCDRs"/>
<from uri="sftp://{{HOST}}//home/admin/ROSS_PULL_CDR/resellers/${body.directoryLocation}/?noop=true&streamDownload=true&username={{USER}}&password={{PASSWD}}"/>
<to uri="file://{{DB_DIR_LOC}}/temp/?fileName=${body.directoryLocation}/&autoCreate=true"/>
</route>
在运行应用程序时，我遇到了下面的异常，我觉得我不应该面对这个异常，因为 from 条目应该拾取文件并将其保存在 to 部分，但是我放置文件的 to 部分似乎就像在路由中获取的主体被引用而不是从 ftp 服务器获取的文件一样。
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor Elapsed (ms)
[_route1 ] [_route1 ] [quartz2://spring?cron=0+*+*+%3F+*+* ] [ 121]
[_route2 ] [_choice1 ] [when[simple{${body.domainName} != null and ${body.countryCode} !=
null] [ 6]
[_route2 ] [_to4 ] [bean:pullCDRProcessorWithFTP?method=createTemporaryDirectoryWithDirLoc ] [ 1]
[_route2 ] [_log6 ] [log ] [ 1]
[_route2 ] [_to4 ] [direct:sftpGetCDRs ] [ 3]
[route5 ] [to1 ] [file://C:/example/export/home/resellerid/wmcdr/cdr/temp/?fileName=${body.director] [ 2]
Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
Id ID-IGL70051-59483-1550660613598-0-7
ExchangePattern InOnly
Headers {breadcrumbId=ID-IGL70051-59483-1550660613598-0-1, calendar=null, CamelFileName=null, CamelRedelivered=false, CamelRedeliveryCounter=0, fireTime=Wed Feb 20 16:34:00 IST 2019, jobDetail=JobDetail 'Camel_camelContext-f6f63d02-4688-4b3f-b3ca-b206a61cbe9f.spring': jobClass: 'org.apache.camel.component.quartz2.CamelJob concurrentExectionDisallowed: false persistJobDataAfterExecution: false isDurable: false requestsRecovers: false, jobInstance=org.apache.camel.component.quartz2.CamelJob@65314fe3, jobRunTime=-1, mergedJobDataMap=org.quartz.JobDataMap@f7fbba19, nextFireTime=Wed Feb 20 16:35:00 IST 2019, previousFireTime=null, refireCount=0, result=null, scheduledFireTime=Wed Feb 20 16:34:00 IST 2019, scheduler=org.quartz.impl.StdScheduler@59c9a1bf, trigger=Trigger 'Camel_camelContext-f6f63d02-4688-4b3f-b3ca-b206a61cbe9f.spring': triggerClass: 'org.quartz.impl.triggers.CronTriggerImpl calendar: 'null' misfireInstruction: 1 nextFireTime: Wed Feb 20 16:35:00 IST 2019, triggerGroup=Camel_camelContext-f6f63d02-4688-4b3f-b3ca-b206a61cbe9f, triggerName=spring}
BodyType com.example.cdr.model.DirectoryLocationListing
Body com.example.cdr.model.DirectoryLocationListing@23a2ecf4
]
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.component.file.GenericFileOperationFailedException: Cannot store file: C:\EXAMPLE\export\home\resellerid\wmcdr\cdr\temp\IN\healthyindia\apj\SERVICEID
at org.apache.camel.component.file.FileOperations.storeFile(FileOperations.java:292)
at org.apache.camel.component.file.GenericFileProducer.writeFile(GenericFileProducer.java:277)
at org.apache.camel.component.file.GenericFileProducer.processExchange(GenericFileProducer.java:165)
at org.apache.camel.component.file.GenericFileProducer.process(GenericFileProducer.java:79)
at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:129)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.component.direct.DirectProducer.process(DirectProducer.java:51)
at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:129)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:118)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)
at org.apache.camel.processor.ChoiceProcessor.process(ChoiceProcessor.java:111)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:590)
at org.apache.camel.processor.MulticastProcessor.doProcessSequential(MulticastProcessor.java:518)
at org.apache.camel.processor.MulticastProcessor.process(MulticastProcessor.java:227)
at org.apache.camel.processor.Splitter.process(Splitter.java:104)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:118)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:118)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.component.direct.DirectProducer.process(DirectProducer.java:51)
at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:129)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:448)
at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:191)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:109)
at org.apache.camel.processor.MulticastProcessor.doProcessParallel(MulticastProcessor.java:736)
at org.apache.camel.processor.MulticastProcessor.access$200(MulticastProcessor.java:83)
at org.apache.camel.processor.MulticastProcessor$1.call(MulticastProcessor.java:304)
at org.apache.camel.processor.MulticastProcessor$1.call(MulticastProcessor.java:289)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.camel.InvalidPayloadException: No body available of type: java.io.InputStream but has value: com.EXAMPLE.cdr.model.DirectoryLocationListing@5a519bb2 of type: com.EXAMPLE.cdr.model.DirectoryLocationListing on: Message: com.EXAMPLE.cdr.model.DirectoryLocationListing@5a519bb2. Caused by: No type converter available to convert from type: com.EXAMPLE.cdr.model.DirectoryLocationListing to the required type: java.io.InputStream with value com.EXAMPLE.cdr.model.DirectoryLocationListing@5a519bb2. Exchange[Message: com.EXAMPLE.cdr.model.DirectoryLocationListing@5a519bb2]. Caused by: [org.apache.camel.NoTypeConversionAvailableException - No type converter available to convert from type: com.EXAMPLE.cdr.model.DirectoryLocationListing to the required type: java.io.InputStream with value com.EXAMPLE.cdr.model.DirectoryLocationListing@5a519bb2]
at org.apache.camel.impl.MessageSupport.getMandatoryBody(MessageSupport.java:101)
at org.apache.camel.component.file.FileOperations.storeFile(FileOperations.java:273)
... 52 more
Caused by: org.apache.camel.NoTypeConversionAvailableException: No type converter available to convert from type: com.EXAMPLE.cdr.model.DirectoryLocationListing to the required type: java.io.InputStream with value com.EXAMPLE.cdr.model.DirectoryLocationListing@5a519bb2
at org.apache.camel.impl.converter.BaseTypeConverterRegistry.mandatoryConvertTo(BaseTypeConverterRegistry.java:177)
at org.apache.camel.impl.MessageSupport.getMandatoryBody(MessageSupport.java:99)
请指导以防有人遇到同样的问题。所以可能需要类型转换，但我认为在这种情况下不需要它。
**更新 **
使用 pollenrich 后，我的路线如下：
<route id="route5">
<from uri="direct:sftpGetCDRs"/>
<process ref="sftpGetDirLocation"/>
<log message="property ${exchangeProperty.ftpGetDirectory}"/>
<pollEnrich uri="sftp://{{HOST}}//home/admin/ROSS_PULL_CDR/resellers/${exchangeProperty.ftpGetDirectory}/?noop=true&streamDownload=true&username={{USER}}&password={{PASSWD}}" timeout="0"/>
<to uri="file://{{DB_DIR_LOC}}/temp/?fileName=${body.directoryLocation}/&autoCreate=true"/>
</route>
我在这里指的处理器是
@Component("sftpGetDirLocation")
public class SFTPGetDirLocation implements Processor{
public void process(Exchange exchange) throws Exception {
exchange.setProperty("ftpGetDirectory", exchange.getIn().getBody(DirectoryLocationListing.class).getDirectoryLocation());
}
}
但是，我在此更改后的假设是 pollEnrich 将保留它从 URI 获得的任何（可能是文件）用于 sftp 并将其保留为正文。相反，我看到身体是空的
Message History
---------------------------------------------------------------------------------------------------------------------------------------
RouteId ProcessorId Processor Elapsed (ms)
[_route1 ] [_route1 ] [quartz2://spring?cron=0+*+*+%3F+*+* ] [ 657]
[_route2 ] [_choice1 ] [when[simple{${body.domainName} != null and ${body.countryCode} !=
null] [ 572]
[_route2 ] [_to4 ] [bean:pullCDRProcessorWithFTP?method=createTemporaryDirectoryWithDirLoc ] [ 2]
[_route2 ] [_log6 ] [log ] [ 7]
[_route2 ] [_to4 ] [direct:sftpGetCDRs ] [ 523]
[route5 ] [process1 ] [ref:sftpGetDirLocation ] [ 0]
[route5 ] [log1 ] [log ] [ 1]
[route5 ] [pollEnrich1 ] [pollEnrich[sftp://10.91.142.11//home/admin/ROSS_PULL_CDR/resellers/${exchangeP] [ 514]
[route5 ] [log2 ] [log ] [ 0]
[route5 ] [to1 ] [file://C:/EXAMPLE/export/home/resellerid/wmcdr/cdr/temp/?fileName=${body.director] [ 0]
Exchange
---------------------------------------------------------------------------------------------------------------------------------------
Exchange[
Id ID-IGL70051-57920-1550721464331-0-4
ExchangePattern InOnly
Headers {breadcrumbId=ID-IGL70051-57920-1550721464331-0-1, calendar=null, CamelFileName=null, CamelRedelivered=false, CamelRedeliveryCounter=0, CamelToEndpoint=sftp://10.91.142.11//home/admin/ROSS_PULL_CDR/resellers/$%7BexchangeProperty.ftpGetDirectory%7D/?noop=true&password=tcpip123&streamDownload=true&username=admin, fireTime=Thu Feb 21 09:28:00 IST 2019, jobDetail=JobDetail 'Camel_camelContext-f6f63d02-4688-4b3f-b3ca-b206a61cbe9f.spring': jobClass: 'org.apache.camel.component.quartz2.CamelJob concurrentExectionDisallowed: false persistJobDataAfterExecution: false isDurable: false requestsRecovers: false, jobInstance=org.apache.camel.component.quartz2.CamelJob@363935fd, jobRunTime=-1, mergedJobDataMap=org.quartz.JobDataMap@f7fbba19, nextFireTime=Thu Feb 21 09:29:00 IST 2019, previousFireTime=null, refireCount=0, result=null, scheduledFireTime=Thu Feb 21 09:28:00 IST 2019, scheduler=org.quartz.impl.StdScheduler@5bf7fd08, trigger=Trigger 'Camel_camelContext-f6f63d02-4688-4b3f-b3ca-b206a61cbe9f.spring': triggerClass: 'org.quartz.impl.triggers.CronTriggerImpl calendar: 'null' misfireInstruction: 1 nextFireTime: Thu Feb 21 09:29:00 IST 2019, triggerGroup=Camel_camelContext-f6f63d02-4688-4b3f-b3ca-b206a61cbe9f, triggerName=spring}
BodyType null
Body [Body is null]
]
因此例外是
Stacktrace
---------------------------------------------------------------------------------------------------------------------------------------
org.apache.camel.component.file.GenericFileOperationFailedException: Cannot write null body to file: C:\EXAMPLE\export\home\resellerid\wmcdr\cdr\temp
at org.apache.camel.component.file.FileOperations.storeFile(FileOperations.java:205)
解决方案
在我看来，代码中存在三个问题。
您正在发送一个
route5带有 body 类型的交换，DirectoryLocationListing并且您希望sftp组件使用带有第二个from定义的消息。穿了没用。当SFTP组件下载文件时，exchnage 中的 body 将被真实文件（类型org.apache.camel.component.file.GenericFile）替换，您将丢失DirectoryLocationListing之前 exchnage 中的对象。如果您的 pojo cam 设置在正文以外的其他位置，Content Enriher EIP是一个不错的选择。在当前情况下，在路由结束时，您期望来自
sftp端点的文件，但实际交付给您的是与类型主体的交换DirectoryLocationListing（您的 SFTP 组件不起作用）。Camel 无法做到这一点并将其持久化到磁盘上。这正是它所抱怨的。2.1。一旦您解决了这个问题（并且您的 SFTP 组件开始执行它的工作），您将遇到下一个情况，即传递到您的
to端点的消息现在是一个文件，您需要以某种方式保留directoryLocation在交换中直到结束路线。fileName定义中的参数to指向一个目录。它行不通。您必须使用动态端点定义
我不熟悉 Camel 的 XML DSL。因此，我将展示如何使用 Java DSL 来实现这一点。我将避免使用 SFTP 来进行酿造，而是使用file组件。我们将尝试一次性解决所有这些问题。
路线定义
from("direct:test")
.routeId("route5")
.log("=> body.directoryLocation is: ${body.directoryLocation}") //Just to see if we can read the directoryLocation property from the POJO
.process(exchange -> exchange.setProperty("WriteTargetDirectory",exchange.getIn().getBody(DirectoryListing.class).directoryLocation)) //Move directoryLocation property to an exchange property named WriteTargetDirectory
.pollEnrich()
.simple("file://source/${exchangeProperty.WriteTargetDirectory}/?noop=true")//Poll enrich magic here!
.log("=> I still have the target directory location as : ${exchangeProperty.WriteTargetDirectory}")
.log("=> I just received file [${in.header." + Exchange.FILE_NAME + "}] and will write it as [${exchangeProperty.WriteTargetDirectory}/${in.header." + Exchange.FILE_NAME + "}]" )
.toD("file://destination/${exchangeProperty.WriteTargetDirectory}/");
DirectoryListing 班级
public class DirectoryListing {
String directoryLocation;
public DirectoryListing(String directoryLocation) {
this.directoryLocation = directoryLocation;
}
public String getDirectoryLocation() {
return directoryLocation;
}
}
路由测试代码
template.sendBody("direct:test",new DirectoryListing("IN/healthyindia/"));
测试记录
[ main] DefaultCamelContext INFO Apache Camel 2.23.1 (CamelContext: camel-1) started in 0.203 seconds
[ main] route5 INFO => body.directoryLocation is: IN/healthyindia/
[ main] FileEndpoint INFO Endpoint is configured with noop=true so forcing endpoint to be idempotent as well
[ main] FileEndpoint INFO Using default memory based idempotent repository with cache max size: 1000
[ main] route5 INFO => I still have the target directory location as : IN/healthyindia/
[ main] route5 INFO => I just received file [test.txt] and will write it as [destination/IN/healthyindia//test.txt]
[ main] MainRouteBuilderTest INFO ********************************************************************************
[ main] MainRouteBuilderTest INFO Testing done: testRouteBuilder(JustTest.MainRouteBuilderTest)
[ main] MainRouteBuilderTest INFO Took: 0.125 seconds (125 millis)