弹簧集成 DSL 自定义错误通道不起作用

Question

我使用Spring Integration的DSL实现。 我在下面有代码，但我无法使用我的自定义错误流。当身份验证方法引发运行时异常时，错误通道开始处理。我丰富标头以使用我的自定义错误流，但不使用。

// In Class - 1
@Bean
public MarshallingWebServiceInboundGateway marshallingWebServiceInboundGateway(BeanFactoryChannelResolver channelResolver, Jaxb2Marshaller marshaller) {
MarshallingWebServiceInboundGateway wsInboundGateway = new MarshallingWebServiceInboundGateway();
wsInboundGateway.setRequestChannel(channelResolver.resolveDestination("incomingRequest.input"));
wsInboundGateway.setReplyChannel(channelResolver.resolveDestination("outgoingResponse.input"));
wsInboundGateway.setErrorChannel(channelResolver.resolveDestination("errorChannel"));
wsInboundGateway.setMarshaller(marshaller);
wsInboundGateway.setUnmarshaller(marshaller);
return wsInboundGateway;
}

// In Class - 2
@Bean
public IntegrationFlow incomingRequest() {
return f -> f.<Object, Class<?>>route(t -> t.getClass(),
mapping -> mapping.subFlowMapping(payloadType1(),
sf -> sf.gateway("type1.input", ConsumerEndpointSpec::transactional))
.subFlowMapping(payloadType2(),
sf -> sf.gateway("type2.input", ConsumerEndpointSpec::transactional)),
conf -> conf.id("router:Incoming request router"));
}
// In Class - 3
@Bean
public IntegrationFlow type1() {
IntegrationFlow integrationFlow = f -> f
.enrichHeaders(h -> h.header(MessageHeaders.ERROR_CHANNEL, "error222", true))
.<Type1>handle((p, h) -> authentication.authenticate(p),
conf -> conf.id("service-activator:Authenticate"))
.transform(transformer::transformType1MsgToDataX,
conf -> conf.id("transform:Unmarshall type1 Message"))
.enrichHeaders(h -> h.headerExpression(TypeDataIntegrationMessageHeaderAccessor.MESSAGE_ID, "payload.id")
.headerExpression(TypeDataIntegrationMessageHeaderAccessor.MESSAGE_TYPE, "payload.messageType"))
.handle((GenericHandler<DataX>) repository::successResponseMessage,
conf -> conf.id("service-activator:return success"))
.channel("outgoingResponse.input")
;
return integrationFlow;
}
// In Class - 3
@Bean
public IntegrationFlow error222Flow() {
return IntegrationFlows.from("error222").handle("repository", "failureResponseMessage").get()
;
}

编辑：

在阿尔乔姆的回答之后，我的代码如下。但是，我仍然无法访问错误流中的标头参数。我收到错误 - "路由器'路由器：错误响应准备'没有解析通道">

// In Class - 1
@Bean
public MarshallingWebServiceInboundGateway marshallingWebServiceInboundGateway(BeanFactoryChannelResolver channelResolver, Jaxb2Marshaller marshaller) {
MarshallingWebServiceInboundGateway wsInboundGateway = new MarshallingWebServiceInboundGateway();
wsInboundGateway.setRequestChannel(channelResolver.resolveDestination("incomingRequest.input"));
wsInboundGateway.setReplyChannel(channelResolver.resolveDestination("outgoingResponse.input"));
wsInboundGateway.setErrorChannel(channelResolver.resolveDestination("errorResponse.input"));
wsInboundGateway.setMarshaller(marshaller);
wsInboundGateway.setUnmarshaller(marshaller);
return wsInboundGateway;
}

// In Class - 2
@Bean
public IntegrationFlow incomingRequest() {
return f -> f.<Object, Class<?>>route(t -> t.getClass(),
mapping -> mapping.subFlowMapping(payloadType1(),
sf -> sf.gateway("type1.input", ConsumerEndpointSpec::transactional))
.subFlowMapping(payloadType2(),
sf -> sf.gateway("type2.input", ConsumerEndpointSpec::transactional)),
conf -> conf.id("router:Incoming request router"));
}
// In Class - 2
@Bean 
public IntegrationFlow errorResponse(){ 
return f -> f.<MessageHandlingException, Object>route(t -> t.getFailedMessage().getHeaders().get("ABCDEF"), 
mapping -> mapping.subFlowMapping("ABCDEF", 
sf -> sf.gateway("customError.input", ConsumerEndpointSpec::transactional)), 
conf -> conf.id("router:error response prepare"));
}
// In Class - 3
@Bean
public IntegrationFlow type1() {
IntegrationFlow integrationFlow = f -> f
.enrichHeaders(h -> h.header("ABCDEF", "ABCDEF", true)) 
.<Type1>handle((p, h) -> authentication.authenticate(p),
conf -> conf.id("service-activator:Authenticate"))
.transform(transformer::transformType1MsgToDataX,
conf -> conf.id("transform:Unmarshall type1 Message"))
.enrichHeaders(h -> h.headerExpression(TypeDataIntegrationMessageHeaderAccessor.MESSAGE_ID, "payload.id")
.headerExpression(TypeDataIntegrationMessageHeaderAccessor.MESSAGE_TYPE, "payload.messageType"))
.handle((GenericHandler<DataX>) repository::successResponseMessage,
conf -> conf.id("service-activator:return success"))
.channel("outgoingResponse.input")
;
return integrationFlow;
}
// In Class - 3
@Bean
public IntegrationFlow customError(){
return f -> f.handle((GenericHandler<MessageHandlingException>)eventRepository::failureResponseMessage,
conf -> conf.id("service-activator:return failure"));
}

编辑 - 2：

我尝试Artem的测试代码，它在这种情况下有效。如果我按如下方式将 type1 流转换为子流映射(我这样做，因为我怀疑我的子流代码块(，则错误流无法打印 ABCDEF 参数值。 之后，我在子流映射中添加了另一个标头(XYZTWR(，但它也无法打印。

@Bean
public IntegrationFlow type1() {
return f -> f.<String, String>route(t -> t.toString(), mapping -> mapping.subFlowMapping("foo",
sf -> sf.gateway("fooFlow.input", ConsumerEndpointSpec::transactional).enrichHeaders(h -> h.header("XYZTRW", "XYZTRW", true))));
}
@Bean
public IntegrationFlow fooFlow() {
return f -> f.enrichHeaders(h -> h.header("ABCDEF", "ABCDEF", true))
.handle((p, h) -> {
throw new RuntimeException("intentional");
});
}

我的 S.OUT 是 ：

GenericMessage [payload=foo, headers={history=testGateway,type1.input, id=1fad7a65-4abe-c41d-0b22-36839a103269, timestamp=1503029553071}]

Answer 1

当我们将消息转移到不同的线程执行器或队列通道时，errorChannel标头开始工作。否则，标准throw和try...catch在同一调用堆栈中工作。

因此，在您的情况下，身份验证异常只是抛给调用方 - WS 入站网关。在这里，您已经配置了全局错误通道。

我做了这个测试：

@Configuration
@EnableIntegration
@IntegrationComponentScan
public static class ContextConfiguration {
@Bean
public IntegrationFlow errorResponse() {
return IntegrationFlows.from(errorChannel())
.<MessagingException, Message<?>>transform(MessagingException::getFailedMessage,
e -> e.poller(p -> p.fixedDelay(100)))
.get();
}
@Bean
public IntegrationFlow type1() {
return f -> f
.enrichHeaders(h -> h.header("ABCDEF", "ABCDEF", true))
.handle((p, h) -> { throw new RuntimeException("intentional"); });
}
@Bean
public PollableChannel errorChannel() {
return new QueueChannel();
}
}
@MessagingGateway(errorChannel = "errorChannel", defaultRequestChannel = "type1.input")
public interface TestGateway {
Message<?> sendTest(String payload);
}
...
@Autowired
private TestGateway testGateway;
@Test
public void testErrorChannel() {
Message<?> message = this.testGateway.sendTest("foo");
System.out.println(message);
}

我的 SOUT 告诉我：

GenericMessage [payload=foo, headers={ABCDEF=ABCDEF, id=ae5d2d44-46b7-912d-17d4-bf2ee656140a, timestamp=1502999446725}]

请为org.springframework.integration类别设置调试日志记录级别，并观察您的消息在哪个步骤中丢失了所需的标头。

更新

还行。我看到你的问题了。由于您使用sf -> sf.gateway("fooFlow.input", ConsumerEndpointSpec::transactional)，换句话说，您通过网关调用下游，您在那里所做的一切都在门后，如果出现错误，您可以像您发送到那里的内容一样内疚 - 网关的请求消息。默认情况下，下游failedMessage被吞噬。

要解决此问题，您应该考虑该.gateway()的附加errorChannel()选项，并在那里处理下游错误。 或者... 只是不要在路由器的子流中使用.gateway()，而是使用简单的channel映射。

该.transactional()也可以在任何.handle()上配置。