spring - Spring Cloud Dataflow 聚合拆分列表

Question

我正在尝试使用 SCDF 来拆分和聚合我的列表。但我无法按预期聚合。

我的 SCDF 每分钟都被 rabbitmq 上的一个列表提供。假设我有一个由以下对象组成的列表：

[{"foo" : "11", "bar" : "21"},{"foo" : "11", "bar" : "22"},{"foo" : "12", "bar" : "23"}]

我将数组拆分为拆分器中的对象：app.splitter.expression=#jsonPath(payload,'$.[*]')并且它按预期工作。
在聚合器中，我想根据 property 的值聚合它们foo，这样聚合的结果应该是两个不同的数组，例如：

[{"foo" : "11", "bar" : "21"},{"foo" : "11", "bar" : "22"}]
[{"foo" : "12", "bar" : "23"}]

我设置correlation为：app.aggregator.correlation=#jsonPath(payload,'$.foo')但我不知道如何设置releaseSpEL 表达式。
你能告诉我应该如何设置release吗？

PS：作为一种解决方法，我尝试将最后一个对象添加到列表中以检测它是列表的末尾。所以，我将{{"foo" : "EoS"}对象添加到列表中。在release我尝试过：app.aggregator.release=!messages.?[new String(payload).contains("EoS")].empty但它没有用。

标签： springspring-cloud-streamspring-cloud-dataflow