Http Inbound Gateway Problem

[Curro]

Hi,

I've a SI project with the following composition:

<int-http:inbound-gateway 
        request-channel="requestChannel"
	name="/req"
	supported-methods="GET"
	mapped-request-headers="*" 			
	reply-channel="responseChannel"/>

	<int:channel id="requestChannel"/>
        <int:channel id="responseChannel"/>
	
	<int:chain input-channel="requestChannel" output-channel="messageSubscriber">
		<int:transformer ref="requestTransformer" />
		<int:header-enricher >
			<int:header name="ad_correlation_id" expression="T(java.util.UUID).randomUUID().toString()" />
		</int:header-enricher>
	</int:chain>		
				
	<int:publish-subscribe-channel id="messageSubscriber" task-executor="executor" />
	<task:executor id="executor" pool-size="6" />
			
	<int:service-activator input-channel="messageSubscriber" 
				        output-channel="channelPreAggregation" 
				        ref="service1"/>

	<int:service-activator input-channel="messageSubscriber"
				        output-channel="channelPreAggregation" 
				        ref="service2"/>
	
	<int:channel id="channelPreAggregation"/>
			
	<int:chain input-channel="channelPreAggregation" output-channel="responseChannel">	
		<int:aggregator correlation-strategy-expression="headers.get('ad_correlation_id')" release-strategy-expression="size() == 2" />		
		<int:transformer ref="toStringlTransformer" />				
	</int:chain>

Everything seem to be working perfectly, no error and the expected message is arriving to my last toString transformer and is sent to the final response channel. The weird thing is that the gateway sometimes doesn't return the message, I get a 200 OK but nothing in the payload despite I can see in log file and debugging that I'm sending to the final response channel the right information.
This only happens sometimes (always in the first call when I start the server), the rest of the times everything works fine.

The problem seems to be in the task executor, because if I remove it, everything run ok, but I would like to call to the two services activator at the same time, not sequentially, so I need the task executor (or is there other possibility???).

Any idea?? Maybe a problem with threads??

Thanks in advance.

[oleg.zhurakousky]

What is the return value for service1 and service2?
It seems to me that any one of these services can produce a reply to the gateway so whoever produces it first wins.

There are probably better ways to configure what you want, but first I want to understand what's going on in service1 and 2

[Curro]

Both services returns the same kind of object, it's just a DTO. After aggregation I have a transformer which select just one of these object (the one returned by service1 or the one returned by service2)

Anyway I'm getting the same problem regardless of which of the object wins so I tend to think that the problem is not related to the services.

[oleg.zhurakousky]

Can you remove the reply-channel from the HTTP Inbound gateway (reply-channel="responseChannel"). You don't need it.
And also in your chain, remove output-channel="responseChannel". You don't need that as well.

Just as an FYI, the reply will go via default replyChannel which is auto-created by the gateway and injected into the message as header.

This will simplify the configuration and hopefully allow us to determine what the problem is.

[Curro]

Done! (and thanks for the useful info) but the bug is still there

[oleg.zhurakousky]

Ok, lets remove more stuff

1. Remove header enricher
2. Add apply-sequence="true" to the pub-sub channel (this will do pretty much what your header enricher did)
3. Remove correlation-strategy-expression="headers.get('ad_correlation_id')" from the aggregator (it will rely on the default correlation ID which will be injected in the pub-sub channel based on apply-sequence="true")

Let's try to simplify it as much as we can without loosing functionality first.

[Cleric]

Good evening!

Can you explain what is the reason to use some custom correlation?
Here is some snapshot of your flow:

<gateway id="testGateway"
			 service-interface="org.springframework.integration.handler.PublishSubscribeAggregationTests$TestGateway">
		<method name="gateway" request-channel="subscriber"/>
	</gateway>

	<publish-subscribe-channel id="subscriber" task-executor="executor" apply-sequence="true"/>

	<task:executor id="executor" pool-size="6"/>

	<service-activator input-channel="subscriber" output-channel="aggregation" expression="payload.toUpperCase()"/>

	<service-activator input-channel="subscriber" output-channel="aggregation" expression="payload.toLowerCase()"/>

	<aggregator input-channel="aggregation"/>

And the test class:

@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class PublishSubscribeAggregationTests {

	@Autowired
	private TestGateway gateway;

	@Test
	public void testPublishSubscribeAggregation() {
		String payload = "gGHdsdHrghdf";
		for (int i = 0; i < 100; i++) {
			List<String> result = gateway.gateway(payload);
			assertEquals(2, result.size());
			assertThat(result, Matchers.hasItems(payload.toUpperCase(), payload.toLowerCase()));
		}
	}

	private static interface TestGateway {

		List<String> gateway(String payload);

	}
}

So, as Oleg said it seems like a problem of your services. Maybe it depends on request parameters which you are using in those services?
Can you simplify them as I've shown above and test again?

Cheers,
Artem

[Cleric]

Oops! Oleg is faster than me
Nevertheless it works with that custom correlation as well...

[Curro]

Done! stufff removed!

Summarizing we've:

<int-http:inbound-gateway 
        request-channel="requestChannel"
	name="/req"
	supported-methods="GET"
	mapped-request-headers="*"/>

	<int:channel id="requestChannel"/>
       	
	<int:chain input-channel="requestChannel" output-channel="messageSubscriber">
		<int:transformer ref="requestTransformer" />
	</int:chain>		
				
	<int:publish-subscribe-channel id="messageSubscriber" task-executor="executor" apply-sequence="true"/>
	<task:executor id="executor" pool-size="6" />
			
	<int:service-activator input-channel="messageSubscriber" 
				        output-channel="channelPreAggregation" 
				        ref="service1"/>

	<int:service-activator input-channel="messageSubscriber"
				        output-channel="channelPreAggregation" 
				        ref="service2"/>
	
	<int:channel id="channelPreAggregation"/>
			
	<int:chain input-channel="channelPreAggregation" >	
		<int:aggregator release-strategy-expression="size() == 2" />		
		<int:transformer ref="toStringlTransformer" />				
	</int:chain>

Bug still there.

[Curro]

Good evening!

Can you explain what is the reason to use some custom correlation?

Artem

I was using a custom correlation because in my first approach to the current solution I was using a recipient-router-list instead of a pub-subscriber. With the router I was sending the same message to both services and I needed an ID to be able to identify the response of each service in the aggregator.

After I changed from router to pub-subcriber because I wanted to call to the services at the same time, not sequentially (perfomance issues)