Spring Cloud Dataflow 将缓冲的消息发送到输出通道

Question

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

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

我建议使用自定义 ReleaseStrategy 的 aggregator 将是一个好方法。当释放策略释放组时，如果输出处理器（配置该方式时为@Aggregator 方法）返回Message<?> 的集合，它们将作为单独的消息释放。

使用 Java 配置时，您可能会发现使用 Java DSL .aggregate(...) 函数比创建 @Beans 更简单。

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