在集成流中,使用其默认策略的拆分从列表中发出一个项目。该项目的处理可能会失败。我想处理该错误并将带有映射信息的新消息从前一个消息(除了自定义错误标头)定向到正常消息传递通道。
在聚合器中,我想自定义聚合逻辑以生成其他类型的消息,其中包含失败进程的计数和未失败消息的结果。
在这里,我解释了如何发送带有标头的错误消息:
@Bean
public IntegrationFlow socialMediaErrorFlow() {
return IntegrationFlows.from("socialMediaErrorChannel")
.wireTap(sf -> sf.handle("errorService", "handleException"))
.<MessagingException>handle((p, h)
-> MessageBuilder.withPayload(Collections.<CommentEntity>emptyList())
.copyHeaders(p.getFailedMessage().getHeaders())
.setHeader("ERROR", true)
.build()
)
.channel("directChannel_1")
.get();
}
我希望聚合器生成这种类型的对象:
public class Result {
private Integer totalTask;
private Integer taskFailed;
private List<CommentEntity> comments;
}
我应该如何处理这个?
提前致谢。
感谢 Artem 的帮助,我做了这个实现:
.aggregate(a -> a.outputProcessor(new MessageGroupProcessor() {
@Override
public Object processMessageGroup(MessageGroup mg) {
Integer failedTaskCount = 0;
Integer totalTaskCount = mg.getMessages().size();
List<CommentEntity> comments = new ArrayList<>();
for(Message<?> message: mg.getMessages()){
if(message.getHeaders().containsKey("ERROR"))
failedTaskCount++;
else
comments.addAll((List<CommentEntity>)message.getPayload());
}
return new IterationResult(totalTaskCount, failedTaskCount, comments);
}
}))