routes - 当 header 值时 Camel 路由

Question

我想根据 REST url 模式将消息路由到不同的服务，我在 myRouteBuilder 类中的路由定义如下，

String module1RestUrls = "/client,/price,/client/add,/client/hello";
String module2RestUrls = "/order,/order/invoice,/suppliers";

from("jetty:http://192.168.1.1?matchOnUriPrefix=true&traceEnabled=true")
   .process(customProcessor)
     .choice()
       .when(simple("${module1RestUrls} contains ${header('CamelHttpUri')}"))
           .to("http4://" + module1Url + "?bridgeEndpoint=true")
       .when(simple("${module2RestUrls} contains ${header('CamelHttpUri')}"))
           .to("http4://" + module2Url + "?bridgeEndpoint=true")
       .otherwise()
           .to("http4://" + genericUrl + "?bridgeEndpoint=true");

module1RestUrls 和 module2RestUrls 当前是硬编码的，但稍后可以从属性文件加载。
我的问题是我不断收到

Caused by: org.apache.camel.language.simple.types.SimpleIllegalSyntaxException: Unknown   
function: module1RestUrls at location 0
${module1RestUrls} contains ${header('CamelHttpUri')}
*

如果有人可以提供帮助，那就太好了。我用谷歌搜索并尝试了不同的方法，但似乎没有任何效果。

完整的轨迹追踪--------------

org.apache.camel.CamelExecutionException: Exception occurred during execution on the exchange: Exchange[Message: [Body is null]]
at org.apache.camel.util.ObjectHelper.wrapCamelExecutionException(ObjectHelper.java:1358)
at org.apache.camel.builder.SimpleBuilder.createPredicate(SimpleBuilder.java:96)
at org.apache.camel.builder.SimpleBuilder.matches(SimpleBuilder.java:74)
at org.apache.camel.processor.ChoiceProcessor.process(ChoiceProcessor.java:66)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:72)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at org.apache.camel.processor.interceptor.BacklogTracerInterceptor.process(BacklogTracerInterceptor.java:84)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at org.apache.camel.processor.interceptor.TraceInterceptor.process(TraceInterceptor.java:91)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at org.apache.camel.processor.RedeliveryErrorHandler.processErrorHandler(RedeliveryErrorHandler.java:391)
at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:273)
at org.apache.camel.processor.RouteContextProcessor.processNext(RouteContextProcessor.java:46)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at org.apache.camel.processor.interceptor.DefaultChannel.process(DefaultChannel.java:335)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:117)
at org.apache.camel.processor.Pipeline.process(Pipeline.java:80)
at org.apache.camel.processor.RouteContextProcessor.processNext(RouteContextProcessor.java:46)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at org.apache.camel.processor.UnitOfWorkProcessor.processAsync(UnitOfWorkProcessor.java:150)
at org.apache.camel.processor.UnitOfWorkProcessor.process(UnitOfWorkProcessor.java:117)
at org.apache.camel.processor.RouteInflightRepositoryProcessor.processNext(RouteInflightRepositoryProcessor.java:48)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:72)
at org.apache.camel.component.jetty.CamelContinuationServlet.service(CamelContinuationServlet.java:139)
at javax.servlet.http.HttpServlet.service(HttpServlet.java:722)
at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:652)
at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1329)
at org.eclipse.jetty.servlets.MultiPartFilter.doFilter(MultiPartFilter.java:117)
at org.apache.camel.component.jetty.CamelFilterWrapper.doFilter(CamelFilterWrapper.java:44)
at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1300)
at org.eclipse.jetty.servlets.MultiPartFilter.doFilter(MultiPartFilter.java:117)
at org.apache.camel.component.jetty.CamelFilterWrapper.doFilter(CamelFilterWrapper.java:44)
at org.eclipse.jetty.servlet.ServletHandler$CachedChain.doFilter(ServletHandler.java:1300)
at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:445)
at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1038)
at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:374)
at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:972)
at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:135)
at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:116)
at org.eclipse.jetty.server.Server.handle(Server.java:359)
at org.eclipse.jetty.server.AbstractHttpConnection.handleRequest(AbstractHttpConnection.java:483)
at org.eclipse.jetty.server.AbstractHttpConnection.headerComplete(AbstractHttpConnection.java:920)
at org.eclipse.jetty.server.AbstractHttpConnection$RequestHandler.headerComplete(AbstractHttpConnection.java:982)
at org.eclipse.jetty.http.HttpParser.parseNext(HttpParser.java:635)
at org.eclipse.jetty.http.HttpParser.parseAvailable(HttpParser.java:235)
at org.eclipse.jetty.server.AsyncHttpConnection.handle(AsyncHttpConnection.java:82)
at org.eclipse.jetty.io.nio.SelectChannelEndPoint.handle(SelectChannelEndPoint.java:628)
at org.eclipse.jetty.io.nio.SelectChannelEndPoint$1.run(SelectChannelEndPoint.java:52)
at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:608)
at org.eclipse.jetty.util.thread.QueuedThreadPool$3.run(QueuedThreadPool.java:543)
at java.lang.Thread.run(Thread.java:662)
Caused by: org.apache.camel.language.simple.types.SimpleIllegalSyntaxException: Unknown     function: module1RestUrls at location 0
 ${module1RestUrls} contains ${header('CamelHttpUri')}
 *
at org.apache.camel.language.simple.SimplePredicateParser.parsePredicate(SimplePredicateParser.java:69)
at org.apache.camel.language.simple.SimpleLanguage.createPredicate(SimpleLanguage.java:137)
at org.apache.camel.builder.SimpleBuilder.createPredicate(SimpleBuilder.java:94)
... 61 more
Caused by: org.apache.camel.language.simple.types.SimpleParserException: Unknown     function: module1RestUrls
at org.apache.camel.language.simple.ast.SimpleFunctionExpression.createSimpleExpression(SimpleFunctionExpression.java:187)
at org.apache.camel.language.simple.ast.SimpleFunctionExpression.createExpression(SimpleFunctionExpression.java:40)
at org.apache.camel.language.simple.ast.SimpleFunctionStart.doCreateLiteralExpression(SimpleFunctionStart.java:58)
at org.apache.camel.language.simple.ast.SimpleFunctionStart.createExpression(SimpleFunctionStart.java:48)
at org.apache.camel.language.simple.ast.BinaryExpression.createExpression(BinaryExpression.java:78)
at org.apache.camel.language.simple.SimplePredicateParser.createPredicates(SimplePredicateParser.java:437)
at org.apache.camel.language.simple.SimplePredicateParser.doParsePredicate(SimplePredicateParser.java:118)
at org.apache.camel.language.simple.SimplePredicateParser.parsePredicate(SimplePredicateParser.java:66)
... 63 more

Answer 1

感谢 U2one，它很好，我按照以下步骤操作，

我在类 MyRouteBuilder extends RouteBuilder 上添加了以下内容，

from("jetty:http://"+myUrl+"?matchOnUriPrefix=true&traceEnabled=true")      
   .process(customProcessor)
     .choice()
    .when(simple("${in.headers.firstUrn} contains ${in.headers.CamelHttpUri}"))
  .to("http4://" + first1Url + "?bridgeEndpoint=true&throwExceptionOnFailure=false")
    .when(simple("${in.headers.secondUrn} contains ${in.headers.CamelHttpUri}"))
      .to("http4://" + second1Url + "?bridgeEndpoint=true&throwExceptionOnFailure=false")
    .otherwise()
      .to("http4://" + thirdUrl + "?bridgeEndpoint=true&throwExceptionOnFailure=false");

我写的 customProcessor 如下，

public void process(Exchange exchange) throws Exception {       
    String[] first1Url = messageSource.getMessage("first1Url", null, null).split(",");
    String[] second1Url= messageSource.getMessage("second1Url", null, null).split(",");

    exchange.getIn().setHeader("firstUrn ", first1Url );
    exchange.getIn().setHeader("secondUrn ", second1Url);

    Map<String, Object> headers = exchange.getIn().getHeaders();     

    exchange.getOut().setHeaders(headers);        
    exchange.getOut().setBody(exchange.getIn().getBody(String.class), String.class);
}

这很好地路由了所有内容，我将其余 url 存储在消息属性文件中，如下所示，

  firstUrn =/client,/client/add,/order,/resource,/others
  secondUrn=/hello,/purchase,/nothing

现在它解决了，AWS beanstalk 和camel 对此进行了一些其他问题的调查。

--- 只是一个更新，
更好的解决方案是使用 Ralf 建议的收件人列表模式，

上扩展 RouteBuilder 我们有，

from("servlet:///?matchOnUriPrefix=true") 
    .process(customPreProcessor) 
        .recipientList(header("urlNew"))
                .process(customPostProcessor);

在 自定义预处理器 我们有以下内容，

public void process(Exchange exchange) throws Exception {

    String CamelHttpUri = (String) exchange.getIn().getHeader("CamelHttpUri");
    String CamelHttpQuery = (String) exchange.getIn().getHeader("CamelHttpQuery");
    String urlReceived = CamelHttpUri;
    StringBuilder urlToMatch = new StringBuilder();
    urlToMatch.append(urlReceived);

    if (!StringUtils.isEmpty(CamelHttpQuery)) {
        urlToMatch.append("?").append(CamelHttpQuery);
    }
    String endpointUrl = getEndPointUrl(serviceUrl, restUrl, urlToMatch.toString());
    }

    if (StringUtils.isEmpty(endpointUrl)) {
        endpointUrl = "http4://" + dummyUrl + "?bridgeEndpoint=true&throwExceptionOnFailure=false";
    }

    exchange.getIn().setHeader("CamelHttpUri", urlReceived);
    exchange.getIn().setHeader("CamelHttpQuery", CamelHttpQuery);
    exchange.getIn().setHeader("restUrl", endpointUrl);

    exchange.getOut().setHeaders(exchange.getIn().getHeaders());
    exchange.getOut().setBody(exchange.getIn().getBody(String.class), String.class);
}

private String getEndPointUrl(String serviceUrls, String restUrl, String urlToMatch) {
    List<String> serviceUrlList = Arrays.asList(serviceUrls.split("\\s*,\\s*"));
    for (String urlPattern: serviceUrlList ) {
        Pattern pattern = Pattern.compile(urlPattern);
        Matcher matcher = pattern.matcher(urlToMatch);
        if (matcher.matches()) {
            return "http4://" + restUrl + "?bridgeEndpoint=true&throwExceptionOnFailure=false";
        }
    }
    return "";
}

在 CusomPostProcessor 上，我们可以在将结果发送给客户端之前添加任何逻辑......