拆分/聚合从不释放组（Spring集成JavaDSL）

Question

我正在尝试做一个GroupBy基于共享ID的GeoJSON功能列表，以便通过使用拆分/聚合来聚合这些功能的单个字段，如下所示：

@Bean
IntegrationFlow myFlow() {
    return IntegrationFlows.from(MY_DIRECT_CHANNEL)
            .handle(Http.outboundGateway(myRestUrl)
                    .httpMethod(HttpMethod.GET)
                    .expectedResponseType(FeatureCollection.class)
                    .mappedResponseHeaders(""))
            .split(FeatureCollection.class, FeatureCollection::getFeatures)
            .aggregate(aggregator -> aggregator
                    .outputProcessor(agg -> {
                        final List<String> collected = agg
                                .getMessages()
                                .stream()
                                .map(m -> ((Number)((Feature) m.getPayload()).getProperties().get("my_field")).intValue() + "")
                                .collect(Collectors.toList());
                        return MyPojo.builder()
                                .myId(((Number) agg.getGroupId()).longValue())
                                .myListString(String.join(",", collected))
                                .build();
                    })
                    .correlationStrategy(m -> ((Feature) m.getPayload()).getProperties().get("shared_id"))
                    // .sendPartialResultOnExpiry(true)
                    // .groupTimeout(10000) // there's got to be a better way ...
                    // .expireGroupsUponTimeout(false)
            )
            .handle(Jpa.updatingGateway(myEntityManagerFactory).namedQuery(MyPojo.QUERY_UPDATE),
                    spec -> spec.transactional(myTransactionManager))
            .nullChannel();
}

除非我取消对这三行的注释，否则聚合器永远不会发布组，数据库也不会收到任何更新。如果我将groupTimeout设置为小于5秒，则会丢失部分结果。

我预计发布策略默认为SimpleSequenceSizeReleaseStrategy，我预计在处理完所有（拆分）功能后会自动释放所有组（REST服务消息中总共只有129个功能）。手动将其设置为发布策略没有帮助。

一旦处理完所有129条消息，释放组的正确方法是什么？

Answer 1

我让它使用变压器而不是拆分/聚合工作：

@Bean
IntegrationFlow myFlow(MyTransformer myTransformer) {
    return IntegrationFlows.from(MY_DIRECT_CHANNEL)
            .handle(Http.outboundGateway(myRestUrl)
                    .httpMethod(HttpMethod.GET)
                    .expectedResponseType(FeatureCollection.class)
                    .mappedResponseHeaders(""))
            .transform(myTransformer)
            .split()
            .handle(Jpa.updatingGateway(myEntityManagerFactory).namedQuery(MyEntity.QUERY_UPDATE),
                    spec -> spec.transactional(myTransactionManager))
            .nullChannel();
}

变压器的签名为：

@Component
public class MyTransformer implements GenericTransformer<FeatureCollection, List<MyEntity>> {

    @Override
    public List<MyEntity> transform(FeatureCollection featureCollection) {
        ...
    }
}