I am trying to aggregate multiple http requests payload and process them together. So, i have created a spring boot app that contain:

• a RestController: where i expect to get the payloads.
• a MessagingGetway: that send requests to request channel and wait a reply after the aggregation group is released and processed
• an IntegrationFlow: listen to the request channel of the Gateway and aggregate messages based on SequenceSizeReleaseStrategy and correlationId unique by RestController endpoint
• a message group handler: implement GenericHandler<List<MyDataObject>> that process the grouped messages together and return response object to the reply channel of the Messaging Gateway

The issue is :

Given a sequence size for example 3, when i send 3 requests consecutively, then the group is released and a reply will be send only once for the last request(the releaser of the group), And my first 2 requests reach a Messaging Gateway timeout and get a null response object.

To illustrate:

1. request 1: gateway timed out and reply object is null
2. request 2: gateway timed out and reply object is null
3. request 3: reply object returned successfully

The RestController code is below:

@RestController
@AllArgsConstructor
public class MyController {

    private final MyIntegrationGateway myIntegrationGateway;

    @PostMapping("/myEndpoint1")
    public ResponseEntity<String> myEndpoint1(@RequestBody MyDataObject payload) {

        var result = myIntegrationGateway.sendToFlow(payload, "GROUP_1", 3);

        return new ResponseEntity<>(result,HttpStatus.CREATED);
    }
}

The MessagingGetway code is below:

@MessagingGateway
public interface MyIntegrationGateway {

    @Gateway(requestChannel = "requestQueue", replyChannel = "responseQueue")
    String sendToFlow(MyDataObject payload,
                        @Header(IntegrationMessageHeaderAccessor.CORRELATION_ID) String correlationId,
                        @Header(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE) int groupSize) throws MessagingException;

}

The IntegrationFlow and channels definitions code is below:

    @Bean
    public IntegrationFlow myAggregatorFlow(MyGroupHandler myGroupHandler) {
        return IntegrationFlow
                .from(requestQueue())
                .aggregate(a -> a
                        .releaseStrategy(new SequenceSizeReleaseStrategy())
                        .expireGroupsUponCompletion(true)
                        .sendPartialResultOnExpiry(true))
                .handle(myGroupHandler)
                .channel("responseQueue")
                .get();
    }

    @Bean
    public MessageChannel requestQueue() {
        return MessageChannels.direct().getObject();
    }

    @Bean
    public MessageChannel responseQueue() {
        return MessageChannels.direct().getObject();
    }

The message group handler code is below :

@MessageEndpoint
public class MyGroupHandler implements GenericHandler<List<MyDataObject>> {


    @Override
    public Object handle(List<MyDataObject> payloadGroup, MessageHeaders headers) {

        if (payloadGroup.isEmpty()) {
            throw new IllegalArgumentException("Payload Group should not be empty");
        }
        .....
        
        return results;
    }
}

Note:
For responseQueue, I have tried also to use MessageChannels.publishSubscribe().getObject() as type of channel but no impact. The reply still sent only for the last request.

What I expect ?

I expect that all requests replied with the same result object sent after group processing without waiting Messaging Gateway timeout.