java - 使用 Apache Camel 的 FluentProducerTemplate 上的 ClosedChannelException

Question

当我测试 Apache Camel 路由时，我的应用程序遇到 ClosedChannelException 。

rest().post("/{{camel.rest.version}}/ufx")
    .type(MyMsg.class)
    .consumes(MediaType.APPLICATION_XML_VALUE)
    .produces(MediaType.APPLICATION_XML_VALUE)
    .responseMessage().code(HttpStatus.SC_OK).responseModel(MyMsg.class).endResponseMessage()
    .responseMessage().code(HttpStatus.SC_BAD_REQUEST).responseModel(MyMsg.class).endResponseMessage()
    .route().routeId("rst_myrest")      
    .to("direct:route_myroute")
    .endRest();

onException(ValidationException.class).
    handled(true).
    log(LoggingLevel.ERROR, LoggingConst.ERROR +" Validation exception: ${body}").
    onExceptionOccurred(validationExceptionProcessor).
    marshal().jaxb(MyMsg.class.getPackage().getName()).
    log(LoggingLevel.INFO, "Error response ${body}").
    end();  

from("direct:route_myroute").routeId("rt_ufx")      
    .unmarshal().jaxb(MyMsg.class.getPackage().getName())
    .process("myvalidator")
    //some other lines
    .marshal().jaxb(MyMsg.class.getPackage().getName())
    .end(); 

在我的测试中，传递了无效数据，因此调用了 ValidationExceptionProcessor。

public class ValidationExceptionProcessor implements Processor {
    @Override
    public void process(Exchange exchange) throws Exception {
        ValidationException request = exchange.getException(ValidationException.class);
        MyMsg msg = new MyMsg();
        //preparing some pretty object
        exchange.getOut().setBody(msg);
        exchange.getOut().setHeader("CamelHttpResponseCode", HttpStatus.BAD_REQUEST);
    }
}

当我直接调用我的应用程序(例如，从 Postman)时，它工作正常。正确的 MyMsg 对象被记录并返回给用户。Junit-test FluentProducerTemplate 用于检查结果:

@RunWith(CamelSpringBootRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)
@DirtiesContext
@Slf4j
@UseAdviceWith
public class myTest {

    @Produce(uri = "undertow:myaddress")
    FluentProducerTemplate holdProducer;

    @Test
    public void invalidDataException() throws IOException, InterruptedException {
        Exchange send = holdProducer.withHeader(Exchange.HTTP_METHOD, HttpMethod.POST)
                .withHeader(Exchange.CONTENT_TYPE, MediaType.APPLICATION_XML_VALUE)
                .withBody(Files.readAllBytes(Paths.get("src/test/resources/Request.xml")))
                .send();
        Exception exception = send.getException(); //here I get java.nio.channels.ClosedChannelException
    }
}

日志:

{"timestamp":"2018-10-30T15:04:12.779+00:00","level":"WARN","logger_name":"org.apache.camel.util.IOHelper","stack_trace":"java.nio.channels.ClosedChannelException: null
    at io.undertow.server.protocol.http.HttpResponseConduit.processWrite(HttpResponseConduit.java:122)
    at io.undertow.server.protocol.http.HttpResponseConduit.write(HttpResponseConduit.java:596)
    at io.undertow.conduits.ChunkedStreamSinkConduit.flush(ChunkedStreamSinkConduit.java:267)
    at org.xnio.conduits.ConduitStreamSinkChannel.flush(ConduitStreamSinkChannel.java:162)
    at io.undertow.channels.DetachableStreamSinkChannel.flush(DetachableStreamSinkChannel.java:119)
    at org.xnio.channels.Channels.flushBlocking(Channels.java:63)
    at io.undertow.servlet.spec.ServletOutputStreamImpl.close(ServletOutputStreamImpl.java:612)
    at org.apache.camel.util.IOHelper.close(IOHelper.java:342)
    at org.apache.camel.util.IOHelper.close(IOHelper.java:406)
    at org.apache.camel.util.IOHelper.close(IOHelper.java:416)
    at org.apache.camel.http.common.DefaultHttpBinding.copyStream(DefaultHttpBinding.java:434)
    at org.apache.camel.http.common.DefaultHttpBinding.doWriteDirectResponse(DefaultHttpBinding.java:496)
    at org.apache.camel.http.common.DefaultHttpBinding.doWriteResponse(DefaultHttpBinding.java:395)
    at org.apache.camel.http.common.DefaultHttpBinding.writeResponse(DefaultHttpBinding.java:322)
    at org.apache.camel.http.common.CamelServlet.doService(CamelServlet.java:209)
    at org.apache.camel.http.common.CamelServlet.service(CamelServlet.java:73)
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:790)
    at io.undertow.servlet.handlers.ServletHandler.handleRequest(ServletHandler.java:85)
    at io.undertow.servlet.handlers.FilterHandler$FilterChainImpl.doFilter(FilterHandler.java:129)
    at org.springframework.security.web.FilterChainProxy.doFilterInternal(FilterChainProxy.java:208)
    at org.springframework.security.web.FilterChainProxy.doFilter(FilterChainProxy.java:177)
    at org.springframework.web.filter.DelegatingFilterProxy.invokeDelegate(DelegatingFilterProxy.java:346)
    at org.springframework.web.filter.DelegatingFilterProxy.doFilter(DelegatingFilterProxy.java:262)
    at io.undertow.servlet.core.ManagedFilter.doFilter(ManagedFilter.java:61)
    at io.undertow.servlet.handlers.FilterHandler$FilterChainImpl.doFilter(FilterHandler.java:131)
    at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:99)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at io.undertow.servlet.core.ManagedFilter.doFilter(ManagedFilter.java:61)
    at io.undertow.servlet.handlers.FilterHandler$FilterChainImpl.doFilter(FilterHandler.java:131)
    at org.springframework.web.filter.HttpPutFormContentFilter.doFilterInternal(HttpPutFormContentFilter.java:105)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at io.undertow.servlet.core.ManagedFilter.doFilter(ManagedFilter.java:61)
    at io.undertow.servlet.handlers.FilterHandler$FilterChainImpl.doFilter(FilterHandler.java:131)
    at org.springframework.web.filter.HiddenHttpMethodFilter.doFilte    rInternal(HiddenHttpMethodFilter.java:81)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at io.undertow.servlet.core.ManagedFilter.doFilter(ManagedFilter.java:61)
    at io.undertow.servlet.handlers.FilterHandler$FilterChainImpl.doFilter(FilterHandler.java:131)
    at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:197)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at io.undertow.servlet.core.ManagedFilter.doFilter(ManagedFilter.java:61)
    at io.undertow.servlet.handlers.FilterHandler$FilterChainImpl.doFilter(FilterHandler.java:131)
    at io.undertow.servlet.handlers.FilterHandler.handleRequest(FilterHandler.java:84)
    at io.undertow.servlet.handlers.security.ServletSecurityRoleHandler.handleRequest(ServletSecurityRoleHandler.java:62)
    at io.undertow.servlet.handlers.ServletDispatchingHandler.handleRequest(ServletDispatchingHandler.java:36)
    at io.undertow.servlet.handlers.security.SSLInformationAssociationHandler.handleRequest(SSLInformationAssociationHandler.java:131)
    at io.undertow.servlet.handlers.security.ServletAuthenticationCallHandler.handleRequest(ServletAuthenticationCallHandler.java:57)
    at io.undertow.server.handlers.PredicateHandler.handleRequest(PredicateHandler.java:43)
    at io.undertow.security.handlers.AbstractConfidentialityHandler.handleRequest(AbstractConfidentialityHandler.java:46)
    at io.undertow.servlet.handlers.security.ServletConfidentialityConstraintHandler.handleRequest(ServletConfidentialityConstraintHandler.java:64)
    at io.undertow.security.handlers.AuthenticationMechanismsHandler.handleRequest(AuthenticationMechanismsHandler.java:60)
    at io.undertow.servlet.handlers.security.CachedAuthenticatedSessionHandler.handleRequest(CachedAuthenticatedSessionHandler.java:77)
    at io.undertow.security.handlers.AbstractSecurityContextAssociationHandler.handleRequest(AbstractSecurityContextAssociationHandler.java:43)
    at io.undertow.server.handlers.PredicateHandler.handleRequest(PredicateHandler.java:43)
    at io.undertow.server.handlers.PredicateHandler.handleRequest(PredicateHandler.java:43)
    at io.undertow.servlet.handlers.ServletInitialHandler.handleFirstRequest(ServletInitialHandler.java:292)
    at io.undertow.servlet.handlers.ServletInitialHandler.access$100(ServletInitialHandler.java:81)
    at io.undertow.servlet.handlers.ServletInitialHandler$2.call(ServletInitialHandler.java:138)
    at io.undertow.servlet.handlers.ServletInitialHandler$2.call(ServletInitialHandler.java:135)
    at io.undertow.servlet.core.ServletRequestContextThreadSetupAction$1.call(ServletRequestContextThreadSetupAction.java:48)
    at io.undertow.servlet.core.ContextClassLoaderSetupAction$1.call(ContextClassLoaderSetupAction.java:43)
    at io.undertow.servlet.handlers.ServletInitialHandler.dispatchRequest(ServletInitialHandler.java:272)
    at io.undertow.servlet.handlers.ServletInitialHandler.access$000(ServletInitialHandler.java:81)
    at io.undertow.servlet.handlers.ServletInitialHandler$1.handleRequest(ServletInitialHandler.java:104)
    at io.undertow.server.Connectors.executeRootHandler(Connectors.java:211)
    at io.undertow.server.HttpServerExchange$1.run(HttpServerExchange.java:809)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
","message":"Cannot close. Reason: null","camel.exchangeId":"ID-VRN26-1540911805592-0-3","camel.contextId":"","camel.breadcrumbId":"ID-VRN26-1540911805592-0-1","camel.messageId":"ID-VRN26-1540911805592-0-4"}

从 Fluent 调用和直接运行应用程序调用有什么区别？如何解决？

P.S.将消息输出更改为 In() 没有帮助

Answer 1

我找到了解决方案，但不在我要找的地方。这是我的 Http 代码集:

exchange.getOut().setHeader("CamelHttpResponseCode", HttpStatus.BAD_REQUEST);

HttpStatus.BAD_REQUEST 是一个 Spring 枚举条目。结果是一个枚举值:

BAD_REQUEST(400, "Bad Request"),

当我的应用程序在 Spring 容器内运行时，Spring 执行了枚举值解码，并且我收到了包含 Http-400 代码的正确正文。

然而，使用 FluentProducerTemplate (属于 Camel 库)，没有进行解码，并且在某处发生流畅崩溃。 (我确实想知道到底在哪里以及为什么。以及为什么错误如此奇怪)。

切换到Camel header 解决了我的问题

 exchange.getOut().setHeader("CamelHttpResponseCode", HttpStatus.SC_BAD_REQUEST); 

因为 Camel httpHeaders 不是枚举，而是 public-static-final 字段

public static final int SC_BAD_REQUEST = 400;

如果你为 Spring 枚举调用 .value() 也可以正常工作

exchange.getOut().setHeader("CamelHttpResponseCode", HttpStatus.BAD_REQUEST.value());