Camel丰富的SQL语法问题

Question

我的任务是使用 Camel 版本 2.20.0 创建一个 Camel 路由，该路由从 CSV 文件中获取一行，在 SQL 语句 where 子句中使用该行中的一个值，并合并结果并再次输出。如果我在 SQL 语句中对标识符进行硬编码，它可以正常工作，如果我尝试使用动态 URI，我会收到错误。

路线是：

from("file:///tmp?fileName=test.csv")
.split()
.tokenize("\n")
.streaming()
.parallelProcessing(true)
.setHeader("userID", constant("1001"))
//.enrich("sql:select emplid,name from employees where emplid = '1001'",
.enrich("sql:select name from employees where emplid = :#userID",
     new AggregationStrategy() {
        public Exchange aggregate(Exchange oldExchange,
                                     Exchange newExchange)    {...

正如我所说，如果我取消注释带有硬编码 1001 的行，它会查询数据库并按预期工作。但是使用 ':#userID' 语法我得到一个 Oracle 错误：

java.sql.SQLSyntaxErrorException: ORA-00942: table or view does not exist


    Message History
    ---------------------------------------------------------------------------------------------------------------------------------------
    RouteId              ProcessorId          Processor                                                                        Elapsed (ms)
    [route3            ] [route3            ] [file:///tmp?fileName=test.csv                                                 ] [        43]
    [route3            ] [log5              ] [log                                                                           ] [         2]
    [route3            ] [setHeader2        ] [setHeader[userID]                                                             ] [         0]
    [route3            ] [enrich2           ] [enrich[constant{sql:select name from employees where emplid = :#userID] [        40]

该表显然在那里，因为它在值被硬编码时工作，因此它与传递动态值有关。我已经尝试了很多关于如何在单引号内传递该变量的变体，使用来自正文而不是标题的值等，但还没有找到有效的组合，尽管我已经看到了很多类似的看似有效的例子.

我已经打开了跟踪，似乎标题设置也正确：

o.a.camel.processor.interceptor.Tracer   :  >>> (route3) setHeader[userID, 1001] --> enrich[constant{sql:select name from employees where emplid = :#userID}] <<< Pattern:InOnly, Headers:{CamelFileAbsolute=true, CamelFileAbsolutePath=/tmp/test.csv, CamelFileLastModified=1513116018000, CamelFileLength=26, CamelFileName=test.csv, CamelFileNameConsumed=test.csv, CamelFileNameOnly=test.csv, CamelFileParent=/tmp, CamelFilePath=/tmp/test.csv, CamelFileRelativePath=test.csv, userID=1001}, BodyType:String, Body:1001,SomeValue,MoreValues

需要进行哪些更改才能完成这项工作？

我还应该注意我已经尝试过这种方法，使用各种语法选项来引用标头值，但没有任何运气：

.enrich().simple("sql:select * from employees where emplid = :#${in.header.userID}").aggregate ...

Answer 1

来自docs：

从 Camel 2.16 onwards enrich 和 pollEnrich 都支持使用表达式来计算 uri 的动态端点，它允许使用来自当前 Exchange 的数据。换句话说，上面所说的所有内容都不再适用，它只是有效。

由于你使用的是 2.20，我想你可以试试这个例子：

from("file:///tmp?fileName=test.csv")
.split()
.tokenize("\n")
.streaming()
.parallelProcessing(true)
.setHeader("userID", constant("1001"))
//.enrich("sql:select emplid,name from employees where emplid = '1001'",
.enrich("sql:select name from employees where emplid = ':#${in.header.userID}'",
    new AggregationStrategy() {
        public Exchange aggregate(Exchange oldExchange,
                                    Exchange newExchange)    {...

查看文档中的Expression 主题以获取更多示例。

总结起来，表达式可以是：

"sql:select name from employees where emplid = ':#${in.header.userID}'"

编辑：

抱歉，我错过了:# suffix。你可以看到unit test working here。

只需注意列类型。如果是整数，则不需要引号。

干杯！

Answer 2

来自 Camel 文档：

pollEnrich 或enrich 不从当前访问任何数据 交换，这意味着在轮询时它不能使用任何现有的 您可能已在 Exchange 上设置的标头。

推荐的实现你想要的方法是使用recipientList，所以我建议你阅读一下。

• Content Enricher
• Recipient List

编辑：

正如 Ricardo Zanini 在他的回答中正确指出的那样，从 2.16 开始，使用 Camel 版本实际上可以实现这一点。由于 OP 使用的是 2.20，我的回答无效。

不过，我会保留我的回答，但要指出的是，这仅在您使用 2.16 之前的旧版本时才有效。