Camel 2.21.0 - 如何处理流媒体异常
Camel 2.21.0 - how to process on exception with streaming
我想在异常时记录错误并在下一个 record/split 上继续,但它不起作用。
我厌倦了 OnExcepiton()
、doTry()
DSL 但它不起作用并转到 ErrorHandler。
onException(IOException.class)
.handled(true).process(exchange -> log.error("error!!"));
from("file:" + rootDir + "/" + account + "/inbox/?move=.done")
.unmarshal(csvDataFormat)
.split(body()).shareUnitOfWork().parallelProcessing().streaming()
.process(fileService)
.end()
日志:
2018-07-18 14:01:59.883 DEBUG 45137 --- [/test1/request/] o.a.camel.processor.MulticastProcessor : Parallel processing failed due IOException reading next record: java.io.IOException: (line 4) invalid char between encapsulated token and delimiter
2018-07-18 14:01:59.885 ERROR 45137 --- [/test1/request/] o.a.camel.processor.DeadLetterChannel : Failed delivery for (MessageId: ID-**********-local-1531936914834-0-3 on ExchangeId: ID-*********-local-1531936914834-0-4). On delivery attempt: 0 caught: java.lang.IllegalStateException: IOException reading next record: java.io.IOException: (line 4) invalid char between encapsulated token and delimiter
@Bedla,感谢您的输入,我发现这适用于我的用例,
- 使用
onException()
仍在向
DeadLetterChannel
,所以不得不使用 doTry()
CasvFormat
使用
maps
- 我无法在 process
中修改 csvFormat
,所以不得不
使用 setBody
从文件中读取 header 并在 body 中附加 csv header
完整路线定义:
CsvDataFormat csvDataFormat = new CsvDataFormat().setUseMaps(true);
from("file:" + rootDir + "/test/")
.log(LoggingLevel.INFO,"Start processing ${file:name}")
.unmarshal().pgp(pgpFileName,pgpUserId,pgpPassword)
.process(exchange -> { /* just to get csv header */
InputStream inputStream = exchange.getIn().getBody(InputStream.class);
try(BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream))){
String header = bufferedReader.readLine();
exchange.getIn().setHeader("CSV_HEADER",header);
csvDataFormat.setHeader(header.split(",")); //<- this does not work, so had to add in body below!
System.out.println("csvHeader is : " + header);// + " ? " + Arrays.asList(csvDataFormat.getHeader()));
}
})
.split(body().tokenize("\n")).shareUnitOfWork()
.parallelProcessing().streaming()
.setBody(exchange -> exchange.getIn().getHeader("CSV_HEADER") + "\n" + exchange.getIn().getBody())
.doTry()
.unmarshal(csvDataFormat)
.process(requestFileService)
.doCatch(IOException.class)
//TODO: custom processing here...
.process(exchange -> log.error("caught in dotry: " + exchange.getIn().getBody())).stop()
.end()//end try/catch
.choice()
.when(simple("${property." + Exchange.SPLIT_COMPLETE + "} == true"))
.log(LoggingLevel.INFO, "Finished processing ${file:name}")
.end();
我想在异常时记录错误并在下一个 record/split 上继续,但它不起作用。
我厌倦了 OnExcepiton()
、doTry()
DSL 但它不起作用并转到 ErrorHandler。
onException(IOException.class)
.handled(true).process(exchange -> log.error("error!!"));
from("file:" + rootDir + "/" + account + "/inbox/?move=.done")
.unmarshal(csvDataFormat)
.split(body()).shareUnitOfWork().parallelProcessing().streaming()
.process(fileService)
.end()
日志:
2018-07-18 14:01:59.883 DEBUG 45137 --- [/test1/request/] o.a.camel.processor.MulticastProcessor : Parallel processing failed due IOException reading next record: java.io.IOException: (line 4) invalid char between encapsulated token and delimiter
2018-07-18 14:01:59.885 ERROR 45137 --- [/test1/request/] o.a.camel.processor.DeadLetterChannel : Failed delivery for (MessageId: ID-**********-local-1531936914834-0-3 on ExchangeId: ID-*********-local-1531936914834-0-4). On delivery attempt: 0 caught: java.lang.IllegalStateException: IOException reading next record: java.io.IOException: (line 4) invalid char between encapsulated token and delimiter
@Bedla,感谢您的输入,我发现这适用于我的用例,
- 使用
onException()
仍在向DeadLetterChannel
,所以不得不使用doTry()
CasvFormat
使用maps
- 我无法在process
中修改csvFormat
,所以不得不 使用setBody
从文件中读取 header 并在 body 中附加 csv header
完整路线定义:
CsvDataFormat csvDataFormat = new CsvDataFormat().setUseMaps(true);
from("file:" + rootDir + "/test/")
.log(LoggingLevel.INFO,"Start processing ${file:name}")
.unmarshal().pgp(pgpFileName,pgpUserId,pgpPassword)
.process(exchange -> { /* just to get csv header */
InputStream inputStream = exchange.getIn().getBody(InputStream.class);
try(BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream))){
String header = bufferedReader.readLine();
exchange.getIn().setHeader("CSV_HEADER",header);
csvDataFormat.setHeader(header.split(",")); //<- this does not work, so had to add in body below!
System.out.println("csvHeader is : " + header);// + " ? " + Arrays.asList(csvDataFormat.getHeader()));
}
})
.split(body().tokenize("\n")).shareUnitOfWork()
.parallelProcessing().streaming()
.setBody(exchange -> exchange.getIn().getHeader("CSV_HEADER") + "\n" + exchange.getIn().getBody())
.doTry()
.unmarshal(csvDataFormat)
.process(requestFileService)
.doCatch(IOException.class)
//TODO: custom processing here...
.process(exchange -> log.error("caught in dotry: " + exchange.getIn().getBody())).stop()
.end()//end try/catch
.choice()
.when(simple("${property." + Exchange.SPLIT_COMPLETE + "} == true"))
.log(LoggingLevel.INFO, "Finished processing ${file:name}")
.end();