Spring Aggregation Aggregator does not have expireGroupUponCompletion and persistenceStore as input

Question

I recently switched to Spring download and started converting all XML files to Java Configs. @Aggregate annotations do not have expireGroupUponCompletion and persistenceStore as input. Is there any other way we can point these things out?

Answer 1

Is there any other way to point these things out?

Using Spring Java DSL Integration :

.aggregate(a -> a.expireGroupsUponCompletion(true)
                 .sendPartialResultOnExpiry(true)
                 .messageStore(messageStore(), null)

null , - endpoint defaul. , Lambda:

.aggregate((AggregatorSpec a) -> a.expireGroupsUponCompletion(true)
                 .sendPartialResultOnExpiry(true)
                 .messageStore(messageStore())

Answer 2

Artem.I .

    @Bean
public IntegrationFlow pollingFlow() {
    return IntegrationFlows.from(splitSegment())
                           .aggregate( a -> a
                                     .expireGroupsUponCompletion(true)
                                     .messageStore(persistenceMessageStore())
                                     .sendPartialResultOnExpiry(true)
                                     .discardChannel(aggregateSegments())
                                     .correlationStrategy(m -> m.getPayload())
                                     .releaseStrategy(g ->  g.getMessages().size() > 5)
                                     .outputProcessor(g -> {
                                                               CountDB ag = new CountDB();
                                                               ag.setId((Long) g.getOne().getPayload());
                                                               ag.setActiveUsers((long)g.getMessages().size());
                                                               return ag;
                                                            })
                                    ,
                                     null)
                           .channel(aggregateSegments())
                           .get();