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

INT-3657: Default MaxMessPerPoll=1 for @InboundCA

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

With XML/DSL, the MMPP defaults to 1 (via the SPCAFB) but
@InboundChannelAdapter defaulted to -1 (infinity).

Change the annotation to be consistent with the XML/DSL.

Add missing schema docs for the `basePollerType`.

* Fix `AbstractMethodAnnotationPostProcessor` for `maxMessagesPerPollValue` override
* Change `EnableIntegrationTests#foo()` `@InboundChannelAdapter` to use `maxMessagesPerPoll = "2"`
and modify appropriate test-case to be sure that `maxMessagesPerPoll` isn't overriden as it has been done in the original commit.
This commit is contained in:
Gary Russell
2015-02-27 11:26:55 -05:00
committed by Artem Bilan
parent 14a5e1f9c5
commit 29e65e68af
9 changed files with 74 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

@@ -195,6 +195,7 @@ subprojects { subproject ->
}
test {
maxHeapSize = "1024m"
jvmArgs "-javaagent:${configurations.jacoco.asPath}=destfile=${buildDir}/jacoco.exec,includes=org.springframework.integration.*"
}

5
spring-integration-core/src/main/java/org/springframework/integration/annotation/InboundChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -43,6 +43,7 @@ import java.lang.annotation.Target;
*
*
* @author Artem Bilan
* @author Gary Russell
* @since 4.0
*/
@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@@ -69,6 +70,8 @@ public @interface InboundChannelAdapter {
* ({@link org.springframework.integration.scheduling.PollerMetadata}).
* This attribute is an {@code array} just to allow an empty default (no poller).
* Only one {@link org.springframework.integration.annotation.Poller} element is allowed.
* NOTE: a {@link Poller} here has {@link Poller#maxMessagesPerPoll()} set to 1 by default.
*/
Poller[] poller() default {};
}

7
spring-integration-core/src/main/java/org/springframework/integration/annotation/Poller.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,9 +22,9 @@ import java.lang.annotation.Retention;
import java.lang.annotation.Target;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.scheduling.Trigger;
/**
* Provides the {@link PollerMetadata} options for the Messaging annotations for
@@ -39,6 +39,7 @@ import org.springframework.scheduling.Trigger;
* Non-reference attributes support Property Placeholder resolutions.
*
* @author Artem Bilan
* @author Gary Russell
* @since 4.0
*/
@Target({})
@@ -63,6 +64,7 @@ public @interface Poller {
/**
* @return The maximum number of messages to receive for each poll.
* Can be specified as 'property placeholder', e.g. {@code ${poller.maxMessagesPerPoll}}.
* Defaults to -1 (infinity) for polling consumers and 1 for polling inbound channel adapters.
*/
String maxMessagesPerPoll() default "";
@@ -83,4 +85,5 @@ public @interface Poller {
* Can be specified as 'property placeholder', e.g. {@code ${poller.cron}}.
*/
String cron() default "";
}

9
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -50,6 +50,7 @@ import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.AbstractPollingEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
@@ -247,7 +248,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
AbstractEndpoint endpoint;
if (inputChannel instanceof PollableChannel) {
PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) inputChannel, handler);
this.configurePollingEndpoint(pollingConsumer, annotations);
configurePollingEndpoint(pollingConsumer, annotations);
endpoint = pollingConsumer;
}
else {
@@ -286,6 +287,10 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
if (StringUtils.hasText(maxMessagesPerPollValue)) {
pollerMetadata.setMaxMessagesPerPoll(Long.parseLong(maxMessagesPerPollValue));
}
else if (pollingEndpoint instanceof SourcePollingChannelAdapter) {
// SPCAs default to 1 message per poll
pollerMetadata.setMaxMessagesPerPoll(1);
}
if (StringUtils.hasText(executorRef)) {
pollerMetadata.setTaskExecutor(this.beanFactory.getBean(executorRef, TaskExecutor.class));
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -66,7 +66,7 @@ public class InboundChannelAdapterAnnotationPostProcessor extends
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
adapter.setOutputChannel(channel);
adapter.setSource(messageSource);
this.configurePollingEndpoint(adapter, annotations);
configurePollingEndpoint(adapter, annotations);
return adapter;
}

44
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.2.xsd
View File

@@ -273,7 +273,7 @@
<xsd:documentation>
<![CDATA[
Allows you to specify the reference to the bean which implements java.util.Comparator&lt;Message&lt;?&gt;&gt;
interface and provides logic based on which Messages will be prioritized. Not allowed if `message-store` is set.
interface and provides logic based on which Messages will be prioritized. Not allowed if 'message-store' is set.
]]>
</xsd:documentation>
</xsd:annotation>
@@ -288,7 +288,7 @@
<xsd:documentation>
A reference to a bean that implements 'org.springframework.integration.store.PriorityCapableChannelMessageStore'.
A message store that supports priority in a manner defined by the store. When set, the underlying
channel will be a 'QueueChannel` that delegates to a `MessageGroupQueue' backed by the store.
channel will be a 'QueueChannel' that delegates to a 'MessageGroupQueue' backed by the store.
Not allowed if 'comparator' is set.
</xsd:documentation>
</xsd:annotation>
@@ -1772,8 +1772,23 @@
</xsd:annotation>
<xsd:sequence>
<xsd:choice>
<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1" />
<xsd:element name="advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1" />
<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation>
Run each poll task in a transaction; mutually exclusive
with 'advice-chain'; add a transaction advice if you need transactions
in addition to other advices.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="advice-chain" type="adviceChainType" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation>
A chain of nested AOP 'Advice' objects applied to the poller task.
Mutually exclusive with 'transactional'.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:choice>
</xsd:sequence>
<xsd:attribute name="fixed-delay" type="xsd:string">
@@ -1826,8 +1841,25 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="receive-timeout" type="xsd:string" />
<xsd:attribute name="max-messages-per-poll" type="xsd:string" />
<xsd:attribute name="receive-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Only applies to polling consumers - the time the poll thread will wait
after the trigger for a new message to arrive. Defaults to 1000 (1 second).
For polled inbound channel adapters, whether or not the polling thread blocks is
dependent on the message source implementation.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="max-messages-per-poll" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The maximum number of messages that will be produced for each poll. Defaults to
infinity (indicated by -1) for polling consumers, and 1 for polled
inbound channel adapters.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>

8
spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -374,6 +374,9 @@ public class EnableIntegrationTests {
Message<?> message = this.fooChannel.receive(1000);
assertNotNull(message);
assertEquals("foo", message.getPayload());
message = this.fooChannel.receive(1000);
assertNotNull(message);
assertEquals("foo", message.getPayload());
assertNull(this.fooChannel.receive(10));
message = this.messageChannel.receive(1000);
@@ -1151,7 +1154,8 @@ public class EnableIntegrationTests {
}
@Override
@InboundChannelAdapter(value = "fooChannel", poller = @Poller(trigger = "onlyOnceTrigger", maxMessagesPerPoll = "1"))
@InboundChannelAdapter(value = "fooChannel",
poller = @Poller(trigger = "onlyOnceTrigger", maxMessagesPerPoll = "2"))
public String foo() {
return "foo";
}

3
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcJavaConfigTests.java
View File

@@ -74,6 +74,7 @@ public class StoredProcJavaConfigTests {
Collection<?> primes = (Collection<?>) received.getPayload();
assertThat(primes, Matchers.<Object>contains(2, 3, 5, 7));
received = fooChannel.receive(100);
// verify maxMessagesPerPoll == 1
assertNull(received);
}
@@ -87,7 +88,7 @@ public class StoredProcJavaConfigTests {
}
@Bean
@InboundChannelAdapter(value = "fooChannel", poller = @Poller(fixedDelay="5000", maxMessagesPerPoll = "1"))
@InboundChannelAdapter(value = "fooChannel", poller = @Poller(fixedDelay="5000"))
public MessageSource<?> storedProc() {
StoredProcPollingChannelAdapter source = new StoredProcPollingChannelAdapter(storedProcExecutor());
source.setExpectSingleResult(true);

9
src/reference/docbook/whats-new.xml
View File

@@ -47,5 +47,14 @@
See <xref linkend="tcp-events"/> for more information.
</para>
</section>
<section>
<title>@InboundChannelAdapter</title>
<para>
Previously, the <classname>@Poller</classname> on an inbound channel adapter defaulted
the <code>maxMessagesPerPoll</code> attribute to <code>-1</code> (infinity). This was inconsistent
with the XML configuration of <code>&lt;inbound-channel-adapter/&gt;</code>s, which defaults
to 1. The annotation now defaults this attribute to 1.
</para>
</section>
</section>
</chapter>