如何让 Camel 内容丰富器处理来自 REST 调用的交换？

Question

我正在构建一个将订单路由到订单管理系统的示例实现。路由的入站是具有两个标识符的交换，一个用于客户 ID，另一个用于目录项 ID。然后，我将入站交换的主体转换为我的订单域对象。然后，我的目标是调用内容丰富器模式，一次用于聚合来自客户的数据，一次用于聚合来自目录项的数据。

我的路线是：

   @Override
   public void configure() {
      // Start by building an instance of RestConfigurationDefinition. Need to
      // specify the component we are going to use for enabling REST endpoints,
      // specifically CamelServlet in this case. Set the binding mode to JSON.
      restConfiguration().
            // Leverage the CamelServlet component for the REST DSL
            component("servlet").
            // Bind using JSON
            bindingMode(RestBindingMode.json).
            // I like pretty things...
            dataFormatProperty("prettyPrint", "true").
            // This is the context path to be used for Swagger API documentation
            apiContextPath("api-doc").
            // Properties for Swagger
            // Title of the API
      apiProperty("api.title", "Order Management API").
            // Version of the API
            apiProperty("api.version", "1.0.0").
            // CORS (resource sharing) enablement
            apiProperty("cors", "true").
            // Use localhost for calls
            apiProperty("host", "localhost:8083").
            // Set base path
            apiProperty("base.path", "nvisia-order-router-camel-service/api");

      // Definition of the post order endpoint
      rest("/orderRouter").
            // This is a POST method call for routing an order using the
            // order form
      post().
            // Description of what the method does
            description("Routes a new order to the order management service").
            // Define the type used for input
            type(OrderForm.class).
            // Define the type used for output, in this case the order
            outType(String.class).
            // Next, define where the message is routed to, first transformation
            to("bean:orderRouterService?method=transformOrderFormToOrder(${body})")
            .to("direct:enrichOrder");

      // Definition of the enrich order endpoint
      from("direct:enrichOrder").
            // Use the Content Enricher EIP to aggregate customer info in the
            // order.
      enrich(
            "http4://localhost:8081/nvisia-customer-camel-service/api/customer/${body.customerId}",
            new AggregationStrategy() {
               @Override
               public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                  Order originalBody = (Order) oldExchange.getIn().getBody();
                  Customer resourceResponse = (Customer) newExchange.getIn().getBody();
                  originalBody.setCustomer(resourceResponse);
                  if (oldExchange.getPattern().isOutCapable()) {
                     oldExchange.getOut().setBody(originalBody);
                  } else {
                     oldExchange.getIn().setBody(originalBody);
                  }
                  return oldExchange;
               }
            }).
            // Use the Content Enricher EIP to aggregate catalog info in the
            // order.
      enrich(
            "http4://localhost:8080/nvisia-catalog-camel-service/api/customer/${body.catalogItemId}",
            new AggregationStrategy() {
               @Override
               public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                  Order originalBody = (Order) oldExchange.getIn().getBody();
                  CatalogItem resourceResponse = (CatalogItem) newExchange.getIn()
                        .getBody();
                  originalBody.setCatalogItem(resourceResponse);
                  if (oldExchange.getPattern().isOutCapable()) {
                     oldExchange.getOut().setBody(originalBody);
                  } else {
                     oldExchange.getIn().setBody(originalBody);
                  }
                  return oldExchange;
               }
            }).to("direct:sendOrder");

      // Definition of the send order endpoint
      from("direct:sendOrder").
            // Need to define the content type on the header
            setHeader(org.apache.camel.Exchange.CONTENT_TYPE,
                  constant("application/json"))
            .
            // Be safe and define this as a post
            setHeader(Exchange.HTTP_METHOD,
                  constant(org.apache.camel.component.http4.HttpMethods.POST))
            .
            // Finally, send the order to be managed and get back the order ID
            to("http4://localhost:8082/nvisia-order-management-camel-service/api/order");
   }

我得到的例外是:

org.apache.camel.InvalidPayloadException: No body available of type: java.io.InputStream but has value: OrderForm [customerId=1, catalogItemId=1] of type: com.nvisia.examples.camel.orderrouter.OrderForm on: Message[]. Caused by: No type converter available to convert from type: com.nvisia.examples.camel.orderrouter.OrderForm to the required type: java.io.InputStream with value OrderForm [customerId=1, catalogItemId=1]. Exchange[ID-nvisia-mhoffman-50981-1463522552963-0-8]. Caused by: [org.apache.camel.NoTypeConversionAvailableException - No type converter available to convert from type: com.nvisia.examples.camel.orderrouter.OrderForm to the required type: java.io.InputStream with value OrderForm [customerId=1, catalogItemId=1]]
    at org.apache.camel.impl.MessageSupport.getMandatoryBody(MessageSupport.java:107)
    at org.apache.camel.component.http4.HttpProducer.createRequestEntity(HttpProducer.java:523)
    at org.apache.camel.component.http4.HttpProducer.createMethod(HttpProducer.java:422)
    at org.apache.camel.component.http4.HttpProducer.process(HttpProducer.java:110)
    at org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
    at org.apache.camel.processor.Enricher.process(Enricher.java:187)
    at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
    at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:468)
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:121)
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:83)
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
    at org.apache.camel.component.direct.DirectProducer.process(DirectProducer.java:62)
    at org.apache.camel.processor.SendProcessor.process(SendProcessor.java:145)
    at org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:77)
    at org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:468)
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:121)
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:83)
    at org.apache.camel.processor.CamelInternalProcessor.process(CamelInternalProcessor.java:190)
    at org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:109)
    at org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:87)
    at org.apache.camel.http.common.CamelServlet.service(CamelServlet.java:143)
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:729)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:291)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:52)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.springframework.boot.actuate.autoconfigure.EndpointWebMvcAutoConfiguration$ApplicationContextHeaderFilter.doFilterInternal(EndpointWebMvcAutoConfiguration.java:261)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.springframework.boot.actuate.trace.WebRequestTraceFilter.doFilterInternal(WebRequestTraceFilter.java:115)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:99)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.springframework.web.filter.HttpPutFormContentFilter.doFilterInternal(HttpPutFormContentFilter.java:87)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.springframework.web.filter.HiddenHttpMethodFilter.doFilterInternal(HiddenHttpMethodFilter.java:77)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:121)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.springframework.boot.actuate.autoconfigure.MetricsFilter.doFilterInternal(MetricsFilter.java:103)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.springframework.boot.context.web.ErrorPageFilter.doFilter(ErrorPageFilter.java:120)
    at org.springframework.boot.context.web.ErrorPageFilter.access$000(ErrorPageFilter.java:61)
    at org.springframework.boot.context.web.ErrorPageFilter$1.doFilterInternal(ErrorPageFilter.java:95)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.springframework.boot.context.web.ErrorPageFilter.doFilter(ErrorPageFilter.java:113)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:239)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:206)
    at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:217)
    at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:106)
    at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:502)
    at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:142)
    at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:79)
    at org.apache.catalina.valves.AbstractAccessLogValve.invoke(AbstractAccessLogValve.java:616)
    at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:88)
    at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:518)
    at org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:1091)
    at org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:673)
    at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1500)
    at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.run(NioEndpoint.java:1456)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
    at java.lang.Thread.run(Thread.java:745)
Caused by: org.apache.camel.NoTypeConversionAvailableException: No type converter available to convert from type: com.nvisia.examples.camel.orderrouter.OrderForm to the required type: java.io.InputStream with value OrderForm [customerId=1, catalogItemId=1]
    at org.apache.camel.impl.converter.BaseTypeConverterRegistry.mandatoryConvertTo(BaseTypeConverterRegistry.java:198)
    at org.apache.camel.impl.MessageSupport.getMandatoryBody(MessageSupport.java:105)
    ... 79 more

注意我使用的是最新的camel，2.17.1。我的问题是，如果我得到一个主体是JSON格式的交换，我如何调用两个内容丰富器来填充我的Order bean？由于我不需要向任何一个内容丰富器调用发送输入流，所以我考虑在每次交换之前有一个处理器，但是我认为在第二个聚合中维护来自第一个聚合的信息会很困难。如果这不是使用内容丰富器的推荐方式，请告诉我。

Answer 1

使用rich的选项类似于此集成模式

不要在扩充中直接使用 http4

<enrich strategyRef="myAgregator">
  <simple>http4://foo-url/</simple> <!-- InvalidPayloadException -->
</enrich>

因此，与直接uri间接使用：

<camel:route id="consume-service1">
  <from uri="direct:consume-service1" />
  <camel:removeHeaders pattern="CamelHttp*" />
  <to uri="http4://order-service/" />
  <unmarshal ref="orderJson" />
  <enrich strategyRef="myAgregator">
    <simple>direct:consume-service2</simple>
  </enrich>
  <marshal ref="myJson" />
</camel:route>

<camel:route id="consume-service2">
  <from uri="direct:consume-service2" />
  <camel:setBody><constant></constant></camel:setBody>
  <camel:removeHeaders pattern="CamelHttp*" />
  <to uri="http4://catalog-service" />
  <unmarshal ref="catalogJson" />
</camel:route>

Answer 2

我能够解决这一问题，不再使用内容丰富器，而是使用两个直接endpoint的多播。我还需要显式地解组和编组，这也导致了一些额外的问题。这是我使用的最终路线，以防对其他人有用。

  // Definition of the post order endpoint
  rest("/orderRouter")
        // This is a POST method call for routing an order using the order
        // form
        .post()
        // Description of what the method does
        .description("Routes a new order to the order management service")
        // Define the type used for input
        .type(OrderForm.class)
        // Define the type used for output, in this case the order
        .outType(OrderNumber.class)
        // Now, process the order
        .to("direct:processOrder");

  // This is the route that processes the order sent. First, we need to take
  // the data from the order form passed, put it in the header and empty out
  // the body of our incoming exchange.
  from("direct:processOrder")
        // Set header for customer ID
        .setHeader("customerId", simple("${body.customerId}"))
        // Set header for catalog item ID
        .setHeader("catalogItemId", simple("${body.catalogItemId}"))
        // Empty the body
        .setBody(constant(""))
        // Now, aggregate the data to an order type
        .end()
        // Use multicasting to call the customer and catalog item services
        // in parallel. Then, use a strategy that groups the exchanges
        // returned from the service calls into a single list for
        // processing.
        .multicast(new GroupedExchangeAggregationStrategy())
        // Use parallel processing
        .parallelProcessing()
        // Send to two direct components to get the data
        .to("direct:getCustomerData", "direct:getCatalogItemData")
        // End the multicast call
        .end()
        // Now process the exchange
        .process(new Processor() {
           @Override
           public void process(Exchange exchange) throws Exception {
              List<Exchange> exchanges = exchange.getIn().getBody(List.class);
              Order order = new Order();
              for (Exchange exchangeToProcess : exchanges) {
                 if (exchangeToProcess.getIn().getBody() instanceof Customer) {
                    order.setCustomer(
                          exchangeToProcess.getIn().getBody(Customer.class));
                 } else if (exchangeToProcess.getIn()
                       .getBody() instanceof CatalogItem) {
                    order.setCatalogItem(
                          exchangeToProcess.getIn().getBody(CatalogItem.class));
                 } else {
                    // Ignore it for now.
                 }
              }
              order.setOrderDate(new Date(System.currentTimeMillis()));
              exchange.getIn().setBody(order);
           }
        })
        // End this processor definition
        .end()
        // Need to marshal the body to JSON
        .marshal()
        // Need to use JSON for marshalling
        .json(JsonLibrary.Jackson)
        // Then convert it to a string
        .convertBodyTo(String.class)
        // We can now send the order to order management. Need to define the
        // content type on the header
        .setHeader(org.apache.camel.Exchange.CONTENT_TYPE,
              constant("application/json"))
        // Be safe and define this as a post
        .setHeader(Exchange.HTTP_METHOD,
              constant(org.apache.camel.component.http4.HttpMethods.POST))
        // Set the HTTP uri to be used.
        .setHeader("CamelHttpUri", simple(
              "http://localhost:8082/nvisia-order-management-camel-service/api/order"))
        // Finally, send the order to be managed and get back the order ID
        .to("http4://localhost:8082/nvisia-order-management-camel-service/api/order")
        // Next, convert the input stream returned to a string
        .convertBodyTo(String.class)
        // Finally, unmarshal the string to an object
        .unmarshal().json(JsonLibrary.Jackson, OrderNumber.class);

  // Retrieves the customer data from the REST service for customer.
  from("direct:getCustomerData")
        // Set the http method as GET
        .setHeader("CamelHttpMethod", constant("GET"))
        // Set the HTTP uri to be used.
        .setHeader("CamelHttpUri", simple(
              "http://localhost:8081/nvisia-customer-camel-service/api/customer/${header.customerId}"))
        // Define the endpoint; though, url will be ignored in favor of
        // header
        .to("http4://localhost:8081/nvisia-customer-camel-service/api/customer/${header.customerId}")
        // Next, convert the input stream returned to a string
        .convertBodyTo(String.class)
        // Finally, unmarshal the string to an object
        .unmarshal().json(JsonLibrary.Jackson, Customer.class);

  // Retrieves the catalog item data from the REST service for catalog
  // items.
  from("direct:getCatalogItemData")
        // Set the http method as GET
        .setHeader("CamelHttpMethod", constant("GET"))
        // Set the HTTP uri to be used.
        .setHeader("CamelHttpUri", simple(
              "http://localhost:8080/nvisia-catalog-camel-service/api/catalogItem/${header.catalogItemId}"))
        // Define the endpoint; though, url will be ignored in favor of
        // header
        .to("http4://localhost:8080/nvisia-catalog-camel-service/api/catalogItem/${header.catalogItemId}")
        // Next, convert the input stream returned to a string
        .convertBodyTo(String.class)
        // Finally, unmarshal the string to an object
        .unmarshal().json(JsonLibrary.Jackson, CatalogItem.class);