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

PollingSourceEndpoint is now just SourceEndpoint.

Browse Source
This commit is contained in:
Mark Fisher
2008-05-19 17:09:32 +00:00
parent 73f4a0c7de
commit 019ec2123d
13 changed files with 179 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceParserTests.java
View File

@@ -27,7 +27,6 @@ import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.adapter.jms.JmsSource;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.message.Message;
/**
@@ -120,8 +119,6 @@ public class JmsSourceParserTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"jmsSourceEndpoint.xml", this.getClass());
context.start();
PollingSourceEndpoint endpoint = (PollingSourceEndpoint) context.getBean("endpoint");
assertEquals(JmsSource.class, endpoint.getSource().getClass());
MessageChannel channel = (MessageChannel) context.getBean("channel");
Message<?> message = channel.receive(3000);
assertNotNull("message should not be null", message);

14
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java
View File

@@ -25,7 +25,7 @@ import org.junit.Test;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.scheduling.PollingSchedule;
@@ -42,7 +42,7 @@ public class ByteStreamSourceTests {
ByteStreamSource source = new ByteStreamSource(stream);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.run();
Message<?> message1 = channel.receive(500);
byte[] payload = (byte[]) message1.getPayload();
@@ -66,7 +66,7 @@ public class ByteStreamSourceTests {
source.setBytesPerMessage(8);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(5);
endpoint.run();
Message<?> message1 = channel.receive(500);
@@ -84,7 +84,7 @@ public class ByteStreamSourceTests {
source.setBytesPerMessage(4);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(1);
endpoint.run();
Message<?> message1 = channel.receive(0);
@@ -109,7 +109,7 @@ public class ByteStreamSourceTests {
source.setBytesPerMessage(4);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(5);
endpoint.run();
Message<?> message1 = channel.receive(0);
@@ -133,7 +133,7 @@ public class ByteStreamSourceTests {
source.setBytesPerMessage(4);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(1);
endpoint.run();
Message<?> message1 = channel.receive(0);
@@ -155,7 +155,7 @@ public class ByteStreamSourceTests {
source.setShouldTruncate(false);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(1);
endpoint.run();
Message<?> message1 = channel.receive(0);

10
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java
View File

@@ -25,7 +25,7 @@ import org.junit.Test;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.scheduling.PollingSchedule;
@@ -41,7 +41,7 @@ public class CharacterStreamSourceTests {
CharacterStreamSource source = new CharacterStreamSource(reader);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.run();
Message<?> message1 = channel.receive(0);
assertEquals("test", message1.getPayload());
@@ -59,7 +59,7 @@ public class CharacterStreamSourceTests {
CharacterStreamSource source = new CharacterStreamSource(reader);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(5);
endpoint.run();
Message<?> message1 = channel.receive(0);
@@ -76,7 +76,7 @@ public class CharacterStreamSourceTests {
CharacterStreamSource source = new CharacterStreamSource(reader);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(1);
endpoint.run();
Message<?> message1 = channel.receive(0);
@@ -96,7 +96,7 @@ public class CharacterStreamSourceTests {
CharacterStreamSource source = new CharacterStreamSource(reader);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(5000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(5);
endpoint.run();
Message<?> message1 = channel.receive(500);

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

@@ -48,7 +48,7 @@ import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.dispatcher.SynchronousChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
@@ -163,7 +163,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
PollingSchedule schedule = new PollingSchedule(period);
schedule.setInitialDelay(initialDelay);
schedule.setFixedRate(fixedRate);
PollingSourceEndpoint sourceEndpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint sourceEndpoint = new SourceEndpoint(source, channel, schedule);
String channelName = beanName + "-inputChannel";
messageBus.registerChannel(channelName, channel);
messageBus.registerEndpoint(beanName + "-sourceEndpoint", sourceEndpoint);

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

@@ -22,7 +22,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.StringUtils;
@@ -36,7 +36,7 @@ import org.springframework.util.xml.DomUtils;
public class SourceEndpointParser extends AbstractSimpleBeanDefinitionParser {
protected final Class<?> getBeanClass(Element element) {
return PollingSourceEndpoint.class;
return SourceEndpoint.class;
}
protected boolean shouldGenerateId() {

40
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractSourceEndpoint.java → spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java
View File

@@ -19,48 +19,32 @@ package org.springframework.integration.endpoint;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Source;
import org.springframework.util.Assert;
/**
* Base class for {@link SourceEndpoint} implementations.
* Base class for {@link MessageEndpoint} implementations.
*
* @author Mark Fisher
*/
public abstract class AbstractSourceEndpoint implements SourceEndpoint {
public abstract class AbstractEndpoint implements MessageEndpoint {
protected final Log logger = LogFactory.getLog(this.getClass());
private final Source<?> source;
private final MessageChannel channel;
private volatile String name;
public AbstractSourceEndpoint(Source<?> source, MessageChannel channel) {
Assert.notNull(source, "source must not be null");
Assert.notNull(channel, "channel must not be null");
this.source = source;
this.channel = channel;
}
public Source<?> getSource() {
return this.source;
}
public void setBeanName(String name) {
this.name = name;
}
public String getName() {
return this.name;
}
protected MessageChannel getChannel() {
return this.channel;
public void setName(String name) {
this.name = name;
}
public void setBeanName(String beanName) {
this.setName(beanName);
}
public String toString() {
return (this.name != null) ? this.name : super.toString();
}
}

158
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java
View File

@@ -1,158 +0,0 @@
/*
* Copyright 2002-2008 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.endpoint;
import java.lang.reflect.Method;
import java.util.List;
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.StaticMethodMatcherPointcutAdvisor;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.dispatcher.PollingDispatcherTask;
import org.springframework.integration.message.Source;
import org.springframework.integration.scheduling.MessagingTask;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
* A channel adapter that retrieves messages from a {@link Source}
* and then sends the resulting messages to the provided {@link MessageChannel}.
*
* @author Mark Fisher
*/
public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask, InitializingBean {
private final Schedule schedule;
private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
private volatile PollingDispatcher dispatcher;
private volatile List<Advice> dispatchAdviceChain;
private volatile MessagingTask task;
private volatile List<Advice> taskAdviceChain;
private volatile boolean taskInitialized;
private final Object taskMonitor = new Object();
public PollingSourceEndpoint(Source<?> source, MessageChannel channel, Schedule schedule) {
super(source, channel);
Assert.notNull(schedule, "schedule must not be null");
this.dispatcher = new DefaultPollingDispatcher(source, this.dispatcherPolicy);
this.dispatcher.subscribe(this.getChannel());
this.schedule = schedule;
}
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
this.dispatcherPolicy.setMaxMessagesPerTask(maxMessagesPerTask);
}
public void setSendTimeout(long sendTimeout) {
this.dispatcher.setSendTimeout(sendTimeout);
}
public void setTaskAdviceChain(List<Advice> taskAdviceChain) {
this.taskAdviceChain = taskAdviceChain;
}
public void setDispatchAdviceChain(List<Advice> dispatchAdviceChain) {
this.dispatchAdviceChain = dispatchAdviceChain;
}
public Schedule getSchedule() {
return this.schedule;
}
public void afterPropertiesSet() {
this.initializeTask();
}
public void initializeTask() {
synchronized (this.taskMonitor) {
if (this.taskInitialized) {
return;
}
this.refreshTask();
this.taskInitialized = true;
}
}
public void refreshTask() {
synchronized (this.taskMonitor) {
PollingDispatcher dispatcherProxy = null;
if (this.dispatchAdviceChain != null && this.dispatchAdviceChain.size() > 0) {
ProxyFactory proxyFactory = new ProxyFactory(this.dispatcher);
proxyFactory.setInterfaces(new Class[] { PollingDispatcher.class });
for (Advice advice : this.dispatchAdviceChain) {
proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "dispatch"));
}
dispatcherProxy = (PollingDispatcher) proxyFactory.getProxy();
}
this.task = new PollingDispatcherTask((dispatcherProxy != null) ? dispatcherProxy : this.dispatcher, this.schedule);
if (this.taskAdviceChain != null && this.taskAdviceChain.size() > 0) {
ProxyFactory proxyFactory = new ProxyFactory(this.task);
proxyFactory.setInterfaces(new Class[] { MessagingTask.class });
for (Advice advice : this.taskAdviceChain) {
proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "run"));
}
this.task = (MessagingTask) proxyFactory.getProxy();
}
}
}
private MessagingTask getTask() {
synchronized (this.taskMonitor) {
if (!this.taskInitialized) {
this.initializeTask();
}
return this.task;
}
}
public void run() {
this.getTask().run();
}
@SuppressWarnings("serial")
private static class MethodNameAdvisor extends StaticMethodMatcherPointcutAdvisor {
private final String methodName;
MethodNameAdvisor(Advice advice, String methodName) {
super(advice);
this.methodName = methodName;
}
@SuppressWarnings("unchecked")
public boolean matches(Method method, Class targetClass) {
return method.getName().equals(methodName);
}
}
}

135
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourceEndpoint.java
View File

@@ -16,15 +16,144 @@
package org.springframework.integration.endpoint;
import java.lang.reflect.Method;
import java.util.List;
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.StaticMethodMatcherPointcutAdvisor;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.dispatcher.PollingDispatcherTask;
import org.springframework.integration.message.Source;
import org.springframework.integration.scheduling.MessagingTask;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
* Base interface for source endpoints.
* A channel adapter that retrieves messages from a {@link Source}
* and then sends the resulting messages to the provided {@link MessageChannel}.
*
* @author Mark Fisher
*/
public interface SourceEndpoint extends MessageEndpoint {
public class SourceEndpoint extends AbstractEndpoint implements MessagingTask, InitializingBean {
Source getSource();
private final Schedule schedule;
private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
private volatile PollingDispatcher dispatcher;
private volatile List<Advice> dispatchAdviceChain;
private volatile MessagingTask task;
private volatile List<Advice> taskAdviceChain;
private volatile boolean taskInitialized;
private final Object taskMonitor = new Object();
public SourceEndpoint(Source<?> source, MessageChannel channel, Schedule schedule) {
Assert.notNull(source, "source must not be null");
Assert.notNull(channel, "channel must not be null");
Assert.notNull(schedule, "schedule must not be null");
this.dispatcher = new DefaultPollingDispatcher(source, this.dispatcherPolicy);
this.dispatcher.subscribe(channel);
this.schedule = schedule;
}
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
this.dispatcherPolicy.setMaxMessagesPerTask(maxMessagesPerTask);
}
public void setSendTimeout(long sendTimeout) {
this.dispatcher.setSendTimeout(sendTimeout);
}
public void setTaskAdviceChain(List<Advice> taskAdviceChain) {
this.taskAdviceChain = taskAdviceChain;
}
public void setDispatchAdviceChain(List<Advice> dispatchAdviceChain) {
this.dispatchAdviceChain = dispatchAdviceChain;
}
public Schedule getSchedule() {
return this.schedule;
}
public void afterPropertiesSet() {
this.initializeTask();
}
public void initializeTask() {
synchronized (this.taskMonitor) {
if (this.taskInitialized) {
return;
}
this.refreshTask();
this.taskInitialized = true;
}
}
public void refreshTask() {
synchronized (this.taskMonitor) {
PollingDispatcher dispatcherProxy = null;
if (this.dispatchAdviceChain != null && this.dispatchAdviceChain.size() > 0) {
ProxyFactory proxyFactory = new ProxyFactory(this.dispatcher);
proxyFactory.setInterfaces(new Class[] { PollingDispatcher.class });
for (Advice advice : this.dispatchAdviceChain) {
proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "dispatch"));
}
dispatcherProxy = (PollingDispatcher) proxyFactory.getProxy();
}
this.task = new PollingDispatcherTask((dispatcherProxy != null) ? dispatcherProxy : this.dispatcher, this.schedule);
if (this.taskAdviceChain != null && this.taskAdviceChain.size() > 0) {
ProxyFactory proxyFactory = new ProxyFactory(this.task);
proxyFactory.setInterfaces(new Class[] { MessagingTask.class });
for (Advice advice : this.taskAdviceChain) {
proxyFactory.addAdvisor(new MethodNameAdvisor(advice, "run"));
}
this.task = (MessagingTask) proxyFactory.getProxy();
}
}
}
private MessagingTask getTask() {
synchronized (this.taskMonitor) {
if (!this.taskInitialized) {
this.initializeTask();
}
return this.task;
}
}
public void run() {
this.getTask().run();
}
@SuppressWarnings("serial")
private static class MethodNameAdvisor extends StaticMethodMatcherPointcutAdvisor {
private final String methodName;
MethodNameAdvisor(Advice advice, String methodName) {
super(advice);
this.methodName = methodName;
}
@SuppressWarnings("unchecked")
public boolean matches(Method method, Class targetClass) {
return method.getName().equals(methodName);
}
}
}

27
spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java
View File

@@ -25,9 +25,6 @@ import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
@@ -44,15 +41,11 @@ import org.springframework.integration.util.ErrorHandler;
import org.springframework.util.Assert;
/**
* Base class for {@link MessageEndpoint} implementations.
* Base class for {@link MessageEndpoint} implementations to which Messages may be sent.
*
* @author Mark Fisher
*/
public class TargetEndpoint implements Target, MessageEndpoint, ChannelRegistryAware, InitializingBean, Lifecycle {
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile String name;
public class TargetEndpoint extends AbstractEndpoint implements Target, ChannelRegistryAware, InitializingBean, Lifecycle {
private volatile Target target;
@@ -80,18 +73,6 @@ public class TargetEndpoint implements Target, MessageEndpoint, ChannelRegistryA
}
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
public void setBeanName(String beanName) {
this.setName(beanName);
}
public Target getTarget() {
return this.target;
}
@@ -227,8 +208,4 @@ public class TargetEndpoint implements Target, MessageEndpoint, ChannelRegistryA
}
}
public String toString() {
return (this.name != null) ? this.name : super.toString();
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java
View File

@@ -33,7 +33,7 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.RendezvousChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.PollingSourceEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
@@ -172,7 +172,7 @@ public class MessageBusTests {
public void testErrorChannelWithFailedDispatch() throws InterruptedException {
MessageBus bus = new MessageBus();
CountDownLatch latch = new CountDownLatch(1);
PollingSourceEndpoint sourceEndpoint = new PollingSourceEndpoint(new FailingSource(latch), new QueueChannel(), new PollingSchedule(1000));
SourceEndpoint sourceEndpoint = new SourceEndpoint(new FailingSource(latch), new QueueChannel(), new PollingSchedule(1000));
bus.registerEndpoint("testEndpoint", sourceEndpoint);
bus.start();
latch.await(1000, TimeUnit.MILLISECONDS);

24
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java → spring-integration-core/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java
View File

@@ -41,14 +41,14 @@ import org.springframework.integration.scheduling.PollingSchedule;
/**
* @author Mark Fisher
*/
public class PollingSourceEndpointTests {
public class SourceEndpointTests {
@Test
public void testPolledSourceSendsToChannel() {
TestSource source = new TestSource("testing", 1);
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(100);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.run();
Message<?> message = channel.receive(1000);
assertNotNull("message should not be null", message);
@@ -61,7 +61,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel(1);
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setSendTimeout(10);
endpoint.run();
Message<?> message1 = channel.receive(1000);
@@ -82,7 +82,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
endpoint.setMaxMessagesPerTask(5);
endpoint.run();
Message<?> message1 = channel.receive(0);
@@ -104,7 +104,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
final StringBuffer buffer = new StringBuffer();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
taskAdviceChain.add(new MethodBeforeAdvice() {
@@ -138,7 +138,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
final StringBuffer buffer = new StringBuffer();
List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
dispatchAdviceChain.add(new MethodBeforeAdvice() {
@@ -172,7 +172,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
final StringBuffer buffer = new StringBuffer();
@@ -216,7 +216,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
final StringBuffer buffer = new StringBuffer();
@@ -278,7 +278,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
final StringBuffer buffer = new StringBuffer();
@@ -317,7 +317,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
final StringBuffer buffer = new StringBuffer();
@@ -356,7 +356,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
final StringBuffer buffer = new StringBuffer();
@@ -396,7 +396,7 @@ public class PollingSourceEndpointTests {
QueueChannel channel = new QueueChannel();
PollingSchedule schedule = new PollingSchedule(1000);
schedule.setInitialDelay(10000);
PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source, channel, schedule);
List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
List<Advice> taskAdviceChain = new ArrayList<Advice>();
final StringBuffer buffer = new StringBuffer();

2
spring-integration-core/src/test/java/org/springframework/integration/handler/adapterTests.xml
View File

@@ -10,7 +10,7 @@
<bean id="channel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="sourceEndpoint" class="org.springframework.integration.endpoint.PollingSourceEndpoint">
<bean id="sourceEndpoint" class="org.springframework.integration.endpoint.SourceEndpoint">
<constructor-arg>
<bean class="org.springframework.integration.message.MethodInvokingSource">
<property name="object">

4
spring-integration-samples/src/main/java/org/springframework/integration/samples/cafe/Barista.java
View File

@@ -26,9 +26,9 @@ import org.springframework.stereotype.Component;
@Component
public class Barista {
private long hotDrinkDelay = 1000;
private long hotDrinkDelay = 2000;
private long coldDrinkDelay = 700;
private long coldDrinkDelay = 1000;
private AtomicInteger hotDrinkCounter = new AtomicInteger();