在Spring Integration DSL中使用带有废弃通道的过滤器

Question

我不知道这个问题是关于spring-integration、spring-integration-dsl还是两者都涉及，所以我添加了这两个标签...

今天我花费了相当多的时间，首先使用一个过滤器创建了一个简单的流程

StandardIntegrationFlow flow = IntegrationFlows.from(...)
                    .filter(messagingFilter)
                    .transform(transformer)
                    .handle((m) -> {
                        (...)
                    })
                    .get();

messagingFilter是一个非常简单的MessageSelector实现。到目前为止，一切都很好，没有花费太多时间。但是，如果MessageSelector返回false，我想记录一条消息，这就是我卡住的地方。

经过相当长的时间，我最终得到了这个：

StandardIntegrationFlow flow = IntegrationFlows.from(...)
                    .filter(messagingFilters, fs -> fs.discardFlow( i -> i.channel(discardChannel()))
                    .transform(transformer)
                    .handle((m) -> {
                        (...)
                    })
                    .get();

(...)

public MessageChannel discardChannel() {
    MessageChannel channel = new MessageChannel(){
        @Override
        public boolean send(Message<?> message) {
            log.warn((String) message.getPayload().get("msg-failure"));
            return true;
        }
        @Override
        public boolean send(Message<?> message, long timeout) {
            return this.send(message);
        }
    };
    return channel;
}

这样写既难看又啰嗦，那么问题来了，我在这里做错了什么，应该如何更好、更干净、更优雅地解决呢？

谢谢。

- amsmota

Answer 1

你的问题在于没有意识到Filter是一种EI模式实现，它所能做的最多只是将被丢弃的消息发送到某个通道。因为这种方式已经不基于消息传递了，所以它不会记录任何日志。

对于你的使用情况，最简单的解决方法是：

.discardFlow(df -> df
        .handle(message -> log.warn((String) message.getPayload().get("msg-failure")))))

您的逻辑只是记录日志。有些人可能会做更复杂的逻辑。因此，您最终会习惯于在端点之间使用通道抽象。

我同意new MessageChannel() {}方法是错误的。日志确实应该在MessageHandler中完成。这是服务责任的层次。还要注意，有LoggingHandler，可以通过Java DSL实现：

 .filter(messagingFilters, fs -> fs.discardFlow( i -> i.log(message -> (String) message.getPayload().get("msg-failure"))))

- Artem Bilan