From 0d0605be783a2648d725e87e365295614c36616d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 25 Jul 2018 13:01:22 -0400 Subject: [PATCH] INT-4381: MessageSources refactoring (#2517) * 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` --- .../endpoint/MethodInvokingMessageSource.java | 5 +- .../ResourceRetrievingMessageSource.java | 9 ++- .../util/AbstractExpressionEvaluator.java | 8 +- .../xml/HeaderEnricherTests-context.xml | 2 +- .../config/xml/HeaderEnricherTests.java | 50 ++++++------ .../feed/inbound/FeedEntryMessageSource.java | 25 ++---- .../inbound/FeedEntryMessageSourceTests.java | 4 +- .../file/FileReadingMessageSource.java | 21 +---- .../dsl/FileInboundChannelAdapterSpec.java | 29 +++---- ...tractRemoteFileStreamingMessageSource.java | 2 +- ...InboundFileSynchronizingMessageSource.java | 3 +- .../file/AutoCreateDirectoryTests.java | 6 +- .../file/remote/StreamingInboundTests.java | 2 +- .../FtpStreamingMessageSourceTests.java | 5 +- ...onEvaluatingSqlParameterSourceFactory.java | 7 +- .../jdbc/JdbcPollingChannelAdapter.java | 30 ++----- .../jdbc/StoredProcPollingChannelAdapter.java | 81 ++++++------------- ...PollingChannelAdapterIntegrationTests.java | 8 +- .../jdbcInboundChannelAdapterCommonConfig.xml | 2 +- .../jms/JmsDestinationPollingSource.java | 50 ++++++------ .../JmsInboundChannelAdapterParser.java | 6 +- .../jpa/inbound/JpaPollingChannelAdapter.java | 29 +++---- .../mail/MailReceivingMessageSource.java | 26 +++--- .../mongodb/inbound/MongoDbMessageSource.java | 24 +++--- .../inbound/RedisStoreMessageSource.java | 39 ++++----- .../SftpStreamingMessageSourceTests.java | 9 +-- .../ByteStreamReadingMessageSource.java | 23 +++--- .../CharacterStreamReadingMessageSource.java | 21 ++--- 28 files changed, 207 insertions(+), 319 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java index 16ccfddb32..a335eb5e9a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MethodInvokingMessageSource.java @@ -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 } @Override - public void afterPropertiesSet() throws Exception { - super.afterPropertiesSet(); + protected void onInit() { synchronized (this.initializationMonitor) { if (this.initialized) { return; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceRetrievingMessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceRetrievingMessageSource.java index 70ce0e0ab1..1762d621c8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceRetrievingMessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/resource/ResourceRetrievingMessageSource.java @@ -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 @@ -79,8 +81,7 @@ public class ResourceRetrievingMessageSource extends AbstractMessageSource - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java index 7bfa50f647..01fbfdd7e8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java @@ -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("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("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("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("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("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("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("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("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("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("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>(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(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("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("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("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("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("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("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("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; } + } } diff --git a/spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java b/spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java index 5e8966dc95..dd64c31f71 100644 --- a/spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java +++ b/spring-integration-feed/src/main/java/org/springframework/integration/feed/inbound/FeedEntryMessageSource.java @@ -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 { +public class FeedEntryMessageSource extends AbstractMessageSource { private final URL feedUrl; @@ -145,18 +144,7 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements } @Override - public Message 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(); } } diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java index 01c2743eed..66ca4a4289 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/inbound/FeedEntryMessageSourceTests.java @@ -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 message1 = source.receive(); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java index 569ff5e0ee..2263db7b7a 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java @@ -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, Lifecycle { +public class FileReadingMessageSource extends AbstractMessageSource + implements Lifecycle { private static final int DEFAULT_INTERNAL_QUEUE_CAPACITY = 5; @@ -350,21 +350,6 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement } @Override - public Message receive() throws MessagingException { - AbstractIntegrationMessageBuilder messageBuilder = doReceive(); - - Message message = null; - - if (messageBuilder != null) { - message = messageBuilder.build(); - if (logger.isInfoEnabled()) { - logger.info("Created message: [" + message + "]"); - } - } - - return message; - } - protected AbstractIntegrationMessageBuilder doReceive() { // rescan only if needed or explicitly configured if (this.scanEachPoll || this.toBeReceived.isEmpty()) { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java index a47e9e504e..e2e6af4d7f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/FileInboundChannelAdapterSpec.java @@ -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 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 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); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java index a8530a4da6..3edbed7ae1 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java @@ -136,7 +136,7 @@ public abstract class AbstractRemoteFileStreamingMessageSource } @Override - public final void afterPropertiesSet() { + public final void onInit() { Assert.state(this.remoteDirectoryExpression != null, "'remoteDirectoryExpression' must not be null"); doInit(); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java index dded0de6f4..44525b074f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java @@ -173,8 +173,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource } @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()) { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/AutoCreateDirectoryTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/AutoCreateDirectoryTests.java index 3eacfea3ba..29822d9996 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/AutoCreateDirectoryTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/AutoCreateDirectoryTests.java @@ -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); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java index ee87c6b945..44c7013c8d 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java @@ -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); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java index 981055b713..00783c068a 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java @@ -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(); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java index 1760eb6082..4a3bd0f26b 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/ExpressionEvaluatingSqlParameterSourceFactory.java @@ -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 { diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java index 210f519497..84c75f5ce0 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapter.java @@ -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 { +public class JdbcPollingChannelAdapter extends AbstractMessageSource { 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 null. - * #return the {@link Message} or {@code null} as a result of query. - */ - @Override - public Message 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; diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapter.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapter.java index 3a5eae1304..393fcbd22a 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapter.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapter.java @@ -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 { +public class StoredProcPollingChannelAdapter extends AbstractMessageSource { 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 null. + * 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 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 resultMap = doPoll(); + Map 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 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; - } - } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java index a098da2dfa..2ef4ba8d7d 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcPollingChannelAdapterIntegrationTests.java @@ -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"); diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml index 1f3b824b5d..6ffe731e4b 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/jdbcInboundChannelAdapterCommonConfig.xml @@ -10,7 +10,7 @@ - + diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java index 2973346101..42c03991bf 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationPollingSource.java @@ -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 { +public class JmsDestinationPollingSource extends AbstractMessageSource { 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 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 builder = (object instanceof Message) ? - getMessageBuilderFactory().fromMessage((Message) 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); - } - } - } - } - } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java index ce442cef2a..5531ad07a6 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsInboundChannelAdapterParser.java @@ -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); diff --git a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapter.java b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapter.java index 6e9de2699a..75b8fb3ba8 100644 --- a/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapter.java +++ b/spring-integration-jpa/src/main/java/org/springframework/integration/jpa/inbound/JpaPollingChannelAdapter.java @@ -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 { +public class JpaPollingChannelAdapter extends AbstractMessageSource { 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 * null. Otherwise, a new {@link Message} is constructed and returned. */ - public Message 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 diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java index d98300cf8f..97b2010627 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java @@ -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, - BeanFactoryAware, BeanNameAware, NamedComponent { +public class MailReceivingMessageSource extends AbstractMessageSource { private final Log logger = LogFactory.getLog(this.getClass()); private final MailReceiver mailReceiver; - private final Queue mailQueue = new ConcurrentLinkedQueue(); + private final Queue 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, this.beanName = name; } - @SuppressWarnings("unchecked") @Override - public Message receive() { + protected Object doReceive() { try { Object mailMessage = this.mailQueue.poll(); if (mailMessage == null) { @@ -120,10 +116,10 @@ public class MailReceivingMessageSource implements MessageSource, this.logger.debug("received mail message [" + mailMessage + "]"); } if (mailMessage instanceof Message) { - return (Message) mailMessage; + return getMessageBuilderFactory().fromMessage((Message) mailMessage); } else { - return getMessageBuilderFactory().withPayload(mailMessage).build(); + return getMessageBuilderFactory().withPayload(mailMessage); } } } diff --git a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java index 973fa49eb1..df79f8c470 100644 --- a/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java +++ b/spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/inbound/MongoDbMessageSource.java @@ -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 { +public class MongoDbMessageSource extends AbstractMessageSource { 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 receive() { + protected Object doReceive() { Assert.isTrue(this.initialized, "This class is not yet initialized. Invoke its afterPropertiesSet() method"); - Message message = null; + AbstractIntegrationMessageBuilder 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; } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java index d0698d6201..d1c798283c 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java @@ -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 { +public class RedisStoreMessageSource extends AbstractMessageSource { - private final ThreadLocal resourceHolder = new ThreadLocal(); + private final ThreadLocal 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 redisTemplate, - Expression keyExpression) { - + public RedisStoreMessageSource(RedisTemplate 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 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) 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(); } + } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java index 9f1bc90e12..b5e5a8e908 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java @@ -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 received = messageSource.receive(); assertNotNull(received); diff --git a/spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamReadingMessageSource.java b/spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamReadingMessageSource.java index 8be6c3b9c6..2d3a16f8d3 100644 --- a/spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamReadingMessageSource.java +++ b/spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamReadingMessageSource.java @@ -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 { +public class ByteStreamReadingMessageSource extends AbstractMessageSource { - 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 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(bytes); + return bytes; } else { byte[] result = new byte[bytesRead]; System.arraycopy(bytes, 0, result, 0, result.length); - return new GenericMessage(result); + return result; } } catch (IOException e) { diff --git a/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java b/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java index b8fc4beccf..c92bbb1380 100644 --- a/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java +++ b/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamReadingMessageSource.java @@ -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, - ApplicationEventPublisherAware { +public class CharacterStreamReadingMessageSource extends AbstractMessageSource + 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 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(line) : null; + return line; } } catch (IOException e) {