Если исключение возникает в потоке webflux с пружинной интеграцией, само исключение (с помощью stacktrace) отправляется вызывающей стороне в качестве полезной нагрузки через MessagePublishingErrorHandler, который использует канал ошибок из заголовка errorChannel, а не канал ошибок по умолчанию.
Как настроить обработчик ошибок, аналогичный WebExceptionHandler ?Я хочу создать Http-код состояния и, возможно, объект DefaultErrorAttributes в качестве ответа.
Простое определение потока, который начинается с errorChannel, не работает, сообщение об ошибке не заканчивается.Я попытался определить свой собственный fluxErrorChannel, но похоже, что он также не используется в качестве канала ошибок, ошибки не заканчиваются в моем errorFlow:
@Bean
public IntegrationFlow fooRestFlow() {
return IntegrationFlows.from(
WebFlux.inboundGateway("/foo")
.requestMapping(r -> r.methods(HttpMethod.POST))
.requestPayloadType(Map.class)
.errorChannel(fluxErrorChannel()))
.channel(bazFlow().getInputChannel())
.get();
}
@Bean
public MessageChannel fluxErrorChannel() {
return MessageChannels.flux().get();
}
@Bean
public IntegrationFlow errorFlow() {
return IntegrationFlows.from(fluxErrorChannel())
.transform(source -> source)
.enrichHeaders(h -> h.header(HttpHeaders.STATUS_CODE, HttpStatus.BAD_GATEWAY))
.get();
}
@Bean
public IntegrationFlow bazFlow() {
return f -> f.split(Map.class, map -> map.get("items"))
.channel(MessageChannels.flux())
.<String>handle((p, h) -> throw new RuntimeException())
.aggregate();
}
ОБНОВЛЕНИЕ
В MessagingGatewaySupport.doSendAndReceiveMessageReactive мой канал ошибок, определенный на WebFlux.inboundGateway, никогда не используется для установки канала ошибок, скорее канал ошибок - это всегда replyChannel, который создается здесь :
FutureReplyChannel replyChannel = new FutureReplyChannel();
Message<?> requestMessage = MutableMessageBuilder.fromMessage(message)
.setReplyChannel(replyChannel)
.setHeader(this.messagingTemplate.getSendTimeoutHeader(), null)
.setHeader(this.messagingTemplate.getReceiveTimeoutHeader(), null)
.setErrorChannel(replyChannel)
.build();
Канал ошибок в конечном итоге сбрасывается на originalErrorChannelHandler в Mono.fromFuture, но в моем случае этот канал ошибок «нулевой».Кроме того, onErrorResume лямбда никогда не вызывается:
return Mono.fromFuture(replyChannel.messageFuture)
.doOnSubscribe(s -> {
if (!error && this.countsEnabled) {
this.messageCount.incrementAndGet();
}
})
.<Message<?>>map(replyMessage ->
MessageBuilder.fromMessage(replyMessage)
.setHeader(MessageHeaders.REPLY_CHANNEL, originalReplyChannelHeader)
.setHeader(MessageHeaders.ERROR_CHANNEL, originalErrorChannelHeader)
.build())
.onErrorResume(t -> error ? Mono.error(t) : handleSendError(requestMessage, t));
Как это должно работать?