java - Spring Cloud Dataflow 将缓冲的消息发送到输出 channel

Question

我有一个过滤器，可以缓冲被拒绝的消息。一旦我的谓词满足，我想将这些缓冲的消息发送到输出 channel :

public class MyFilter implements MessageSelector {

    private Buffer buff;

    ...

    @Override
    public boolean accept(final Message<?> message) {
        if (isAcceptable(message))) {
            for (final Message<?> msg : buff.getBuffered()) {
                // accept these as well
            }
            return true;
        } else {
            buff.put(message);
            return false;
        }
    }
}

如何接受缓冲的消息？有更好的方法吗？

Answer 1

我建议 aggregator与定制ReleaseStrategy将是一个很好的方法。当释放策略释放组时，如果输出处理器(以这种方式配置时的 @Aggregator 方法)返回 Message<?> 的集合它们将作为单独的消息发布。

使用 Java 配置时，您可能会发现使用 Java DSL .aggregate(...) 更简单功能胜于创造 @Bean s。

目前有一个 pull request对于聚合器应用程序启动器。