diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java new file mode 100644 index 0000000000..2db89fb6ca --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java @@ -0,0 +1,84 @@ +/* + * Copyright 2002-2007 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.adapter; + +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageMapper; +import org.springframework.integration.message.SimplePayloadMessageMapper; +import org.springframework.util.Assert; + +/** + * Target adapter implementation that delegates to a {@link MessageMapper} + * and then passes the resulting object to the provided {@link Target}. + * + * @author Mark Fisher + */ +public class DefaultTargetAdapter implements TargetAdapter { + + private String name; + + private MessageChannel channel; + + private Target target; + + private MessageMapper mapper = new SimplePayloadMessageMapper(); + + private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(5); + + + public DefaultTargetAdapter(Target target) { + this.target = target; + } + + public void setName(String name) { + this.name = name; + } + + public String getName() { + return this.name; + } + + public void setChannel(MessageChannel channel) { + Assert.notNull(channel, "'channel' must not be null"); + this.channel = channel; + } + + public MessageChannel getChannel() { + return this.channel; + } + + public void setMessageMapper(MessageMapper mapper) { + Assert.notNull(mapper, "'mapper' must not be null"); + this.mapper = mapper; + } + + public void setConsumerPolicy(ConsumerPolicy policy) { + Assert.notNull(policy, "'policy' must not be null"); + this.policy = policy; + } + + public ConsumerPolicy getConsumerPolicy() { + return this.policy; + } + + public void messageReceived(Message message) { + this.target.send(this.mapper.fromMessage(message)); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingTarget.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingTarget.java new file mode 100644 index 0000000000..3d40ad2a46 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingTarget.java @@ -0,0 +1,78 @@ +/* + * Copyright 2002-2007 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.adapter; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.endpoint.ArgumentListPreparer; +import org.springframework.integration.endpoint.SimpleMethodInvoker; +import org.springframework.util.Assert; + +/** + * A messaging target that invokes the specified method on the provided object. + * + * @author Mark Fisher + */ +public class MethodInvokingTarget implements Target, InitializingBean { + + private Log logger = LogFactory.getLog(this.getClass()); + + private T object; + + private String method; + + private SimpleMethodInvoker invoker; + + private ArgumentListPreparer argumentListPreparer; + + + public void setObject(T object) { + Assert.notNull(object, "'object' must not be null"); + this.object = object; + } + + public void setMethod(String method) { + Assert.notNull(method, "'method' must not be null"); + this.method = method; + } + + public void setArgumentListPreparer(ArgumentListPreparer argumentListPreparer) { + this.argumentListPreparer = argumentListPreparer; + } + + public void afterPropertiesSet() { + this.invoker = new SimpleMethodInvoker(this.object, this.method); + } + + public boolean send(Object object) { + Object args[] = null; + if (this.argumentListPreparer != null) { + args = this.argumentListPreparer.prepare(object); + } + else { + args = new Object[] { object }; + } + Object result = this.invoker.invokeMethod(args); + if (result != null && logger.isWarnEnabled()) { + logger.warn("ignoring outbound channel adapter's return value"); + } + return true; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index a3f1b77d02..3e6ce5f962 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -86,7 +86,7 @@ public class PollingSourceAdapter implements SourceAdapter, MessageDispatcher return this.policy; } - public int receiveAndDispatch() { + public int dispatch() { int messagesProcessed = 0; int limit = this.policy.getMaxMessagesPerTask(); Collection results = this.source.poll(limit); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/Target.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/Target.java new file mode 100644 index 0000000000..8b2c4dae92 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/Target.java @@ -0,0 +1,29 @@ +/* + * Copyright 2002-2007 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.adapter; + +/** + * Interface for any external target that may receive data from an outgoing + * channel adapter. + * + * @author Mark Fisher + */ +public interface Target { + + boolean send(T t); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java new file mode 100644 index 0000000000..b2380eca08 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java @@ -0,0 +1,36 @@ +/* + * Copyright 2002-2007 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.adapter; + +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.MessageReceiver; + +/** + * Base interface for target adapters. + * + * @author Mark Fisher + */ +public interface TargetAdapter extends MessageReceiver { + + String getName(); + + void setChannel(MessageChannel channel); + + ConsumerPolicy getConsumerPolicy(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java index 5ac976386d..68f54b047b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java @@ -37,7 +37,7 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher { private MessageRetriever retriever; - private List endpointExecutors = new CopyOnWriteArrayList(); + private List executors = new CopyOnWriteArrayList(); public AbstractMessageDispatcher(MessageRetriever retriever) { @@ -45,20 +45,21 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher { } - public void addEndpointExecutor(EndpointExecutor executor) { + public void addExecutor(MessageReceivingExecutor executor) { executor.start(); - this.endpointExecutors.add(executor); + this.executors.add(executor); } - protected List getEndpointExecutors() { - return this.endpointExecutors; + protected List getExecutors() { + return this.executors; } /** - * Receives messages and dispatches to the endpoints. Returns the number of - * messages processed. + * Retrieves messages and dispatches to the executors. + * + * @return the number of messages processed */ - public int receiveAndDispatch() { + public int dispatch() { int messagesProcessed = 0; Collection> messages = this.retriever.retrieveMessages(); if (messages == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index e33c09a12b..b7f27a26a8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -30,12 +30,15 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.integration.MessagingException; +import org.springframework.integration.adapter.DefaultTargetAdapter; import org.springframework.integration.adapter.SourceAdapter; +import org.springframework.integration.adapter.TargetAdapter; import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; -import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.message.MessageReceiver; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -53,9 +56,11 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private Map endpoints = new ConcurrentHashMap(); + private Map targetAdapters = new ConcurrentHashMap(); + private List dispatcherTasks = new CopyOnWriteArrayList(); - private Map endpointExecutors = new ConcurrentHashMap(); + private Map receiverExecutors = new ConcurrentHashMap(); private ScheduledThreadPoolExecutor dispatcherExecutor; @@ -70,6 +75,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif Assert.notNull(applicationContext, "applicationContext must not be null"); this.registerChannels(applicationContext); this.registerEndpoints(applicationContext); + this.registerSourceAdapters(applicationContext); + this.registerTargetAdapters(applicationContext); this.activateSubscriptions(applicationContext); } @@ -95,6 +102,24 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } + @SuppressWarnings("unchecked") + private void registerSourceAdapters(ApplicationContext context) { + Map sourceAdapterBeans = + (Map) context.getBeansOfType(SourceAdapter.class); + for (Map.Entry entry : sourceAdapterBeans.entrySet()) { + this.registerSourceAdapter(entry.getKey(), entry.getValue()); + } + } + + @SuppressWarnings("unchecked") + private void registerTargetAdapters(ApplicationContext context) { + Map targetAdapterBeans = + (Map) context.getBeansOfType(TargetAdapter.class); + for (Map.Entry entry : targetAdapterBeans.entrySet()) { + this.registerTargetAdapter(entry.getKey(), entry.getValue()); + } + } + @SuppressWarnings("unchecked") private void activateSubscriptions(ApplicationContext context) { Map subscriptionBeans = @@ -151,7 +176,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - public void registerSourceAdapter(SourceAdapter adapter) { + public void registerSourceAdapter(String name, SourceAdapter adapter) { + // TODO: use the name if (adapter instanceof MessageDispatcher) { ConsumerPolicy policy = adapter.getConsumerPolicy(); DispatcherTask dispatcherTask = new DispatcherTask((MessageDispatcher) adapter, policy); @@ -159,6 +185,20 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } + public void registerTargetAdapter(String name, TargetAdapter targetAdapter) { + if (targetAdapter instanceof DefaultTargetAdapter) { + DefaultTargetAdapter adapter = (DefaultTargetAdapter) targetAdapter; + adapter.setName(name); + this.targetAdapters.put(name, targetAdapter); + MessageChannel channel = adapter.getChannel(); + ConsumerPolicy policy = adapter.getConsumerPolicy(); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); + dispatcher.addExecutor(new MessageReceivingExecutor(adapter, policy.getConcurrency(), policy.getMaxConcurrency())); + this.addDispatcherTask(new DispatcherTask(dispatcher, policy)); + } + } + public void activateSubscription(Subscription subscription) { String channelName = subscription.getChannel(); String endpointName = subscription.getEndpoint(); @@ -183,19 +223,19 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif logger.info("activated subscription to channel '" + channelName + "' for endpoint '" + endpointName + "'"); } - EndpointExecutor endpointExecutor = new EndpointExecutor(endpoint, policy.getConcurrency(), policy.getMaxConcurrency()); - endpointExecutors.put(endpoint, endpointExecutor); + MessageReceivingExecutor executor = new MessageReceivingExecutor(endpoint, policy.getConcurrency(), policy.getMaxConcurrency()); + receiverExecutors.put(endpoint, executor); MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); - dispatcher.addEndpointExecutor(endpointExecutor); + dispatcher.addExecutor(executor); DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); if (this.isRunning()) { - endpointExecutor.start(); + executor.start(); } this.addDispatcherTask(dispatcherTask); if (this.logger.isInfoEnabled()) { logger.info("registered dispatcher task: channel='" + - channelName + "' endpoint='" + endpointName + "'"); + channelName + "' receiver='" + endpointName + "'"); } } @@ -209,10 +249,13 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - public int getActiveCountForEndpoint(String endpointName) { - MessageEndpoint endpoint = this.endpoints.get(endpointName); - if (endpoint != null) { - EndpointExecutor executor = this.endpointExecutors.get(endpoint); + public int getActiveCountForReceiver(String receiverName) { + MessageReceiver receiver = this.endpoints.get(receiverName); + if (receiver == null) { + receiver = this.targetAdapters.get(receiverName); + } + if (receiver != null) { + MessageReceivingExecutor executor = this.receiverExecutors.get(receiver); if (executor != null) { return executor.getActiveCount(); } @@ -252,7 +295,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif synchronized (this.lifecycleMonitor) { if (!this.isRunning()) { this.running = true; - for (EndpointExecutor executor : endpointExecutors.values()) { + for (MessageReceivingExecutor executor : receiverExecutors.values()) { executor.start(); } for (DispatcherTask task : this.dispatcherTasks) { @@ -266,7 +309,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif synchronized (this.lifecycleMonitor) { if (this.isRunning()) { this.running = false; - for (EndpointExecutor executor : endpointExecutors.values()) { + for (MessageReceivingExecutor executor : receiverExecutors.values()) { executor.stop(); } this.dispatcherExecutor.shutdownNow(); @@ -293,7 +336,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void run() { - dispatcher.receiveAndDispatch(); + dispatcher.dispatch(); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java index 6d37db2f45..7e070d6399 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java @@ -23,6 +23,6 @@ package org.springframework.integration.bus; */ public interface MessageDispatcher { - int receiveAndDispatch(); + int dispatch(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageReceivingExecutor.java similarity index 79% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java rename to spring-integration-core/src/main/java/org/springframework/integration/bus/MessageReceivingExecutor.java index 0fa4cf019d..4a42a0557c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageReceivingExecutor.java @@ -25,8 +25,8 @@ import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageReceiver; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; @@ -35,11 +35,11 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class EndpointExecutor implements Lifecycle { +public class MessageReceivingExecutor implements Lifecycle { private Log logger = LogFactory.getLog(this.getClass()); - private MessageEndpoint endpoint; + private MessageReceiver receiver; private ThreadPoolExecutor threadPoolExecutor; @@ -60,12 +60,12 @@ public class EndpointExecutor implements Lifecycle { private int totalErrorThreshold = -1; - public EndpointExecutor(MessageEndpoint endpoint, int corePoolSize, int maxPoolSize) { - Assert.notNull(endpoint, "'endpoint' must not be null"); + public MessageReceivingExecutor(MessageReceiver receiver, int corePoolSize, int maxPoolSize) { + Assert.notNull(receiver, "'receiver' must not be null"); Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1"); Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1"); Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'"); - this.endpoint = endpoint; + this.receiver = receiver; this.corePoolSize = corePoolSize; this.maxPoolSize = maxPoolSize; } @@ -85,7 +85,7 @@ public class EndpointExecutor implements Lifecycle { public void start() { synchronized (this.lifecycleMonitor) { if (!this.running) { - this.threadPoolExecutor = new EndpointThreadPoolExecutor(this.corePoolSize, this.maxPoolSize); + this.threadPoolExecutor = new MessageReceivingThreadPoolExecutor(this.corePoolSize, this.maxPoolSize); } this.running = true; } @@ -105,12 +105,12 @@ public class EndpointExecutor implements Lifecycle { if (threadPoolExecutor == null) { throw new MessageHandlingException("executor is not running"); } - this.threadPoolExecutor.execute(new EndpointTask(this.endpoint, message)); + this.threadPoolExecutor.execute(new MessageReceivingTask(this.receiver, message)); } /** * Set the maximum number of errors allowed in successive - * endpoint executions. If this threshold is ever exceeded, the executor + * executions. If this threshold is ever exceeded, the executor * will shutdown. */ public void setSuccessiveErrorThreshold(int successiveErrorThreshold) { @@ -118,9 +118,8 @@ public class EndpointExecutor implements Lifecycle { } /** - * Set the maximum number of total errors allowed in endpoint - * executions. If this threshold is ever exceeded, the executor will - * shutdown. + * Set the maximum number of total errors allowed in executions + * If this threshold is ever exceeded, the executor will shutdown. */ public void setTotalErrorThreshold(int totalErrorThreshold) { this.totalErrorThreshold = totalErrorThreshold; @@ -138,17 +137,17 @@ public class EndpointExecutor implements Lifecycle { } - private static class EndpointTask implements Runnable { + private static class MessageReceivingTask implements Runnable { - private MessageEndpoint endpoint; + private MessageReceiver receiver; private Message message; private Throwable error; - EndpointTask(MessageEndpoint endpoint, Message message) { - this.endpoint = endpoint; + MessageReceivingTask(MessageReceiver receiver, Message message) { + this.receiver = receiver; this.message = message; } @@ -158,7 +157,7 @@ public class EndpointExecutor implements Lifecycle { public void run() { try { - this.endpoint.messageReceived(this.message); + this.receiver.messageReceived(this.message); } catch (Throwable t) { this.error = t; @@ -167,9 +166,9 @@ public class EndpointExecutor implements Lifecycle { } - private class EndpointThreadPoolExecutor extends ThreadPoolExecutor { + private class MessageReceivingThreadPoolExecutor extends ThreadPoolExecutor { - public EndpointThreadPoolExecutor(int corePoolSize, int maximumPoolSize) { + public MessageReceivingThreadPoolExecutor(int corePoolSize, int maximumPoolSize) { super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue()); CustomizableThreadFactory threadFactory = new CustomizableThreadFactory(); threadFactory.setThreadNamePrefix("endpoint-executor-"); @@ -178,10 +177,10 @@ public class EndpointExecutor implements Lifecycle { @Override protected void afterExecute(Runnable r, Throwable t) { - EndpointTask task = (EndpointTask) r; + MessageReceivingTask task = (MessageReceivingTask) r; if (task.getError() != null) { if (logger.isWarnEnabled()) { - logger.warn("Exception occurred in endpoint execution", task.getError()); + logger.warn("Exception occurred during task execution", task.getError()); } successiveErrorCount++; totalErrorCount++; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java index 59c738d235..fd38e5a3ac 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java @@ -23,7 +23,7 @@ import org.springframework.integration.message.Message; /** * A {@link MessageDispatcher} implementation that dispatches each retrieved - * {@link Message} to a single {@link EndpointExecutor}. + * {@link Message} to a single {@link MessageReceivingExecutor}. * * @author Mark Fisher */ @@ -41,19 +41,19 @@ public class UnicastMessageDispatcher extends AbstractMessageDispatcher { @Override protected boolean dispatchMessage(Message message) { int attempts = 0; - Iterator iter = this.getEndpointExecutors().iterator(); + Iterator iter = this.getExecutors().iterator(); if (!iter.hasNext()) { if (logger.isWarnEnabled()) { - logger.warn("dispatcher has no active endpoint executors"); + logger.warn("dispatcher has no active executors"); } return false; } while (iter.hasNext()) { - EndpointExecutor executor = iter.next(); + MessageReceivingExecutor executor = iter.next(); try { if (executor == null || !executor.isRunning()) { if (logger.isInfoEnabled()) { - logger.info("removing inactive endpoint executor"); + logger.info("removing inactive executor"); } iter.remove(); continue; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java index 54a1050914..8c6d776005 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java @@ -25,8 +25,10 @@ import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.BeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.MessagingConfigurationException; -import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter; -import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter; +import org.springframework.integration.adapter.DefaultTargetAdapter; +import org.springframework.integration.adapter.MethodInvokingSource; +import org.springframework.integration.adapter.MethodInvokingTarget; +import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.util.StringUtils; /** @@ -52,21 +54,27 @@ public class ChannelAdapterParser implements BeanDefinitionParser { public BeanDefinition parse(Element element, ParserContext parserContext) { - RootBeanDefinition adapterDef = null; - if (this.isInbound) { - adapterDef = new RootBeanDefinition(InboundMethodInvokingChannelAdapter.class); - } - else { - adapterDef = new RootBeanDefinition(OutboundMethodInvokingChannelAdapter.class); - } - adapterDef.setSource(parserContext.extractSource(element)); String ref = element.getAttribute(REF_ATTRIBUTE); String method = element.getAttribute(METHOD_ATTRIBUTE); if (!StringUtils.hasText(ref) || !StringUtils.hasText(method)) { throw new MessagingConfigurationException("'ref' and 'method' are both required"); } - adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref)); - adapterDef.getPropertyValues().addPropertyValue("method", method); + RootBeanDefinition adapterDef = null; + RootBeanDefinition invokerDef = null; + if (this.isInbound) { + adapterDef = new RootBeanDefinition(PollingSourceAdapter.class); + invokerDef = new RootBeanDefinition(MethodInvokingSource.class); + } + else { + adapterDef = new RootBeanDefinition(DefaultTargetAdapter.class); + invokerDef = new RootBeanDefinition(MethodInvokingTarget.class); + } + invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref)); + invokerDef.getPropertyValues().addPropertyValue("method", method); + String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef); + parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName)); + adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName)); + adapterDef.setSource(parserContext.extractSource(element)); String beanName = element.getAttribute(ID_ATTRIBUTE); if (!StringUtils.hasText(beanName)) { beanName = parserContext.getReaderContext().generateBeanName(adapterDef); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ComponentConfigurer.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ComponentConfigurer.java deleted file mode 100644 index 7e26935df7..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ComponentConfigurer.java +++ /dev/null @@ -1,85 +0,0 @@ -/* - * Copyright 2002-2007 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.config; - -import org.springframework.beans.factory.config.RuntimeBeanReference; -import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.beans.factory.support.BeanNameGenerator; -import org.springframework.beans.factory.support.DefaultBeanNameGenerator; -import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.integration.endpoint.GenericMessageEndpoint; -import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter; -import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter; -import org.springframework.integration.handler.DefaultMessageHandlerAdapter; -import org.springframework.util.Assert; - -/** - * Factory for creating integration component bean definitions. - * - * @author Mark Fisher - */ -public class ComponentConfigurer { - - private BeanDefinitionRegistry registry; - - private BeanNameGenerator beanNameGenerator; - - - public ComponentConfigurer(BeanDefinitionRegistry registry, BeanNameGenerator beanNameGenerator) { - Assert.notNull(registry, "registry must not be null"); - this.registry = registry; - this.beanNameGenerator = (beanNameGenerator != null ? beanNameGenerator : new DefaultBeanNameGenerator()); - } - - public String serviceActivator(String inputChannel, String outputChannel, String objectRef, String method) { - RootBeanDefinition endpointDef = new RootBeanDefinition(GenericMessageEndpoint.class); - RootBeanDefinition adapterDef = new RootBeanDefinition(DefaultMessageHandlerAdapter.class); - adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef)); - adapterDef.getPropertyValues().addPropertyValue("methodName", method); - String adapterName = beanNameGenerator.generateBeanName(adapterDef, this.registry); - this.registry.registerBeanDefinition(adapterName, adapterDef); - endpointDef.getPropertyValues().addPropertyValue("handler", new RuntimeBeanReference(adapterName)); - if (inputChannel != null) { - endpointDef.getPropertyValues().addPropertyValue("inputChannelName", inputChannel); - } - if (outputChannel != null) { - endpointDef.getPropertyValues().addPropertyValue("defaultOutputChannelName", outputChannel); - } - String endpointName = beanNameGenerator.generateBeanName(endpointDef, this.registry); - this.registry.registerBeanDefinition(endpointName, endpointDef); - return endpointName; - } - - public String inboundChannelAdapter(String objectRef, String method) { - RootBeanDefinition bd = new RootBeanDefinition(InboundMethodInvokingChannelAdapter.class); - bd.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef)); - bd.getPropertyValues().addPropertyValue("method", method); - String beanName = this.beanNameGenerator.generateBeanName(bd, this.registry); - this.registry.registerBeanDefinition(beanName, bd); - return beanName; - } - - public String outboundChannelAdapter(String objectRef, String method) { - RootBeanDefinition bd = new RootBeanDefinition(OutboundMethodInvokingChannelAdapter.class); - bd.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef)); - bd.getPropertyValues().addPropertyValue("method", method); - String beanName = this.beanNameGenerator.generateBeanName(bd, this.registry); - this.registry.registerBeanDefinition(beanName, bd); - return beanName; - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 23730f74b1..28f8b55c64 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -33,6 +33,8 @@ import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.core.OrderComparator; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.MessagingConfigurationException; +import org.springframework.integration.adapter.MethodInvokingSource; +import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.annotation.DefaultOutput; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; @@ -42,8 +44,9 @@ import org.springframework.integration.annotation.Splitter; import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistryAware; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; -import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter; import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; @@ -124,14 +127,18 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { Annotation annotation = AnnotationUtils.getAnnotation(method, Polled.class); if (annotation != null) { - InboundMethodInvokingChannelAdapter adapter = new InboundMethodInvokingChannelAdapter(); - adapter.setObject(bean); - adapter.setMethod(method.getName()); - adapter.afterPropertiesSet(); - String channelName = beanName + "-inputChannel"; - messageBus.registerChannel(channelName, adapter); - endpoint.setInputChannelName(channelName); int period = ((Polled) annotation).period(); + MethodInvokingSource source = new MethodInvokingSource(); + source.setObject(bean); + source.setMethod(method.getName()); + PollingSourceAdapter adapter = new PollingSourceAdapter(source); + MessageChannel channel = new PointToPointChannel(); + adapter.setChannel(channel); + adapter.setPeriod(period); + String channelName = beanName + "-inputChannel"; + messageBus.registerChannel(channelName, channel); + messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter); + endpoint.setInputChannelName(channelName); endpoint.getConsumerPolicy().setPeriod(period); if (period > 0) { endpoint.getConsumerPolicy().setConcurrency(1); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/InboundMethodInvokingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/InboundMethodInvokingChannelAdapter.java deleted file mode 100644 index 361323b407..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/InboundMethodInvokingChannelAdapter.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * Copyright 2002-2007 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 org.springframework.integration.MessagingConfigurationException; -import org.springframework.util.Assert; - -/** - * An inbound channel adapter for invoking a no-argument method and receiving - * its return value. - * - * @author Mark Fisher - */ -public class InboundMethodInvokingChannelAdapter extends AbstractInboundChannelAdapter { - - private T object; - - private String method; - - private SimpleMethodInvoker invoker; - - - public void setObject(T object) { - Assert.notNull(object, "'object' must not be null"); - this.object = object; - } - - public void setMethod(String method) { - Assert.notNull(method, "'method' must not be null"); - this.method = method; - } - - @Override - public void initialize() { - this.invoker = new SimpleMethodInvoker(this.object, this.method); - this.invoker.setMethodValidator(new MessageReceivingMethodValidator()); - } - - @Override - protected Object doReceiveObject() { - return this.invoker.invokeMethod(new Object[] {}); - } - - - public static class MessageReceivingMethodValidator implements MethodValidator { - - public void validate(Method method) { - if (method.getReturnType().equals(void.class)) { - throw new MessagingConfigurationException( - "Inbound channel adapter requires a non-void returning method."); - } - } - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 838600b595..c326a72713 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -18,14 +18,14 @@ package org.springframework.integration.endpoint; import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.ChannelRegistry; -import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageReceiver; /** * Base interface for message endpoints. * * @author Mark Fisher */ -public interface MessageEndpoint { +public interface MessageEndpoint extends MessageReceiver { void setInputChannelName(String inputChannelName); @@ -39,6 +39,4 @@ public interface MessageEndpoint { void setChannelRegistry(ChannelRegistry channelRegistry); - void messageReceived(Message message); - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java new file mode 100644 index 0000000000..d58ec6546e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/MessageReceiver.java @@ -0,0 +1,29 @@ +/* + * Copyright 2002-2007 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.message; + +/** + * The primary callback interface for any component capable of receiving + * messages. This includes message endpoints as well as target adapters. + * + * @author Mark Fisher + */ +public interface MessageReceiver { + + void messageReceived(Message message); + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/AdapterBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/AdapterBusTests.java similarity index 96% rename from spring-integration-core/src/test/java/org/springframework/integration/endpoint/AdapterBusTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/adapter/AdapterBusTests.java index 67ac4519f8..895859a0c3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/AdapterBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/AdapterBusTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.endpoint; +package org.springframework.integration.adapter; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java new file mode 100644 index 0000000000..ce5e7ba86b --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java @@ -0,0 +1,68 @@ +/* + * Copyright 2002-2007 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.adapter; + +import static org.junit.Assert.assertTrue; + +import org.junit.Test; + +import org.springframework.integration.MessageDeliveryException; + +/** + * @author Mark Fisher + */ +public class MethodInvokingTargetTests { + + @Test + public void testValidMethod() { + MethodInvokingTarget target = new MethodInvokingTarget(); + target.setObject(new TestSink()); + target.setMethod("validMethod"); + target.afterPropertiesSet(); + boolean result = target.send("test"); + assertTrue(result); + } + + @Test(expected=MessageDeliveryException.class) + public void testInvalidMethodWithNoArgs() { + MethodInvokingTarget target = new MethodInvokingTarget(); + target.setObject(new TestSink()); + target.setMethod("invalidMethodWithNoArgs"); + target.afterPropertiesSet(); + target.send("test"); + } + + @Test + public void testValidMethodWithIgnoredReturnValue() { + MethodInvokingTarget target = new MethodInvokingTarget(); + target.setObject(new TestSink()); + target.setMethod("validMethodWithIgnoredReturnValue"); + target.afterPropertiesSet(); + boolean result = target.send("test"); + assertTrue(result); + } + + @Test(expected=MessageDeliveryException.class) + public void testNoMatchingMethodName() { + MethodInvokingTarget target = new MethodInvokingTarget(); + target.setObject(new TestSink()); + target.setMethod("noSuchMethod"); + target.afterPropertiesSet(); + target.send("test"); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java index c0a37c98a8..bb7fe6b283 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java @@ -43,7 +43,7 @@ public class PollingSourceAdapterTests { PollingSourceAdapter adapter = new PollingSourceAdapter(source); adapter.setChannel(channel); adapter.setPeriod(100); - adapter.receiveAndDispatch(); + adapter.dispatch(); Message message = channel.receive(); assertNotNull("message should not be null", message); assertEquals("testing.1", message.getPayload()); @@ -57,14 +57,14 @@ public class PollingSourceAdapterTests { adapter.setChannel(channel); adapter.setPeriod(500); adapter.setSendTimeout(10); - adapter.receiveAndDispatch(); - adapter.receiveAndDispatch(); + adapter.dispatch(); + adapter.dispatch(); Message message1 = channel.receive(); assertNotNull("message should not be null", message1); assertEquals("testing.1", message1.getPayload()); Message message2 = channel.receive(0); assertNull("second message should be null", message2); - adapter.receiveAndDispatch(); + adapter.dispatch(); Message message3 = channel.receive(0); assertNotNull("third message should not be null", message3); assertEquals("testing.3", message3.getPayload()); @@ -78,7 +78,7 @@ public class PollingSourceAdapterTests { adapter.setChannel(channel); adapter.setPeriod(1000); adapter.setMaxMessagesPerTask(5); - adapter.receiveAndDispatch(); + adapter.dispatch(); Message message1 = channel.receive(0); assertNotNull("message should not be null", message1); assertEquals("testing.1", message1.getPayload()); @@ -100,7 +100,7 @@ public class PollingSourceAdapterTests { adapter.setChannel(channel); adapter.setPeriod(1000); adapter.setMaxMessagesPerTask(2); - adapter.receiveAndDispatch(); + adapter.dispatch(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestSink.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java similarity index 95% rename from spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestSink.java rename to spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java index 2d30faf37b..e7953172d0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestSink.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.endpoint; +package org.springframework.integration.adapter; /** * @author Mark Fisher diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestSource.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSource.java similarity index 95% rename from spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestSource.java rename to spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSource.java index f027592a88..da6cafc459 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/TestSource.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSource.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.endpoint; +package org.springframework.integration.adapter; /** * @author Mark Fisher diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterBusTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterBusTests.xml new file mode 100644 index 0000000000..0da1a1ccd5 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterBusTests.xml @@ -0,0 +1,47 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java index 45488c5865..118bf2109a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/UnicastMessageDispatcherTests.java @@ -59,9 +59,9 @@ public class UnicastMessageDispatcherTests { channel.send(new StringMessage(1, "test")); MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); - dispatcher.addEndpointExecutor(new EndpointExecutor(endpoint1, 1, 1)); - dispatcher.addEndpointExecutor(new EndpointExecutor(endpoint2, 1, 1)); - dispatcher.receiveAndDispatch(); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint1, 1, 1)); + dispatcher.addExecutor(new MessageReceivingExecutor(endpoint2, 1, 1)); + dispatcher.dispatch(); latch.await(500, TimeUnit.MILLISECONDS); assertTrue("exactly one endpoint should have received message", endpoint1Received.get() ^ endpoint2Received.get()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ComponentConfigurerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ComponentConfigurerTests.java deleted file mode 100644 index 87bdd2ceab..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ComponentConfigurerTests.java +++ /dev/null @@ -1,100 +0,0 @@ -/* - * Copyright 2002-2007 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.config; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; - -import org.junit.Test; - -import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.context.support.GenericApplicationContext; -import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.PointToPointChannel; -import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.GenericMessage; - -/** - * @author Mark Fisher - */ -public class ComponentConfigurerTests { - - @Test - public void testServiceActivator() { - GenericApplicationContext context = new GenericApplicationContext(); - ComponentConfigurer cr = new ComponentConfigurer(context, null); - context.registerBeanDefinition("tester", new RootBeanDefinition(Tester.class)); - context.registerBeanDefinition("out", new RootBeanDefinition(PointToPointChannel.class)); - context.registerBeanDefinition("bus", new RootBeanDefinition(MessageBus.class)); - String name = cr.serviceActivator(null, "out", "tester", "test"); - context.refresh(); - MessageEndpoint endpoint = (MessageEndpoint) context.getBean(name); - endpoint.messageReceived(new GenericMessage(1, "world")); - MessageChannel channel = (MessageChannel) context.getBean("out"); - assertEquals("hello world", channel.receive().getPayload()); - } - - @Test - public void testInboundChannelAdapter() { - GenericApplicationContext context = new GenericApplicationContext(); - ComponentConfigurer cr = new ComponentConfigurer(context, null); - context.registerBeanDefinition("tester", new RootBeanDefinition(Tester.class)); - String adapter = cr.inboundChannelAdapter("tester", "foo"); - MessageChannel channel = (MessageChannel) context.getBean(adapter); - Message message = channel.receive(); - assertEquals("bar", message.getPayload()); - } - - @Test - public void testOutboundChannelAdapter() { - GenericApplicationContext context = new GenericApplicationContext(); - ComponentConfigurer cr = new ComponentConfigurer(context, null); - context.registerBeanDefinition("tester", new RootBeanDefinition(Tester.class)); - String adapter = cr.outboundChannelAdapter("tester", "store"); - Tester tester = (Tester) context.getBean("tester"); - assertNull(tester.getStoredValue()); - MessageChannel channel = (MessageChannel) context.getBean(adapter); - boolean result = channel.send(new GenericMessage(1, "foo")); - assertTrue(result); - assertEquals("foo", tester.getStoredValue()); - } - - - public static class Tester { - - private String storedValue; - - public String test(String s) { - return "hello " + s; - } - - public String foo() { - return "bar"; - } - - public void store(String s) { - this.storedValue = s; - } - - public String getStoredValue() { - return this.storedValue; - } - } -} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/InboundChannelAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/InboundChannelAdapterTests.java deleted file mode 100644 index b886b2e898..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/InboundChannelAdapterTests.java +++ /dev/null @@ -1,73 +0,0 @@ -/* - * Copyright 2002-2007 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 static org.junit.Assert.assertEquals; - -import org.junit.Test; - -import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.endpoint.InboundMethodInvokingChannelAdapter; -import org.springframework.integration.message.Message; - -/** - * @author Mark Fisher - */ -public class InboundChannelAdapterTests { - - @Test - public void testValidMethod() { - InboundMethodInvokingChannelAdapter adapter = - new InboundMethodInvokingChannelAdapter(); - adapter.setObject(new TestSource()); - adapter.setMethod("validMethod"); - adapter.afterPropertiesSet(); - Message message = adapter.receive(); - assertEquals("valid", message.getPayload()); - } - - @Test(expected=MessageHandlingException.class) - public void testInvalidMethodWithArg() { - InboundMethodInvokingChannelAdapter adapter = - new InboundMethodInvokingChannelAdapter(); - adapter.setObject(new TestSource()); - adapter.setMethod("invalidMethodWithArg"); - adapter.afterPropertiesSet(); - adapter.receive(); - } - - @Test(expected=MessageHandlingException.class) - public void testInvalidMethodWithNoReturnValue() { - InboundMethodInvokingChannelAdapter adapter = - new InboundMethodInvokingChannelAdapter(); - adapter.setObject(new TestSource()); - adapter.setMethod("invalidMethodWithNoReturnValue"); - adapter.afterPropertiesSet(); - adapter.receive(); - } - - @Test(expected=MessageHandlingException.class) - public void testNoMatchingMethodName() { - InboundMethodInvokingChannelAdapter adapter = - new InboundMethodInvokingChannelAdapter(); - adapter.setObject(new TestSource()); - adapter.setMethod("noSuchMethod"); - adapter.afterPropertiesSet(); - adapter.receive(); - } - -} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/OutboundChannelAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/OutboundChannelAdapterTests.java deleted file mode 100644 index 4d7679f904..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/OutboundChannelAdapterTests.java +++ /dev/null @@ -1,74 +0,0 @@ -/* - * Copyright 2002-2007 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 static org.junit.Assert.assertTrue; - -import org.junit.Test; - -import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.endpoint.OutboundMethodInvokingChannelAdapter; -import org.springframework.integration.message.StringMessage; - -/** - * @author Mark Fisher - */ -public class OutboundChannelAdapterTests { - - @Test - public void testValidMethod() { - OutboundMethodInvokingChannelAdapter adapter = - new OutboundMethodInvokingChannelAdapter(); - adapter.setObject(new TestSink()); - adapter.setMethod("validMethod"); - adapter.afterPropertiesSet(); - boolean result = adapter.send(new StringMessage(1, "test")); - assertTrue(result); - } - - @Test(expected=MessageHandlingException.class) - public void testInvalidMethodWithNoArgs() { - OutboundMethodInvokingChannelAdapter adapter = - new OutboundMethodInvokingChannelAdapter(); - adapter.setObject(new TestSink()); - adapter.setMethod("invalidMethodWithNoArgs"); - adapter.afterPropertiesSet(); - adapter.send(new StringMessage(1, "test")); - } - - @Test - public void testValidMethodWithIgnoredReturnValue() { - OutboundMethodInvokingChannelAdapter adapter = - new OutboundMethodInvokingChannelAdapter(); - adapter.setObject(new TestSink()); - adapter.setMethod("validMethodWithIgnoredReturnValue"); - adapter.afterPropertiesSet(); - boolean result = adapter.send(new StringMessage(1, "test")); - assertTrue(result); - } - - @Test(expected=MessageHandlingException.class) - public void testNoMatchingMethodName() { - OutboundMethodInvokingChannelAdapter adapter = - new OutboundMethodInvokingChannelAdapter(); - adapter.setObject(new TestSink()); - adapter.setMethod("noSuchMethod"); - adapter.afterPropertiesSet(); - adapter.send(new StringMessage(1, "test")); - } - -} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/adapterBusTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/adapterBusTests.xml deleted file mode 100644 index 5103ef0488..0000000000 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/adapterBusTests.xml +++ /dev/null @@ -1,45 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -