Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1463 Fix PriorityChannel#getRemainingCapacity

Browse Source 
JIRA: https://jira.spring.io/browse/INT-1463

An existing `QueueChannel#getRemainingCapacity()` is based on the `BlockingQueue#getRemainingCapacity()` which is always `Integer.MAX_VALUE`
and doesn't reflect reality for provided `capacity`

* Fix `PriorityChannel` to return `upperBound.availablePermits()` for `getRemainingCapacity()`
* Add test for `PriorityChannel` on the matter
* Also increase `reply-timeout` for `<int-redis:queue-outbound-gateway>` in `RedisQueueGatewayIntegrationTests-context.xml` from 1 sec to 10 secs
This commit is contained in:
Artem Bilan
2016-07-01 11:11:09 -04:00
parent 0187e49f70
commit cfa00c5707
4 changed files with 22 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java
View File

@@ -84,6 +84,11 @@ public class PriorityChannel extends QueueChannel {
this(0, null);
}
@Override
public int getRemainingCapacity() {
return this.upperBound.availablePermits();
}
@Override
protected boolean doSend(Message<?> message, long timeout) {
if (!this.upperBound.tryAcquire(timeout)) {

7
spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelCapacityPlaceholderTests-context.xml
View File

@@ -13,11 +13,12 @@
<context:property-placeholder
location="/org/springframework/integration/channel/config/ChannelCapacityPlaceholderTests.properties"/>
<gateway id="gateway"
service-interface="org.springframework.integration.channel.config.ChannelCapacityPlaceholderTests$TestService"/>
<channel id="channel">
<queue capacity="${capacity}"/>
</channel>
<channel id="priorityChannel">
<priority-queue capacity="${capacity}"/>
</channel>
</beans:beans>

18
spring-integration-core/src/test/java/org/springframework/integration/channel/config/ChannelCapacityPlaceholderTests.java
View File

@@ -24,6 +24,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.channel.PriorityChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
@@ -31,6 +32,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@@ -51,12 +53,16 @@ public class ChannelCapacityPlaceholderTests {
assertEquals(98, channel.getRemainingCapacity());
}
public interface TestService {
@org.springframework.integration.annotation.Gateway(requestChannel = "channel")
void test();
@Test
public void testCapacityOnPriorityChannel() {
PriorityChannel channel = context.getBean("priorityChannel", PriorityChannel.class);
assertNotNull(channel);
assertEquals(99, channel.getRemainingCapacity());
channel.send(MessageBuilder.withPayload("test1").build());
channel.send(MessageBuilder.withPayload("test2").build());
assertEquals(97, channel.getRemainingCapacity());
assertNotNull(channel.receive(0));
assertEquals(98, channel.getRemainingCapacity());
}
}

2
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests-context.xml
View File

@@ -26,7 +26,7 @@
<int-redis:queue-outbound-gateway id="outboundGateway"
request-channel="sendChannel"
queue="#{redisQueue.toString()}"
reply-timeout="1000"
reply-timeout="10000"
requires-reply="true"
reply-channel="outputChannel"/>