首页 文章

如何让Camel内容加载程序从REST调用处理Exchange？

提问于
浏览 次
0

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

我的路线是：

@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

注意我使用的是最新的骆驼，2.17.1 . 我的问题是，如果我的主体是JSON格式的交换，我如何使两个内容更丰富的调用来填充我的Order bean？由于我不需要向任何内容丰富的呼叫发送输入流，我认为在每次交换之前都有一个处理器，但我认为第二次聚合将难以保持第一次聚合的信息 . 如果这不是推荐使用内容丰富的方式，请告诉我 .

apache-camel

1 回答

  • 0

    通过远离使用内容丰富而不是仅使用具有两个直接 endpoints 的多播，我能够解决这个问题 . 我还需要明确解组和编组，这也导致了一些其他问题 . 这是我用过的最终路线，以防它对其他人有用 .

    // 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);
    回复于

相关问题