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

INT-4381: MessageSources refactoring (#2517)

Browse Source 
* INT-4381: MessageSources refactoring

JIRA: https://jira.spring.io/browse/INT-4381

* Make all the out-of-the-box `MessageSource` implementations based
on the `AbstractMessageSource`
* Fix `JdbcPollingChannelAdapterIntegrationTests` for sporadic failure
because of `fixed-rate` for the poller

* Fix HeaderEnricherTests race condition

The `errorChannel()` expect an error in the `testErrorChannel` after
`1000` ms, but at the same time the `poller` in configured for the
`3000` ms.

* Increase all the timeouts for replies
* Decrease a `fixed-delay` on the `poller`
* Some other code style polishing for the `HeaderEnricherTests`
This commit is contained in:
Artem Bilan
2018-07-25 13:01:22 -04:00
committed by Gary Russell
parent d6c8baf50f
commit 0d0605be78
28 changed files with 207 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -68,8 +68,7 @@ public class MethodInvokingMessageSource extends AbstractMessageSource<Object>
}
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
protected void onInit() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;

9
spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceRetrievingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2018 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.
@@ -39,6 +39,8 @@ import org.springframework.util.ObjectUtils;
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1
*/
public class ResourceRetrievingMessageSource extends AbstractMessageSource<Resource[]>
@@ -79,8 +81,7 @@ public class ResourceRetrievingMessageSource extends AbstractMessageSource<Resou
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
protected void onInit() {
if (this.patternResolver == null) {
this.patternResolver = this.applicationContext;
}
@@ -100,7 +101,7 @@ public class ResourceRetrievingMessageSource extends AbstractMessageSource<Resou
resources = null;
}
else {
resources = filteredResources.toArray(new Resource[filteredResources.size()]);
resources = filteredResources.toArray(new Resource[0]);
}
}
return resources;

8
spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java
View File

@@ -88,11 +88,13 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
}
@Override
public void afterPropertiesSet() throws Exception {
public final void afterPropertiesSet() throws Exception {
getEvaluationContext();
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
onInit();
}
protected StandardEvaluationContext getEvaluationContext() {
@@ -169,4 +171,8 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
return expression.getValue(this.getEvaluationContext(), input, expectedType);
}
protected void onInit() {
}
}

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

@@ -30,7 +30,7 @@
</channel>
<header-enricher input-channel="errorChannelInput" output-channel="failInput">
<poller max-messages-per-poll="1" fixed-delay="3000"/>
<poller max-messages-per-poll="1" fixed-delay="10"/>
<error-channel ref="testErrorChannel"/>
</header-enricher>

50
spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
* @author Artem Bilan
*
* @since 2.0
*/
@ContextConfiguration
@@ -63,8 +64,8 @@ public class HeaderEnricherTests {
public void replyChannel() {
PollableChannel replyChannel = context.getBean("testReplyChannel", PollableChannel.class);
MessageChannel inputChannel = context.getBean("replyChannelInput", MessageChannel.class);
inputChannel.send(new GenericMessage<String>("test"));
Message<?> result = replyChannel.receive(0);
inputChannel.send(new GenericMessage<>("test"));
Message<?> result = replyChannel.receive(10000);
assertNotNull(result);
assertEquals("TEST", result.getPayload());
assertEquals(replyChannel, result.getHeaders().getReplyChannel());
@@ -74,8 +75,8 @@ public class HeaderEnricherTests {
public void replyChannelName() {
PollableChannel replyChannel = context.getBean("testReplyChannel", PollableChannel.class);
MessageChannel inputChannel = context.getBean("replyChannelNameInput", MessageChannel.class);
inputChannel.send(new GenericMessage<String>("test"));
Message<?> result = replyChannel.receive(0);
inputChannel.send(new GenericMessage<>("test"));
Message<?> result = replyChannel.receive(10000);
assertNotNull(result);
assertEquals("TEST", result.getPayload());
assertEquals("testReplyChannel", result.getHeaders().getReplyChannel());
@@ -85,8 +86,8 @@ public class HeaderEnricherTests {
public void replyChannelExpression() {
PollableChannel replyChannel = context.getBean("testReplyChannel", PollableChannel.class);
MessageChannel inputChannel = context.getBean("replyChannelExpressionInput", MessageChannel.class);
inputChannel.send(new GenericMessage<String>("test"));
Message<?> result = replyChannel.receive(0);
inputChannel.send(new GenericMessage<>("test"));
Message<?> result = replyChannel.receive(10000);
assertNotNull(result);
assertEquals("TEST", result.getPayload());
assertEquals(replyChannel, result.getHeaders().getReplyChannel());
@@ -96,8 +97,8 @@ public class HeaderEnricherTests {
public void errorChannel() {
PollableChannel errorChannel = context.getBean("testErrorChannel", PollableChannel.class);
MessageChannel inputChannel = context.getBean("errorChannelInput", MessageChannel.class);
inputChannel.send(new GenericMessage<String>("test"));
Message<?> errorMessage = errorChannel.receive(1000);
inputChannel.send(new GenericMessage<>("test"));
Message<?> errorMessage = errorChannel.receive(10000);
assertNotNull(errorMessage);
Object errorPayload = errorMessage.getPayload();
assertEquals(MessageTransformationException.class, errorPayload.getClass());
@@ -110,7 +111,7 @@ public class HeaderEnricherTests {
public void correlationIdValue() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("correlationIdValueInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals("ABC", new IntegrationMessageHeaderAccessor(result).getCorrelationId());
}
@@ -119,7 +120,7 @@ public class HeaderEnricherTests {
public void correlationIdValueWithType() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("correlationIdValueWithTypeInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
Object correlationId = new IntegrationMessageHeaderAccessor(result).getCorrelationId();
assertEquals(Long.class, correlationId.getClass());
@@ -130,7 +131,7 @@ public class HeaderEnricherTests {
public void correlationIdRef() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("correlationIdRefInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(123, new IntegrationMessageHeaderAccessor(result).getCorrelationId());
}
@@ -139,7 +140,7 @@ public class HeaderEnricherTests {
public void expirationDateValue() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("expirationDateValueInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(new Long(1111), new IntegrationMessageHeaderAccessor(result).getExpirationDate());
}
@@ -148,7 +149,7 @@ public class HeaderEnricherTests {
public void expirationDateRef() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("expirationDateRefInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(new Long(9999), new IntegrationMessageHeaderAccessor(result).getExpirationDate());
}
@@ -157,7 +158,7 @@ public class HeaderEnricherTests {
public void priority() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("priorityInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(new Integer(42), new IntegrationMessageHeaderAccessor(result).getPriority());
}
@@ -167,7 +168,7 @@ public class HeaderEnricherTests {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("priorityExpressionInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel,
new GenericMessage<Map<String, String>>(Collections.singletonMap("priority", "-10")));
new GenericMessage<>(Collections.singletonMap("priority", "-10")));
assertNotNull(result);
assertEquals(new Integer(-10), new IntegrationMessageHeaderAccessor(result).getPriority());
}
@@ -176,7 +177,7 @@ public class HeaderEnricherTests {
public void expressionUsingPayload() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("payloadExpressionInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<TestBean>(new TestBean("foo")));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>(new TestBean("foo")));
assertNotNull(result);
assertEquals("foobar", result.getHeaders().get("testHeader"));
}
@@ -195,7 +196,7 @@ public class HeaderEnricherTests {
public void expressionWithDateType() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("expressionWithDateTypeInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
Object headerValue = result.getHeaders().get("currentDate");
assertEquals(Date.class, headerValue.getClass());
@@ -207,7 +208,7 @@ public class HeaderEnricherTests {
public void expressionWithLongType() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("expressionWithLongTypeInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(Long.class, result.getHeaders().get("number").getClass());
assertEquals(12345L, result.getHeaders().get("number"));
@@ -217,7 +218,7 @@ public class HeaderEnricherTests {
public void refWithMethod() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("refWithMethod", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(String.class, result.getHeaders().get("testHeader").getClass());
assertEquals("testBeanForMethodInvoker", result.getHeaders().get("testHeader"));
@@ -227,7 +228,7 @@ public class HeaderEnricherTests {
public void ref() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("ref", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(TestBean.class, result.getHeaders().get("testHeader").getClass());
TestBean testBeanForRef = context.getBean("testBean1", TestBean.class);
@@ -238,7 +239,7 @@ public class HeaderEnricherTests {
public void innerBean() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("innerBean", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(TestBean.class, result.getHeaders().get("testHeader").getClass());
TestBean testBeanForInnerBean = new TestBean("testBeanForInnerBean");
@@ -249,7 +250,7 @@ public class HeaderEnricherTests {
public void innerBeanWithMethod() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("innerBeanWithMethod", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
assertEquals(String.class, result.getHeaders().get("testHeader").getClass());
assertEquals("testBeanForInnerBeanWithMethod", result.getHeaders().get("testHeader"));
@@ -265,7 +266,7 @@ public class HeaderEnricherTests {
public void testRoutingSlip() {
MessagingTemplate template = new MessagingTemplate();
MessageChannel channel = context.getBean("routingSlipInput", MessageChannel.class);
Message<?> result = template.sendAndReceive(channel, new GenericMessage<String>("test"));
Message<?> result = template.sendAndReceive(channel, new GenericMessage<>("test"));
assertNotNull(result);
Object routingSlip = new IntegrationMessageHeaderAccessor(result)
.getHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP);
@@ -311,6 +312,7 @@ public class HeaderEnricherTests {
public int hashCode() {
return name != null ? name.hashCode() : 0;
}
}
}

25
spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java
View File

@@ -28,11 +28,10 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.io.Resource;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -55,7 +54,7 @@ import com.rometools.rome.io.XmlReader;
*
* @since 2.0
*/
public class FeedEntryMessageSource extends IntegrationObjectSupport implements MessageSource<SyndEntry> {
public class FeedEntryMessageSource extends AbstractMessageSource<SyndEntry> {
private final URL feedUrl;
@@ -145,18 +144,7 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
}
@Override
public Message<SyndEntry> receive() {
Assert.isTrue(this.initialized,
"'FeedEntryReaderMessageSource' must be initialized before it can produce Messages.");
SyndEntry entry = doReceive();
if (entry == null) {
return null;
}
return this.getMessageBuilderFactory().withPayload(entry).build();
}
@Override
protected void onInit() throws Exception {
protected void onInit() {
if (this.metadataStore == null) {
// first try to look for a 'messageStore' in the context
BeanFactory beanFactory = this.getBeanFactory();
@@ -176,13 +164,16 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
this.initialized = true;
}
private SyndEntry doReceive() {
@Override
protected SyndEntry doReceive() {
Assert.isTrue(this.initialized,
"'FeedEntryReaderMessageSource' must be initialized before it can produce Messages.");
SyndEntry nextEntry = null;
synchronized (this.monitor) {
nextEntry = getNextEntry();
if (nextEntry == null) {
// read feed and try again
this.populateEntryList();
populateEntryList();
nextEntry = getNextEntry();
}
}

4
spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -75,7 +75,7 @@ public class FeedEntryMessageSourceTests {
public void testReceiveFeedWithEntriesSorted() throws Exception {
ClassPathResource resource = new ClassPathResource("org/springframework/integration/feed/sample.rss");
FeedEntryMessageSource source = new FeedEntryMessageSource(resource, "foo");
source.setComponentName("feedReader");
source.setBeanName("feedReader");
source.setBeanFactory(mock(BeanFactory.class));
source.afterPropertiesSet();
Message<SyndEntry> message1 = source.receive();

21
spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -46,15 +46,14 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
import org.springframework.integration.file.filters.DiscardAwareFileListFilter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.filters.ResettableFileListFilter;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**
@@ -85,7 +84,8 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Artem Bilan
*/
public class FileReadingMessageSource extends IntegrationObjectSupport implements MessageSource<File>, Lifecycle {
public class FileReadingMessageSource extends AbstractMessageSource<File>
implements Lifecycle {
private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5;
@@ -350,21 +350,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
}
@Override
public Message<File> receive() throws MessagingException {
AbstractIntegrationMessageBuilder<File> messageBuilder = doReceive();
Message<File> message = null;
if (messageBuilder != null) {
message = messageBuilder.build();
if (logger.isInfoEnabled()) {
logger.info("Created message: [" + message + "]");
}
}
return message;
}
protected AbstractIntegrationMessageBuilder<File> doReceive() {
// rescan only if needed or explicitly configured
if (this.scanEachPoll || this.toBeReceived.isEmpty()) {

29
spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 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,23 +66,7 @@ public class FileInboundChannelAdapterSpec
}
FileInboundChannelAdapterSpec(Comparator<File> receptionOrderComparator) {
this.target = new FileReadingMessageSource(receptionOrderComparator) {
@Override
protected void onInit() {
if (FileInboundChannelAdapterSpec.this.scanner == null ||
FileInboundChannelAdapterSpec.this.filtersSet) {
try {
setFilter(FileInboundChannelAdapterSpec.this.fileListFilterFactoryBean.getObject());
}
catch (Exception e) {
throw new BeanCreationException("The bean for the [" + this + "] can not be instantiated.", e);
}
}
super.onInit();
}
};
this.target = new FileReadingMessageSource(receptionOrderComparator);
}
/**
@@ -275,6 +259,15 @@ public class FileInboundChannelAdapterSpec
@Override
public Map<Object, String> getComponentsToRegister() {
if (this.scanner == null || this.filtersSet) {
try {
this.target.setFilter(this.fileListFilterFactoryBean.getObject());
}
catch (Exception e) {
throw new BeanCreationException("The bean for the [" + this + "] can not be instantiated.", e);
}
}
if (this.expressionFileListFilter != null) {
return Collections.singletonMap(this.expressionFileListFilter, null);
}

2
spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java
View File

@@ -136,7 +136,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource<F>
}
@Override
public final void afterPropertiesSet() {
public final void onInit() {
Assert.state(this.remoteDirectoryExpression != null, "'remoteDirectoryExpression' must not be null");
doInit();
}

3
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java
View File

@@ -173,8 +173,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
}
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
protected void onInit() {
Assert.notNull(this.localDirectory, "localDirectory must not be null");
try {
if (!this.localDirectory.exists()) {

6
spring-integration-file/src/test/java/org/springframework/integration/file/AutoCreateDirectoryTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2018 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.
@@ -59,7 +59,7 @@ public class AutoCreateDirectoryTests {
}
@Test
public void autoCreateForInboundEnabledByDefault() {
public void autoCreateForInboundEnabledByDefault() throws Exception {
FileReadingMessageSource source = new FileReadingMessageSource();
source.setDirectory(new File(INBOUND_PATH));
source.setBeanFactory(mock(BeanFactory.class));
@@ -68,7 +68,7 @@ public class AutoCreateDirectoryTests {
}
@Test(expected = IllegalArgumentException.class)
public void autoCreateForInboundDisabled() {
public void autoCreateForInboundDisabled() throws Exception {
FileReadingMessageSource source = new FileReadingMessageSource();
source.setDirectory(new File(INBOUND_PATH));
source.setAutoCreateDirectory(false);

2
spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java
View File

@@ -137,7 +137,7 @@ public class StreamingInboundTests {
}
@Test
public void testExceptionOnFetch() {
public void testExceptionOnFetch() throws Exception {
exception.expect(MessagingException.class);
StringSessionFactory sessionFactory = new StringSessionFactory();
Streamer streamer = new Streamer(new StringRemoteFileTemplate(sessionFactory), null);

5
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java
View File

@@ -23,7 +23,6 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.util.Comparator;
import java.util.concurrent.ConcurrentHashMap;
@@ -128,7 +127,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
}
@Test
public void testMaxFetch() throws IOException {
public void testMaxFetch() throws Exception {
FtpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.afterPropertiesSet();
@@ -143,7 +142,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
}
@Test
public void testMaxFetchNoFilter() throws IOException {
public void testMaxFetchNoFilter() throws Exception {
FtpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(null);
messageSource.afterPropertiesSet();

7
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -158,9 +158,8 @@ public class ExpressionEvaluatingSqlParameterSourceFactory extends AbstractExpre
}
@Override
public void afterPropertiesSet() throws Exception {
super.afterPropertiesSet();
this.getEvaluationContext().setVariable("staticParameters", this.staticParameters);
protected void onInit() {
getEvaluationContext().setVariable("staticParameters", this.staticParameters);
}
private final class ExpressionEvaluatingSqlParameterSource extends AbstractSqlParameterSource {

30
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java
View File

@@ -22,8 +22,7 @@ import java.util.function.Consumer;
import javax.sql.DataSource;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.jdbc.core.ColumnMapRowMapper;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
@@ -33,7 +32,6 @@ import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
@@ -47,7 +45,7 @@ import org.springframework.util.Assert;
*
* @since 2.0
*/
public class JdbcPollingChannelAdapter extends IntegrationObjectSupport implements MessageSource<Object> {
public class JdbcPollingChannelAdapter extends AbstractMessageSource<Object> {
private final NamedParameterJdbcOperations jdbcOperations;
@@ -152,8 +150,7 @@ public class JdbcPollingChannelAdapter extends IntegrationObjectSupport implemen
}
@Override
protected void onInit() throws Exception {
super.onInit();
protected void onInit() {
if (!this.sqlParameterSourceFactorySet && getBeanFactory() != null) {
((ExpressionEvaluatingSqlParameterSourceFactory) this.sqlParameterSourceFactory)
.setBeanFactory(getBeanFactory());
@@ -165,30 +162,13 @@ public class JdbcPollingChannelAdapter extends IntegrationObjectSupport implemen
return "jdbc:inbound-channel-adapter";
}
/**
* Execute the query. If a query result set contains one or more rows, the
* Message payload will contain either a List of Maps for each row or, if a
* RowMapper has been provided, the values mapped from those rows. If the
* query returns no rows, this method will return <code>null</code>.
* #return the {@link Message} or {@code null} as a result of query.
*/
@Override
public Message<Object> receive() {
Object payload = poll();
if (payload == null) {
return null;
}
return getMessageBuilderFactory()
.withPayload(payload)
.build();
}
/**
* Execute the select query and the update query if provided. Returns the
* rows returned by the select query. If a RowMapper has been provided, the
* mapped results are returned.
*/
private Object poll() {
@Override
protected Object doReceive() {
List<?> payload = doPoll(this.sqlQueryParameterSource);
if (payload.size() < 1) {
payload = null;

81
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -19,8 +19,7 @@ package org.springframework.integration.jdbc;
import java.sql.CallableStatement;
import java.util.Map;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
@@ -32,9 +31,11 @@ import org.springframework.util.Assert;
* processed rows.
*
* @author Gunnar Hillert
* @author Artem Bilan
*
* @since 2.1
*/
public class StoredProcPollingChannelAdapter extends IntegrationObjectSupport implements MessageSource<Object> {
public class StoredProcPollingChannelAdapter extends AbstractMessageSource<Object> {
private final StoredProcExecutor executor;
@@ -42,30 +43,31 @@ public class StoredProcPollingChannelAdapter extends IntegrationObjectSupport im
/**
* Constructor taking {@link StoredProcExecutor}.
*
* @param storedProcExecutor Must not be null.
*
*/
public StoredProcPollingChannelAdapter(StoredProcExecutor storedProcExecutor) {
Assert.notNull(storedProcExecutor, "storedProcExecutor must not be null.");
this.executor = storedProcExecutor;
}
/**
* Executes the query. If a query result set contains one or more rows, the
* Message payload will contain either a List of Maps for each row or, if a
* RowMapper has been provided, the values mapped from those rows. If the
* query returns no rows, this method will return <code>null</code>.
* This parameter indicates that only one result object shall be returned from
* the Stored Procedure/Function Call. If set to true, a resultMap that contains
* only 1 element, will have that 1 element extracted and returned as payload.
* If the resultMap contains more than 1 element and expectSingleResult is true,
* then a {@link MessagingException} is thrown.
* Otherwise the complete resultMap is returned as the {@link Message} payload.
* Important Note: Several databases such as H2 are not fully supported.
* The H2 database, for example, does not fully support the {@link CallableStatement}
* semantics and when executing function calls against H2, a result list is
* returned rather than a single value.
* Therefore, even if you set expectSingleResult = true, you may end up with
* a collection being returned.
* @param expectSingleResult true if a single result is expected.
*/
@Override
public Message<Object> receive() {
Object payload = poll();
if (payload == null) {
return null;
}
return this.getMessageBuilderFactory().withPayload(payload).build();
public void setExpectSingleResult(boolean expectSingleResult) {
this.expectSingleResult = expectSingleResult;
}
/**
@@ -73,39 +75,30 @@ public class StoredProcPollingChannelAdapter extends IntegrationObjectSupport im
* rows returned by the select query. If a RowMapper has been provided, the
* mapped results are returned.
*/
private Object poll() {
@Override
protected Object doReceive() {
Object payload;
final Object payload;
Map<String, ?> resultMap = doPoll();
Map<String, ?> resultMap = this.executor.executeStoredProcedure();
if (resultMap.isEmpty()) {
payload = null;
}
else {
if (this.expectSingleResult && resultMap.size() == 1) {
payload = resultMap.values().iterator().next();
}
else if (this.expectSingleResult && resultMap.size() > 1) {
throw new MessagingException(
"Stored Procedure/Function call returned more than "
+ "1 result object and expectSingleResult was 'true'. ");
+ "1 result object and expectSingleResult was 'true'.");
}
else {
payload = resultMap;
}
}
return payload;
}
protected Map<String, ?> doPoll() {
return this.executor.executeStoredProcedure();
}
@Override
@@ -113,28 +106,4 @@ public class StoredProcPollingChannelAdapter extends IntegrationObjectSupport im
return "stored-proc:inbound-channel-adapter";
}
/**
* This parameter indicates that only one result object shall be returned from
* the Stored Procedure/Function Call. If set to true, a resultMap that contains
* only 1 element, will have that 1 element extracted and returned as payload.
*
* If the resultMap contains more than 1 element and expectSingleResult is true,
* then a {@link MessagingException} is thrown.
*
* Otherwise the complete resultMap is returned as the {@link Message} payload.
*
* Important Note: Several databases such as H2 are not fully supported.
* The H2 database, for example, does not fully support the {@link CallableStatement}
* semantics and when executing function calls against H2, a result list is
* returned rather than a single value.
*
* Therefore, even if you set expectSingleResult = true, you may end up with
* a collection being returned.
*
* @param expectSingleResult true if a single result is expected.
*/
public void setExpectSingleResult(boolean expectSingleResult) {
this.expectSingleResult = expectSingleResult;
}
}

8
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java
View File

@@ -142,7 +142,7 @@ public class JdbcPollingChannelAdapterIntegrationTests {
}
@Test
public void testSimplePollForListWithRowMapperAndOneUpdate() {
public void testSimplePollForListWithRowMapperAndOneUpdate() throws Exception {
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(embeddedDatabase,
"select * from item where status=2");
adapter.setUpdateSql("update item set status = 10 where id in (:id)");
@@ -170,7 +170,7 @@ public class JdbcPollingChannelAdapterIntegrationTests {
}
@Test
public void testSimplePollForListWithRowMapperAndUpdatePerRow() {
public void testSimplePollForListWithRowMapperAndUpdatePerRow() throws Exception {
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(embeddedDatabase,
"select * from item where status=2");
adapter.setUpdateSql("update item set status = 10 where id = :id");
@@ -199,7 +199,7 @@ public class JdbcPollingChannelAdapterIntegrationTests {
}
@Test
public void testSimplePollForListWithRowMapperAndInsertPerRowAndMaxRows() {
public void testSimplePollForListWithRowMapperAndInsertPerRowAndMaxRows() throws Exception {
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(embeddedDatabase,
"select * from item where id not in (select id from copy)");
adapter.setUpdateSql("insert into copy values(:id,10)");
@@ -229,7 +229,7 @@ public class JdbcPollingChannelAdapterIntegrationTests {
}
@Test
public void testSimplePollForListWithRowMapperAndUpdatePerRowWithMaxRows() {
public void testSimplePollForListWithRowMapperAndUpdatePerRowWithMaxRows() throws Exception {
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(embeddedDatabase,
"select * from item where status=2");
adapter.setUpdateSql("update item set status = 10 where id = :id");

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml
View File

@@ -10,7 +10,7 @@
<si:queue />
</si:channel>
<si:poller default="true" fixed-rate="100"/>
<si:poller default="true" fixed-delay="100"/>
<jdbc:embedded-database type="HSQL" id="dataSource">
<jdbc:script location="org/springframework/integration/jdbc/config/inboundSchema.sql" />

50
spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -20,8 +20,7 @@ import java.util.Map;
import javax.jms.Destination;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.jms.util.JmsAdapterUtils;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.jms.core.JmsTemplate;
@@ -37,8 +36,9 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Artem Bilan
*/
public class JmsDestinationPollingSource extends IntegrationObjectSupport implements MessageSource<Object> {
public class JmsDestinationPollingSource extends AbstractMessageSource<Object> {
private final JmsTemplate jmsTemplate;
@@ -100,14 +100,28 @@ public class JmsDestinationPollingSource extends IntegrationObjectSupport implem
this.sessionAcknowledgeMode = sessionAcknowledgeMode;
}
@Override
protected void onInit() {
if (this.sessionAcknowledgeMode != null) {
Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(this.sessionAcknowledgeMode);
if (acknowledgeMode != null) {
if (JmsAdapterUtils.SESSION_TRANSACTED == acknowledgeMode) {
this.jmsTemplate.setSessionTransacted(true);
}
else {
this.jmsTemplate.setSessionAcknowledgeMode(acknowledgeMode);
}
}
}
}
/**
* Will receive a JMS {@link javax.jms.Message} converting and returning it as
* a Spring Integration {@link Message}. This method will also use the current
* {@link JmsHeaderMapper} instance to map JMS properties to the MessageHeaders.
*/
@Override
@SuppressWarnings("unchecked")
public Message<Object> receive() {
protected Object doReceive() {
javax.jms.Message jmsMessage = doReceiveJmsMessage();
if (jmsMessage == null) {
return null;
@@ -119,10 +133,11 @@ public class JmsDestinationPollingSource extends IntegrationObjectSupport implem
if (this.extractPayload) {
object = this.jmsTemplate.getMessageConverter().fromMessage(jmsMessage);
}
AbstractIntegrationMessageBuilder<Object> builder = (object instanceof Message) ?
getMessageBuilderFactory().fromMessage((Message<Object>) object) :
getMessageBuilderFactory().withPayload(object);
return builder.copyHeadersIfAbsent(mappedHeaders).build();
AbstractIntegrationMessageBuilder<?> builder =
(object instanceof Message)
? getMessageBuilderFactory().fromMessage((Message<?>) object)
: getMessageBuilderFactory().withPayload(object);
return builder.copyHeadersIfAbsent(mappedHeaders);
}
catch (Exception e) {
throw new MessagingException(e.getMessage(), e);
@@ -143,19 +158,4 @@ public class JmsDestinationPollingSource extends IntegrationObjectSupport implem
return jmsMessage;
}
@Override
protected void onInit() {
if (this.sessionAcknowledgeMode != null) {
Integer acknowledgeMode = JmsAdapterUtils.parseAcknowledgeMode(this.sessionAcknowledgeMode);
if (acknowledgeMode != null) {
if (JmsAdapterUtils.SESSION_TRANSACTED == acknowledgeMode) {
this.jmsTemplate.setSessionTransacted(true);
}
else {
this.jmsTemplate.setSessionAcknowledgeMode(acknowledgeMode);
}
}
}
}
}

6
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2018 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.
@@ -47,10 +47,6 @@ public class JmsInboundChannelAdapterParser extends AbstractPollingInboundChanne
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsDestinationPollingSource.class);
String componentName = this.resolveId(element, builder.getBeanDefinition(), parserContext);
if (StringUtils.hasText(componentName)) {
builder.addPropertyValue("componentName", componentName);
}
String jmsTemplate = element.getAttribute(JmsParserUtils.JMS_TEMPLATE_ATTRIBUTE);
String destination = element.getAttribute(JmsParserUtils.DESTINATION_ATTRIBUTE);
String destinationName = element.getAttribute(JmsParserUtils.DESTINATION_NAME_ATTRIBUTE);

29
spring-integration-jpa/src/main/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -16,8 +16,7 @@
package org.springframework.integration.jpa.inbound;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.jpa.core.JpaExecutor;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -44,11 +43,12 @@ import org.springframework.util.Assert;
*
* @author Amol Nayak
* @author Gunnar Hillert
* @author Artem Bilan
*
* @since 2.2
*
*/
public class JpaPollingChannelAdapter extends IntegrationObjectSupport implements MessageSource<Object> {
public class JpaPollingChannelAdapter extends AbstractMessageSource<Object> {
private final JpaExecutor jpaExecutor;
@@ -59,7 +59,6 @@ public class JpaPollingChannelAdapter extends IntegrationObjectSupport implement
* @param jpaExecutor Must not be null.
*/
public JpaPollingChannelAdapter(JpaExecutor jpaExecutor) {
super();
Assert.notNull(jpaExecutor, "jpaExecutor must not be null.");
this.jpaExecutor = jpaExecutor;
}
@@ -68,26 +67,18 @@ public class JpaPollingChannelAdapter extends IntegrationObjectSupport implement
* Check for mandatory attributes
*/
@Override
protected void onInit() throws Exception {
super.onInit();
this.jpaExecutor.setBeanFactory(this.getBeanFactory());
protected void onInit() {
this.jpaExecutor.setBeanFactory(getBeanFactory());
}
/**
* Uses {@link JpaExecutor#poll()} to executes the JPA operation.
*
* Use {@link JpaExecutor#poll()} to executes the JPA operation.
* If {@link JpaExecutor#poll()} returns null, this method will return
* <code>null</code>. Otherwise, a new {@link Message} is constructed and returned.
*/
public Message<Object> receive() {
final Object payload = this.jpaExecutor.poll();
if (payload == null) {
return null;
}
return this.getMessageBuilderFactory().withPayload(payload).build();
@Override
protected Object doReceive() {
return this.jpaExecutor.poll();
}
@Override

26
spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -24,12 +24,10 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
@@ -46,22 +44,21 @@ import org.springframework.util.Assert;
* @author Oleg Zhurakousky
* @author Artem Bilan
*/
public class MailReceivingMessageSource implements MessageSource<Object>,
BeanFactoryAware, BeanNameAware, NamedComponent {
public class MailReceivingMessageSource extends AbstractMessageSource<Object> {
private final Log logger = LogFactory.getLog(this.getClass());
private final MailReceiver mailReceiver;
private final Queue<Object> mailQueue = new ConcurrentLinkedQueue<Object>();
private final Queue<Object> mailQueue = new ConcurrentLinkedQueue<>();
private volatile BeanFactory beanFactory;
private BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private boolean messageBuilderFactorySet;
private volatile String beanName;
private String beanName;
public MailReceivingMessageSource(MailReceiver mailReceiver) {
@@ -103,9 +100,8 @@ public class MailReceivingMessageSource implements MessageSource<Object>,
this.beanName = name;
}
@SuppressWarnings("unchecked")
@Override
public Message<Object> receive() {
protected Object doReceive() {
try {
Object mailMessage = this.mailQueue.poll();
if (mailMessage == null) {
@@ -120,10 +116,10 @@ public class MailReceivingMessageSource implements MessageSource<Object>,
this.logger.debug("received mail message [" + mailMessage + "]");
}
if (mailMessage instanceof Message) {
return (Message<Object>) mailMessage;
return getMessageBuilderFactory().fromMessage((Message<?>) mailMessage);
}
else {
return getMessageBuilderFactory().withPayload(mailMessage).build();
return getMessageBuilderFactory().withPayload(mailMessage);
}
}
}

24
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2016 the original author or authors.
* Copyright 2007-2018 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.
@@ -29,10 +29,11 @@ import org.springframework.expression.TypeLocator;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.expression.spel.support.StandardTypeLocator;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.mongodb.support.MongoHeaders;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.transaction.IntegrationResourceHolder;
import org.springframework.messaging.Message;
import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -58,11 +59,11 @@ import com.mongodb.DBObject;
* @author Amol Nayak
* @author Oleg Zhurakousky
* @author Yaron Yamin
* @author Artem Bilan
*
* @since 2.2
*/
public class MongoDbMessageSource extends IntegrationObjectSupport
implements MessageSource<Object> {
public class MongoDbMessageSource extends AbstractMessageSource<Object> {
private final Expression queryExpression;
@@ -169,9 +170,9 @@ public class MongoDbMessageSource extends IntegrationObjectSupport
}
@Override
protected void onInit() throws Exception {
protected void onInit() {
this.evaluationContext =
ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
TypeLocator typeLocator = this.evaluationContext.getTypeLocator();
if (typeLocator instanceof StandardTypeLocator) {
//Register MongoDB query API package so FQCN can be avoided in query-expression.
@@ -192,9 +193,9 @@ public class MongoDbMessageSource extends IntegrationObjectSupport
* query will be provided in the {@link MongoHeaders#COLLECTION_NAME} header.
*/
@Override
public Message<Object> receive() {
protected Object doReceive() {
Assert.isTrue(this.initialized, "This class is not yet initialized. Invoke its afterPropertiesSet() method");
Message<Object> message = null;
AbstractIntegrationMessageBuilder<Object> messageBuilder = null;
Object value = this.queryExpression.getValue(this.evaluationContext);
Assert.notNull(value, "'queryExpression' must not evaluate to null");
Query query;
@@ -226,9 +227,8 @@ public class MongoDbMessageSource extends IntegrationObjectSupport
}
}
if (result != null) {
message = this.getMessageBuilderFactory().withPayload(result)
.setHeader(MongoHeaders.COLLECTION_NAME, collectionName)
.build();
messageBuilder = this.getMessageBuilderFactory().withPayload(result)
.setHeader(MongoHeaders.COLLECTION_NAME, collectionName);
}
Object holder = TransactionSynchronizationManager.getResource(this);
@@ -237,7 +237,7 @@ public class MongoDbMessageSource extends IntegrationObjectSupport
((IntegrationResourceHolder) holder).addAttribute("mongoTemplate", this.mongoTemplate);
}
return message;
return messageBuilder;
}
}

39
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -27,11 +27,9 @@ import org.springframework.data.redis.support.collections.RedisCollectionFactory
import org.springframework.data.redis.support.collections.RedisStore;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.transaction.IntegrationResourceHolder;
import org.springframework.messaging.Message;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.Assert;
@@ -43,12 +41,13 @@ import org.springframework.util.Assert;
*
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.2
*/
public class RedisStoreMessageSource extends IntegrationObjectSupport
implements MessageSource<RedisStore> {
public class RedisStoreMessageSource extends AbstractMessageSource<RedisStore> {
private final ThreadLocal<RedisStore> resourceHolder = new ThreadLocal<RedisStore>();
private final ThreadLocal<RedisStore> resourceHolder = new ThreadLocal<>();
private volatile StandardEvaluationContext evaluationContext;
@@ -63,13 +62,10 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
* which should resolve to a 'key' name of the collection to be used.
* It assumes that {@link RedisTemplate} is fully initialized and ready to be used.
* The 'keyExpression' will be evaluated on every call to the {@link #receive()} method.
*
* @param redisTemplate The Redis template.
* @param keyExpression The key expression.
*/
public RedisStoreMessageSource(RedisTemplate<String, ?> redisTemplate,
Expression keyExpression) {
public RedisStoreMessageSource(RedisTemplate<String, ?> redisTemplate, Expression keyExpression) {
Assert.notNull(keyExpression, "'keyExpression' must not be null");
Assert.notNull(redisTemplate, "'redisTemplate' must not be null");
@@ -82,9 +78,7 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
* which should resolve to a 'key' name of the collection to be used.
* It will create and initialize an instance of {@link StringRedisTemplate} that uses
* {@link StringRedisSerializer} for all serialization.
*
* The 'keyExpression' will be evaluated on every call to the {@link #receive()} method.
*
* @param connectionFactory The connection factory.
* @param keyExpression The key expression.
*/
@@ -106,13 +100,17 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
this.collectionType = collectionType;
}
@Override
protected void onInit() {
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
/**
* Returns a Message with the view into a {@link RedisStore} identified
* by {@link #keyExpression}
*/
@Override
@SuppressWarnings("unchecked")
public Message<RedisStore> receive() {
protected RedisStore doReceive() {
String key = this.keyExpression.getValue(this.evaluationContext, String.class);
Assert.hasText(key, "Failed to determine the key for the collection");
@@ -124,11 +122,11 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
((IntegrationResourceHolder) holder).addAttribute("store", store);
}
if (store instanceof Collection<?> && ((Collection<Object>) store).size() < 1) {
if (store instanceof Collection<?> && ((Collection<?>) store).size() < 1) {
return null;
}
else {
return this.getMessageBuilderFactory().withPayload(store).build();
return store;
}
}
@@ -146,12 +144,6 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
return "redis:store-inbound-channel-adapter";
}
@Override
protected void onInit() throws Exception {
this.evaluationContext =
ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
}
public RedisStore getResource() {
return this.resourceHolder.get();
}
@@ -163,4 +155,5 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport
public void afterRollback(Object object) {
this.resourceHolder.remove();
}
}

9
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
View File

@@ -23,7 +23,6 @@ import static org.hamcrest.Matchers.equalTo;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays;
import java.util.Comparator;
@@ -120,7 +119,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
}
@Test
public void testMaxFetch() throws IOException {
public void testMaxFetch() throws Exception {
SftpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.afterPropertiesSet();
@@ -133,7 +132,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
}
@Test
public void testMaxFetchNoFilter() throws IOException {
public void testMaxFetchNoFilter() throws Exception {
SftpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(null);
messageSource.afterPropertiesSet();
@@ -146,9 +145,9 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
}
@Test
public void testMaxFetchLambdaFilter() throws IOException {
public void testMaxFetchLambdaFilter() throws Exception {
SftpStreamingMessageSource messageSource = buildSource();
messageSource.setFilter(f -> Arrays.asList(f));
messageSource.setFilter(Arrays::asList);
messageSource.afterPropertiesSet();
Message<InputStream> received = messageSource.receive();
assertNotNull(received);

23
spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamReadingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -20,11 +20,8 @@ import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.Message;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
/**
* A pollable source for receiving bytes from an {@link InputStream}.
@@ -32,11 +29,9 @@ import org.springframework.messaging.support.GenericMessage;
* @author Mark Fisher
* @author Artem Bilan
*/
public class ByteStreamReadingMessageSource extends IntegrationObjectSupport implements MessageSource<byte[]> {
public class ByteStreamReadingMessageSource extends AbstractMessageSource<byte[]> {
private BufferedInputStream stream;
private Object streamMonitor;
private final BufferedInputStream stream;
private int bytesPerMessage = 1024;
@@ -48,7 +43,6 @@ public class ByteStreamReadingMessageSource extends IntegrationObjectSupport imp
}
public ByteStreamReadingMessageSource(InputStream stream, int bufferSize) {
this.streamMonitor = stream;
if (stream instanceof BufferedInputStream) {
this.stream = (BufferedInputStream) stream;
}
@@ -74,11 +68,12 @@ public class ByteStreamReadingMessageSource extends IntegrationObjectSupport imp
return "stream:stdin-channel-adapter(byte)";
}
public Message<byte[]> receive() {
@Override
protected byte[] doReceive() {
try {
byte[] bytes;
int bytesRead = 0;
synchronized (this.streamMonitor) {
synchronized (this.stream) {
if (this.stream.available() == 0) {
return null;
}
@@ -89,12 +84,12 @@ public class ByteStreamReadingMessageSource extends IntegrationObjectSupport imp
return null;
}
if (!this.shouldTruncate) {
return new GenericMessage<byte[]>(bytes);
return bytes;
}
else {
byte[] result = new byte[bytesRead];
System.arraycopy(bytes, 0, result, 0, result.length);
return new GenericMessage<byte[]>(result);
return result;
}
}
catch (IOException e) {

21
spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -24,11 +24,8 @@ import java.io.UnsupportedEncodingException;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.Message;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
/**
@@ -36,14 +33,13 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*/
public class CharacterStreamReadingMessageSource extends IntegrationObjectSupport implements MessageSource<String>,
ApplicationEventPublisherAware {
public class CharacterStreamReadingMessageSource extends AbstractMessageSource<String>
implements ApplicationEventPublisherAware {
private final BufferedReader reader;
private final Object monitor;
private final boolean blockToDetectEOF;
private ApplicationEventPublisher applicationEventPublisher;
@@ -91,7 +87,6 @@ public class CharacterStreamReadingMessageSource extends IntegrationObjectSuppor
*/
public CharacterStreamReadingMessageSource(Reader reader, int bufferSize, boolean blockToDetectEOF) {
Assert.notNull(reader, "reader must not be null");
this.monitor = reader;
if (reader instanceof BufferedReader) {
this.reader = (BufferedReader) reader;
}
@@ -115,9 +110,9 @@ public class CharacterStreamReadingMessageSource extends IntegrationObjectSuppor
}
@Override
public Message<String> receive() {
public String doReceive() {
try {
synchronized (this.monitor) {
synchronized (this.reader) {
if (!this.blockToDetectEOF && !this.reader.ready()) {
return null;
}
@@ -125,7 +120,7 @@ public class CharacterStreamReadingMessageSource extends IntegrationObjectSuppor
if (line == null && this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new StreamClosedEvent(this));
}
return (line != null) ? new GenericMessage<String>(line) : null;
return line;
}
}
catch (IOException e) {