2

Currently I am working on a Spring Integration application which has a following scenario:

  1. An int-http:outbound-gateway read from a REST-Services a list of paginated elements: about in
  2. Each page content splitted and stored in a folder to be processed later by a spring batch job.

I'm quite new with spring-integration and I don't know if it's possibile to create a kind of loop with `int-http:outbound-gateway' to read all pages until the last one.

We're talking about 66254 elements splitted in 2651 pages. What I'm looking for is a best practice to read and download all pages and collecting data without have any memory issue.

Any suggestion will be appreciated

Thanks

1

2 Answers 2

3

Yes, it is possible, although a bit tricky.

Assume your REST service require page as request param, so, you would like to make a request from the page #1 and loop (increment page param) until the service returns empty result.

So, you may have configuration for the REST service like:

<int-http:outbound-gateway url="http://service/elements?page={page}">
    <int-http:uri-variable name="page" expression="headers['page']"/>
</int-http:outbound-gateway>

Pay attention to that <int-http:uri-variable> definition. From the beginning you have to send the message to this <int-http:outbound-gateway> with the page header as a 1.

The reply from this gateway you should send to something like <recipient-list-router>, or <publish-subscribe-channel>, where one of the subscriber is still your splitter to to store items into the folder.

Another subscriber is a bit smart. It starts from <filter> to check if the payload (a result from the REST call) is empty, meaning that we have done and no more pages on the service to retrieve. Otherwise you use <header-enricher> to increment and replace the page header and send the result into that our first <int-http:outbound-gateway>.

0
1

I did something similar using the Java DSL for Spring Integration:

@Bean
IntegrationFlow servicePointIntegrationFlow() {
    return IntegrationFlows.from("servicePointExportChannel")
            .enrichHeaders(headerEnricherSpec -> headerEnricherSpec.header("pageNumber", new AtomicInteger()))
            .channel("singlePageRequestChannel")
            .get();
}
/**
 * Get a single page, and persist the contents of that page to the DB
 * Then, check if there are more pages.
 *   * If yes, request the next page by adding another message to the same channel (but with incremented pageNumber in the message header)
 *   * If no, publish a message to the next channel
 */
@Bean
IntegrationFlow paginatedDataFlow(ServicePointTransformer servicePointTransformer) {
    return IntegrationFlows.from("singlePageRequestChannel")
            .log(INFO, m -> "paginatedDataFlow called with pageNumber: " + m.getHeaders().get("pageNumber"))
            .handle(Http.outboundGateway(m -> myUrl + "&pageNumber={pageNumber}")
                    .uriVariable("pageNumber", m -> m.getHeaders().get("pageNumber", AtomicInteger.class).getAndIncrement())
                    .httpMethod(HttpMethod.GET)
                    .expectedResponseType(MyPage.class)
                    .mappedResponseHeaders(""))
            .routeToRecipients(route -> route
                    .recipientFlow(flow -> flow
                            .transform(MyPage::getServicePoints)
                            .transform(servicePointTransformer)
                            .handle(Jpa.updatingGateway(myEntityManagerFactory),
                                    spec -> spec.transactional(myTransactionManager))
                    .recipientFlow("payload.isNextPageAvailable", f -> f.channel("singlePageRequestChannel"))
                    .recipientFlow("!payload.isNextPageAvailable", f -> f.channel("someOtherChannel")))
            .get();
}

The payload/page is JSON and contains a top level field nextPageAvailable which can be access in its deserialized form as MyPage.isNextPageAvailable().

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.