diff --git a/spring-eai-core/src/main/java/org/springframework/integration/MessageSource.java b/spring-eai-core/src/main/java/org/springframework/integration/MessageSource.java deleted file mode 100644 index 97b726b354..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/MessageSource.java +++ /dev/null @@ -1,48 +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; - -import org.springframework.integration.message.Message; - -/** - * Base interface for any component that enables message reception. - * - * @author Mark Fisher - */ -public interface MessageSource { - - /** - * Receive a message, blocking indefinitely if necessary. - * - * @return the next available {@link Message} or - * null if interrupted - */ - Message receive(); - - /** - * Receive a message, blocking until either a message is - * available or the specified timeout period elapses. - * - * @param timeout the timeout in milliseconds - * - * @return the next available {@link Message} or - * null if the specified timeout period - * elapses or the message reception is interrupted - */ - Message receive(long timeout); - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/MessageTarget.java b/spring-eai-core/src/main/java/org/springframework/integration/MessageTarget.java deleted file mode 100644 index 26c37c876f..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/MessageTarget.java +++ /dev/null @@ -1,51 +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; - -import org.springframework.integration.message.Message; - -/** - * Base interface for any component that enables message sending. - * - * @author Mark Fisher - */ -public interface MessageTarget { - - /** - * Send a message, blocking indefinitely if necessary. - * - * @param message the {@link Message} to send - * - * @return true if the message is sent - * successfully, false if interrupted - */ - boolean send(Message message); - - /** - * Send a message, blocking until either the message is - * accepted or the specified timeout period elapses. - * - * @param message the {@link Message} to send - * @param timeout the timeout in milliseconds - * - * @return true if the message is sent - * successfully, false if the specified - * timeout period elapses or the send is interrupted - */ - boolean send(Message message, long timeout); - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/MessagingConfigurationException.java b/spring-eai-core/src/main/java/org/springframework/integration/MessagingConfigurationException.java new file mode 100644 index 0000000000..da834b6bf0 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/MessagingConfigurationException.java @@ -0,0 +1,35 @@ +/* + * 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; + +/** + * Exception that indicates an incorrectly configured messaging component. + * + * @author Mark Fisher + */ +@SuppressWarnings("serial") +public class MessagingConfigurationException extends MessagingException { + + public MessagingConfigurationException(String message) { + super(message); + } + + public MessagingConfigurationException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 643600c24a..b40c2b48de 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -40,7 +40,7 @@ import org.springframework.util.Assert; /** * The messaging bus. Serves as a registry for channels and endpoints, manages their lifecycle, - * and all subscriptions. + * and activates subscriptions. * * @author Mark Fisher */ @@ -60,8 +60,6 @@ public class MessageBus implements ChannelResolver, ApplicationContextAware, Lif private boolean autoCreateChannels; - private ApplicationContext applicationContext; - private boolean running; private Object lifecycleMonitor = new Object(); @@ -69,15 +67,15 @@ public class MessageBus implements ChannelResolver, ApplicationContextAware, Lif public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { Assert.notNull(applicationContext, "applicationContext must not be null"); - this.applicationContext = applicationContext; - this.registerChannelsFromContext(); - this.registerEndpointsFromContext(); + this.registerChannels(applicationContext); + this.registerEndpoints(applicationContext); + this.activateSubscriptions(applicationContext); } @SuppressWarnings("unchecked") - private void registerChannelsFromContext() { - Map channelBeans = (Map) this.applicationContext - .getBeansOfType(MessageChannel.class); + private void registerChannels(ApplicationContext context) { + Map channelBeans = + (Map) context.getBeansOfType(MessageChannel.class); for (Map.Entry entry : channelBeans.entrySet()) { this.registerChannel(entry.getKey(), entry.getValue()); if (logger.isInfoEnabled()) { @@ -87,9 +85,9 @@ public class MessageBus implements ChannelResolver, ApplicationContextAware, Lif } @SuppressWarnings("unchecked") - private void registerEndpointsFromContext() { - Map endpointBeans = (Map) this.applicationContext - .getBeansOfType(MessageEndpoint.class); + private void registerEndpoints(ApplicationContext context) { + Map endpointBeans = + (Map) context.getBeansOfType(MessageEndpoint.class); for (Map.Entry entry : endpointBeans.entrySet()) { this.registerEndpoint(entry.getKey(), entry.getValue()); if (logger.isInfoEnabled()) { @@ -98,6 +96,19 @@ public class MessageBus implements ChannelResolver, ApplicationContextAware, Lif } } + @SuppressWarnings("unchecked") + private void activateSubscriptions(ApplicationContext context) { + Map subscriptionBeans = + (Map) context.getBeansOfType(Subscription.class); + for (Subscription subscription : subscriptionBeans.values()) { + this.activateSubscription(subscription); + if (logger.isInfoEnabled()) { + logger.info("activated subscription to channel '" + subscription.getChannel() + + "' for endpoint '" + subscription.getEndpoint() + "'"); + } + } + } + public void initialize() { this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherTasks.size() > 0 ? this.dispatcherTasks.size() : 1); } @@ -112,9 +123,13 @@ public class MessageBus implements ChannelResolver, ApplicationContextAware, Lif public void registerEndpoint(String name, MessageEndpoint endpoint) { this.endpoints.put(name, endpoint); + endpoint.setChannelResolver(this); } - public void activateSubscription(String channelName, String endpointName, ConsumerPolicy policy) { + public void activateSubscription(Subscription subscription) { + String channelName = subscription.getChannel(); + String endpointName = subscription.getEndpoint(); + ConsumerPolicy policy = subscription.getPolicy(); MessageChannel channel = this.channels.get(channelName); if (channel == null) { if (this.autoCreateChannels == false) { diff --git a/spring-eai-core/src/main/java/org/springframework/integration/bus/Subscription.java b/spring-eai-core/src/main/java/org/springframework/integration/bus/Subscription.java new file mode 100644 index 0000000000..e11e2977c1 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/bus/Subscription.java @@ -0,0 +1,57 @@ +/* + * 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.bus; + +/** + * Configuration metadata for activating a subscription. + * + * @author Mark Fisher + */ +public class Subscription { + + private String channel; + + private String endpoint; + + private ConsumerPolicy policy = new ConsumerPolicy(); + + + public String getChannel() { + return this.channel; + } + + public void setChannel(String channel) { + this.channel = channel; + } + + public String getEndpoint() { + return this.endpoint; + } + + public void setEndpoint(String endpoint) { + this.endpoint = endpoint; + } + + public ConsumerPolicy getPolicy() { + return this.policy; + } + + public void setPolicy(ConsumerPolicy policy) { + this.policy = policy; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index 651b469ee0..e415c0683c 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -16,17 +16,56 @@ package org.springframework.integration.channel; -import org.springframework.integration.MessageSource; -import org.springframework.integration.MessageTarget; +import org.springframework.integration.message.Message; /** - * Base channel interface that combines the definitions of {@link MessageSource} - * for message reception and {@link MessageTarget} for message sending. + * Base channel interface defining common behavior for message reception and sending. * * @author Mark Fisher - * @see MessageSource - * @see MessageTarget */ -public interface MessageChannel extends MessageSource, MessageTarget { +public interface MessageChannel { + + /** + * Send a message, blocking indefinitely if necessary. + * + * @param message the {@link Message} to send + * + * @return true if the message is sent + * successfully, false if interrupted + */ + boolean send(Message message); + + /** + * Send a message, blocking until either the message is + * accepted or the specified timeout period elapses. + * + * @param message the {@link Message} to send + * @param timeout the timeout in milliseconds + * + * @return true if the message is sent + * successfully, false if the specified + * timeout period elapses or the send is interrupted + */ + boolean send(Message message, long timeout); + + /** + * Receive a message, blocking indefinitely if necessary. + * + * @return the next available {@link Message} or + * null if interrupted + */ + Message receive(); + + /** + * Receive a message, blocking until either a message is + * available or the specified timeout period elapses. + * + * @param timeout the timeout in milliseconds + * + * @return the next available {@link Message} or + * null if the specified timeout period + * elapses or the message reception is interrupted + */ + Message receive(long timeout); } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractChannelAdapter.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractChannelAdapter.java new file mode 100644 index 0000000000..ce069de5ad --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractChannelAdapter.java @@ -0,0 +1,162 @@ +/* + * 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.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.MessageHandlingException; +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; + +/** + * Convenience base class for channel adapters. + * + * @author Mark Fisher + */ +public abstract class AbstractChannelAdapter implements MessageChannel, InitializingBean { + + protected Log logger = LogFactory.getLog(this.getClass()); + + private MessageMapper mapper = new SimplePayloadMessageMapper(); + + private volatile boolean initialized; + + + public final void afterPropertiesSet() { + this.initialize(); + this.initialized = true; + } + + public void setMapper(MessageMapper mapper) { + Assert.notNull(mapper, "'mapper' must not be null"); + this.mapper = mapper; + } + + protected MessageMapper getMapper() { + return this.mapper; + } + + public boolean send(Message message) { + if (!this.initialized) { + throw new MessageHandlingException("adapter not initialized"); + } + try { + Object source = this.getMapper().fromMessage(message); + return this.sendObject(source); + } + catch (Exception e) { + throw new MessageHandlingException("Failed to send message to target", e); + } + } + + public boolean send(final Message message, long timeout) { + if (!this.initialized) { + throw new MessageHandlingException("adapter not initialized"); + } + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future result = executor.submit(new Callable() { + public Boolean call() throws Exception { + return send(message); + } + }); + try { + result.get(timeout, TimeUnit.MILLISECONDS); + if (result.isDone()) { + return result.get(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + catch (TimeoutException e) { + return false; + } + catch (ExecutionException e) { + throw new MessageHandlingException("Exception occurred in message source", e); + } + result.cancel(true); + return false; + } + + public Message receive() { + if (!this.initialized) { + throw new MessageHandlingException("adapter not initialized"); + } + try { + Object result = this.receiveObject(); + if (result != null) { + return this.getMapper().toMessage(result); + } + } + catch (Exception e) { + throw new MessageHandlingException("Failed to receive message from source", e); + } + return null; + } + + public Message receive(long timeout) { + if (!this.initialized) { + throw new MessageHandlingException("adapter not initialized"); + } + ExecutorService executor = Executors.newSingleThreadExecutor(); + Future result = executor.submit(new Callable() { + public Message call() throws Exception { + return receive(); + } + }); + try { + result.get(timeout, TimeUnit.MILLISECONDS); + if (result.isDone()) { + return result.get(); + } + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } + catch (TimeoutException e) { + return null; + } + catch (ExecutionException e) { + throw new MessageHandlingException("Exception occurred in message source", e); + } + result.cancel(true); + return null; + } + + protected void initialize() { + } + + protected abstract boolean sendObject(Object object) throws Exception; + + protected abstract Object receiveObject() throws Exception; + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractInboundChannelAdapter.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractInboundChannelAdapter.java new file mode 100644 index 0000000000..0c81fbc2f8 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractInboundChannelAdapter.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.endpoint; + +/** + * Convenience base class for inbound channel adapters. + * + * @author Mark Fisher + */ +public abstract class AbstractInboundChannelAdapter extends AbstractChannelAdapter { + + protected boolean sendObject(Object object) throws Exception { + return false; + } + + protected Object receiveObject() throws Exception { + return this.doReceiveObject(); + } + + protected abstract Object doReceiveObject() throws Exception; + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractOutboundChannelAdapter.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractOutboundChannelAdapter.java new file mode 100644 index 0000000000..6cdf2449c9 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractOutboundChannelAdapter.java @@ -0,0 +1,38 @@ +/* + * 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; + +/** + * A convenience base class for outbound channel adapters. + * + * @author Mark Fisher + */ +public abstract class AbstractOutboundChannelAdapter extends AbstractChannelAdapter { + + @Override + protected Object receiveObject() throws Exception { + return null; + } + + @Override + protected boolean sendObject(Object object) throws Exception { + return this.doSendObject(object); + } + + protected abstract boolean doSendObject(Object object) throws Exception; + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingMessageSource.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingMessageSource.java deleted file mode 100644 index 66c19ada67..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingMessageSource.java +++ /dev/null @@ -1,66 +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 org.springframework.integration.MessageSource; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageMapper; -import org.springframework.integration.message.SimplePayloadMessageMapper; -import org.springframework.util.Assert; - -/** - * A {@link MessageSource} adapter for any source that can be polled for - * objects. - * - * @author Mark Fisher - */ -public abstract class AbstractPollingMessageSource implements MessageSource { - - private MessageMapper mapper = new SimplePayloadMessageMapper(); - - - public void setMapper(MessageMapper mapper) { - Assert.notNull(mapper, "mapper must not be null"); - this.mapper = mapper; - } - - public Message receive() { - return this.receive(-1); - } - - public Message receive(long timeout) { - long start = System.currentTimeMillis(); - while (timeout <= 0 || System.currentTimeMillis() - start < timeout) { - Object o = this.pollForObject(); - if (o != null) { - return this.mapper.toMessage(o); - } - if (timeout == 0) { - return null; - } - } - return null; - } - - - /** - * Method for subclasses to implement. Returns an object to be mapped to a - * {@link Message} by the message mapper. - */ - protected abstract Object pollForObject(); - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractPrefetchingMessageSource.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractPrefetchingMessageSource.java deleted file mode 100644 index f8b157358b..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/AbstractPrefetchingMessageSource.java +++ /dev/null @@ -1,68 +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.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; - -import org.springframework.integration.MessageSource; -import org.springframework.integration.message.Message; - -/** - * A {@link MessageSource} adapter for any source that can be polled for - * objects. This version allows for pre-fetching multiple results so that - * subsequent calls to {@link #pollForObject()} may be more efficient. - * - * @author Mark Fisher - */ -public abstract class AbstractPrefetchingMessageSource extends AbstractPollingMessageSource { - - private BlockingQueue queue = new LinkedBlockingQueue(); - - - public Object pollForObject() { - Object o = queue.poll(); - if (o == null) { - this.prefetch(); - o = queue.poll(); - } - return o; - } - - private void prefetch() { - Object[] results = this.pollForObjects(); - if (results != null) { - try { - for (Object o : results) { - queue.put(o); - } - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; - } - } - } - - - /** - * Method for subclasses to implement. Returns objects to be mapped to - * {@link Message Messages} by the message mapper. - */ - protected abstract Object[] pollForObjects(); - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/ArgumentListPreparer.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/ArgumentListPreparer.java new file mode 100644 index 0000000000..41419af973 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/ArgumentListPreparer.java @@ -0,0 +1,28 @@ +/* + * 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; + +/** + * A strategy for preparing an argument list from a single source object. + * + * @author Mark Fisher + */ +public interface ArgumentListPreparer { + + Object[] prepare(Object source); + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java index fed3b75c7e..0de86946ac 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java @@ -17,9 +17,10 @@ package org.springframework.integration.endpoint; import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.MessageSource; -import org.springframework.integration.MessageTarget; +import org.springframework.integration.MessagingConfigurationException; +import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; @@ -38,34 +39,44 @@ import org.springframework.integration.message.Message; */ public class GenericMessageEndpoint implements MessageEndpoint { - private MessageSource source; + private String inputChannelName; - private MessageTarget target; + private String defaultOutputChannelName; private MessageHandler handler; private ChannelResolver channelResolver; + private ConsumerPolicy consumerPolicy; + /** - * Set the source from which this endpoint receives messages. + * Set the name of the channel from which this endpoint receives messages. */ - public void setSource(MessageSource source) { - this.source = source; + public void setInputChannelName(String inputChannelName) { + this.inputChannelName = inputChannelName; } /** - * Return the source from which this endpoint receives messages. + * Return the name of the channel from which this endpoint receives messages. */ - public MessageSource getSource() { - return this.source; + public String getInputChannelName() { + return this.inputChannelName; + } + + public void setConsumerPolicy(ConsumerPolicy consumerPolicy) { + this.consumerPolicy = consumerPolicy; + } + + public ConsumerPolicy getConsumerPolicy() { + return this.consumerPolicy; } /** - * Set the target to which this endpoint can send messages. + * Set the name of the channel to which this endpoint can send reply messages by default. */ - public void setTarget(MessageTarget target) { - this.target = target; + public void setDefaultOutputChannelName(String defaultOutputChannelName) { + this.defaultOutputChannelName = defaultOutputChannelName; } /** @@ -76,40 +87,44 @@ public class GenericMessageEndpoint implements MessageEndpoint { } /** - * Set the channel resolver strategy to use when a message - * provides a 'replyChannelName'. + * Set the channel resolver to use for resolving channels by name. */ - public void setChannelResolver(final ChannelResolver channelResolver) { + public void setChannelResolver(ChannelResolver channelResolver) { this.channelResolver = channelResolver; } public void messageReceived(Message message) { if (this.handler == null) { - target.send(message); + if (this.defaultOutputChannelName == null) { + throw new MessagingConfigurationException( + "endpoint must have either a 'handler' or 'defaultOutputChannelName'"); + } + MessageChannel replyChannel = this.channelResolver.resolve(this.defaultOutputChannelName); + replyChannel.send(message); return; } Message replyMessage = handler.handle(message); if (replyMessage != null) { - MessageTarget replyTarget = resolveReplyTarget(message); - if (replyTarget == null) { - throw new MessageHandlingException("Unable to determine reply target for message. " - + "Provide a 'replyChannelName' in the message header or a 'target' " + MessageChannel replyChannel = this.resolveReplyChannel(message); + if (replyChannel == null) { + throw new MessageHandlingException("Unable to determine reply channel for message. " + + "Provide a 'replyChannelName' in the message header or a 'defaultReplyChannelName' " + "on the message endpoint."); } - replyTarget.send(replyMessage); + replyChannel.send(replyMessage); } } - private MessageTarget resolveReplyTarget(Message message) { - MessageTarget replyTo = null; - if (this.channelResolver != null) { - String replyChannelName = message.getHeader().getReplyChannelName(); - if (replyChannelName != null && replyChannelName.trim().length() > 0) { - replyTo = this.channelResolver.resolve(replyChannelName); - } + private MessageChannel resolveReplyChannel(Message message) { + if (this.channelResolver == null) { + return null; } - return (replyTo != null ? replyTo : target); + String replyChannelName = message.getHeader().getReplyChannelName(); + if (replyChannelName != null && replyChannelName.trim().length() > 0) { + return this.channelResolver.resolve(replyChannelName); + } + return this.channelResolver.resolve(this.defaultOutputChannelName); } } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/InboundMethodInvokingChannelAdapter.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/InboundMethodInvokingChannelAdapter.java new file mode 100644 index 0000000000..361323b407 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/InboundMethodInvokingChannelAdapter.java @@ -0,0 +1,71 @@ +/* + * 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-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index f7f24eb59d..66e6dff963 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -16,7 +16,6 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.MessageSource; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.message.Message; @@ -27,7 +26,9 @@ import org.springframework.integration.message.Message; */ public interface MessageEndpoint { - MessageSource getSource(); + void setInputChannelName(String inputChannelName); + + void setDefaultOutputChannelName(String defaultOutputChannelName); void setChannelResolver(ChannelResolver channelResolver); diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageHandlerAdapter.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageHandlerAdapter.java new file mode 100644 index 0000000000..f5c28fd211 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageHandlerAdapter.java @@ -0,0 +1,66 @@ +/* + * 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 org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageMapper; +import org.springframework.util.Assert; + +/** + * An implementation of {@link MessageHandler} that invokes the specified method + * on the provided target object. It then uses a {@link MessageMapper} strategy + * for converting the object to a {@link Message}. If the method has a non-null + * return value, a reply message will be generated by the mapper. + * + * @author Mark Fisher + */ +public class MessageHandlerAdapter implements MessageHandler, InitializingBean { + + private T object; + + private String method; + + private MessageMapper mapper; + + 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; + } + + public void afterPropertiesSet() { + this.invoker = new SimpleMethodInvoker(this.object, this.method); + } + + public Message handle(Message message) { + Object result = this.invoker.invokeMethod(this.mapper.fromMessage(message)); + if (result != null) { + return this.mapper.toMessage(result); + } + return null; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MethodValidator.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MethodValidator.java new file mode 100644 index 0000000000..8da2754fb9 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MethodValidator.java @@ -0,0 +1,33 @@ +/* + * 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; + +/** + * Interface for method validation. Implementations should throw an exception if + * the method is invalid for its purpose. + * + * @author Mark Fisher + */ +public interface MethodValidator { + + void validate(Method method) throws MessagingConfigurationException; + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/OutboundMethodInvokingChannelAdapter.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/OutboundMethodInvokingChannelAdapter.java new file mode 100644 index 0000000000..9c154fdac3 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/OutboundMethodInvokingChannelAdapter.java @@ -0,0 +1,75 @@ +/* + * 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 org.springframework.integration.message.MessageMapper; +import org.springframework.util.Assert; + +/** + * An outbound channel adapter for invoking the specified method on the provided + * object. Delegates to a {@link MessageMapper} for converting between objects + * and messages. An optional {@link ArgumentListPreparer} may also be provided. + * + * @author Mark Fisher + */ +public class OutboundMethodInvokingChannelAdapter extends AbstractOutboundChannelAdapter { + + 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; + } + + @Override + public void initialize() { + this.invoker = new SimpleMethodInvoker(this.object, this.method); + } + + @Override + public boolean doSendObject(Object object) throws Exception { + 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-eai-core/src/main/java/org/springframework/integration/endpoint/SimpleMethodInvoker.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/SimpleMethodInvoker.java new file mode 100644 index 0000000000..c791613a3e --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/SimpleMethodInvoker.java @@ -0,0 +1,73 @@ +/* + * 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.InvocationTargetException; + +import org.springframework.integration.MessageDeliveryException; +import org.springframework.util.Assert; +import org.springframework.util.MethodInvoker; +import org.springframework.util.ObjectUtils; + +/** + * A simple wrapper for {@link MethodInvoker}. + * + * @author Mark Fisher + */ +public class SimpleMethodInvoker { + + private T object; + + private String method; + + private MethodValidator methodValidator; + + + public SimpleMethodInvoker(T object, String method) { + Assert.notNull(object, "'object' must not be null"); + Assert.notNull(method, "'method' must not be null"); + this.object = object; + this.method = method; + } + + public void setMethodValidator(MethodValidator methodValidator) { + this.methodValidator = methodValidator; + } + + public Object invokeMethod(Object ... args) { + try { + MethodInvoker methodInvoker = new MethodInvoker(); + methodInvoker.setTargetObject(this.object); + methodInvoker.setTargetMethod(this.method); + methodInvoker.setArguments(args); + methodInvoker.prepare(); + if (this.methodValidator != null) { + this.methodValidator.validate(methodInvoker.getPreparedMethod()); + } + return methodInvoker.invoke(); + } + catch (InvocationTargetException e) { + throw new MessageDeliveryException( + "Method '" + this.method + "' threw exception", e.getTargetException()); + } + catch (Throwable e) { + throw new MessageDeliveryException("Failed to invoke method '" + this.method + + "' with arguments " + ObjectUtils.nullSafeToString(args), e); + } + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java index 9c0b37ed66..fe93dbbce5 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java @@ -39,13 +39,10 @@ import org.springframework.integration.message.Message; public class EventDrivenConsumerTests { @Test - public void stub() {} - - // TODO: make this a @Test public void testDynamicConcurrency() throws Exception { int messagesToSend = 200; int concurrency = 1; - int maxConcurrency = 100; + int maxConcurrency = 40; final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); final AtomicInteger maxActive = new AtomicInteger(0); @@ -54,8 +51,10 @@ public class EventDrivenConsumerTests { PointToPointChannel channel = new PointToPointChannel(); MessageEndpoint endpoint = new GenericMessageEndpoint() { public void messageReceived(Message message) { + System.out.println("[count:" + latch.getCount() + "] received: " + message.getPayload()); counter.incrementAndGet(); latch.countDown(); + try { Thread.sleep(3); } catch (InterruptedException e) {} activeSum.set(activeSum.addAndGet(bus.getActiveCountForEndpoint("testEndpoint"))); maxActive.set(Math.max(bus.getActiveCountForEndpoint("testEndpoint"), maxActive.get())); } @@ -69,30 +68,35 @@ public class EventDrivenConsumerTests { policy.setRejectionLimit(1); policy.setPeriod(0); policy.setReceiveTimeout(100); - bus.activateSubscription("testChannel", "testEndpoint", policy); + Subscription subscription = new Subscription(); + subscription.setChannel("testChannel"); + subscription.setEndpoint("testEndpoint"); + subscription.setPolicy(policy); + bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend - 110; i++) { channel.send(new DocumentMessage(1, "fast-1." + (i+1))); } int activeCountAfterFirstBurst = bus.getActiveCountForEndpoint("testEndpoint"); - System.out.println("after-first: " + activeCountAfterFirstBurst); + //System.out.println("after-first: " + activeCountAfterFirstBurst); for (int i = 0; i < 10; i++) { channel.send(new DocumentMessage(1, "slow-1." + (i+1))); - Thread.sleep(50); + Thread.sleep(10); } int activeCountAfterSlowDown = bus.getActiveCountForEndpoint("testEndpoint"); - System.out.println("after-slowdown: " + activeCountAfterSlowDown); + //System.out.println("after-slowdown: " + activeCountAfterSlowDown); for (int i = 0; i < 100; i++) { channel.send(new DocumentMessage(1, "fast-2." + (i+1))); } int activeCountAfterLastBurst = bus.getActiveCountForEndpoint("testEndpoint"); - System.out.println("after-last: " + activeCountAfterLastBurst); - latch.await(10, TimeUnit.SECONDS); + //System.out.println("after-last: " + activeCountAfterLastBurst); + latch.await(100, TimeUnit.SECONDS); int averageActive = activeSum.get() / messagesToSend; assertTrue(activeCountAfterSlowDown < activeCountAfterFirstBurst); assertTrue(activeCountAfterLastBurst > activeCountAfterSlowDown); assertEquals(messagesToSend, counter.get()); - assertEquals(maxConcurrency, maxActive.get()); + assertTrue(maxActive.get() > concurrency); + assertTrue(maxActive.get() <= maxConcurrency); assertTrue(averageActive > concurrency); assertTrue(averageActive < maxActive.get()); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java index ff548049b6..241ecbdc64 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; + import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PointToPointChannel; @@ -58,7 +59,11 @@ public class FixedDelayConsumerTests { policy.setMaxMessagesPerTask(1); policy.setFixedRate(true); policy.setPeriod(10); - bus.activateSubscription("testChannel", "testEndpoint", policy); + Subscription subscription = new Subscription(); + subscription.setChannel("testChannel"); + subscription.setEndpoint("testEndpoint"); + subscription.setPolicy(policy); + bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); @@ -88,7 +93,11 @@ public class FixedDelayConsumerTests { policy.setMaxMessagesPerTask(1); policy.setFixedRate(true); policy.setPeriod(10); - bus.activateSubscription("testChannel", "testEndpoint", policy); + Subscription subscription = new Subscription(); + subscription.setChannel("testChannel"); + subscription.setEndpoint("testEndpoint"); + subscription.setPolicy(policy); + bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java index 9a9cac604a..771800b2c9 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java @@ -56,7 +56,11 @@ public class FixedRateConsumerTests { ConsumerPolicy policy = new ConsumerPolicy(); policy.setFixedRate(true); policy.setPeriod(10); - bus.activateSubscription("testChannel", "testEndpoint", policy); + Subscription subscription = new Subscription(); + subscription.setChannel("testChannel"); + subscription.setEndpoint("testEndpoint"); + subscription.setPolicy(policy); + bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); @@ -86,14 +90,18 @@ public class FixedRateConsumerTests { policy.setMaxMessagesPerTask(1); policy.setFixedRate(true); policy.setPeriod(10); - bus.activateSubscription("testChannel", "testEndpoint", policy); + Subscription subscription = new Subscription(); + subscription.setChannel("testChannel"); + subscription.setEndpoint("testEndpoint"); + subscription.setPolicy(policy); + bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } latch.await(80, TimeUnit.MILLISECONDS); assertTrue(counter.get() < 10); - assertTrue(counter.get() > 7); + assertTrue("only " + counter.get() + " messages received", counter.get() > 7); } } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java similarity index 89% rename from spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java rename to spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index fafed69c0c..0a6a2aead9 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -14,20 +14,16 @@ * limitations under the License. */ -package org.springframework.integration; +package org.springframework.integration.bus; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; import org.junit.Test; -import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.PointToPointChannel; -import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; @@ -74,7 +70,11 @@ public class MessageBusTests { // TODO: add metadata for this MessageBus bus = (MessageBus) context.getBean("bus"); ConsumerPolicy policy = new ConsumerPolicy(); - bus.activateSubscription("sourceChannel", "endpoint", policy); + Subscription subscription = new Subscription(); + subscription.setChannel("sourceChannel"); + subscription.setEndpoint("endpoint"); + subscription.setPolicy(policy); + bus.activateSubscription(subscription); Message result = targetChannel.receive(10); assertEquals("test", result.getPayload()); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml b/spring-eai-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml similarity index 83% rename from spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml rename to spring-eai-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml index cade803fbf..ca1249bab7 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -11,8 +11,8 @@ - - + + diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java index ec7116c6e3..35f30c2398 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java @@ -23,6 +23,7 @@ import org.junit.Test; import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.bus.Subscription; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; @@ -45,15 +46,20 @@ public class GenericMessageEndpointTests { } }; GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); - endpoint.setSource(channel); + endpoint.setInputChannelName("testChannel"); endpoint.setHandler(handler); - endpoint.setTarget(replyChannel); + endpoint.setDefaultOutputChannelName("replyChannel"); MessageBus bus = new MessageBus(); bus.registerChannel("testChannel", channel); bus.registerEndpoint("testEndpoint", endpoint); + bus.registerChannel("replyChannel", replyChannel); ConsumerPolicy policy = new ConsumerPolicy(); policy.setPeriod(0); - bus.activateSubscription("testChannel", "testEndpoint", policy); + Subscription subscription = new Subscription(); + subscription.setChannel("testChannel"); + subscription.setEndpoint("testEndpoint"); + subscription.setPolicy(policy); + bus.activateSubscription(subscription); bus.start(); DocumentMessage testMessage = new DocumentMessage(1, "test"); channel.send(testMessage); @@ -80,15 +86,20 @@ public class GenericMessageEndpointTests { } }; GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); - endpoint.setSource(channel); + endpoint.setInputChannelName("testChannel"); endpoint.setHandler(handler); endpoint.setChannelResolver(channelResolver); MessageBus bus = new MessageBus(); bus.registerChannel("testChannel", channel); bus.registerEndpoint("testEndpoint", endpoint); + bus.registerChannel("replyChannel", replyChannel); ConsumerPolicy policy = new ConsumerPolicy(); policy.setPeriod(0); - bus.activateSubscription("testChannel", "testEndpoint", policy); + Subscription subscription = new Subscription(); + subscription.setChannel("testChannel"); + subscription.setEndpoint("testEndpoint"); + subscription.setPolicy(policy); + bus.activateSubscription(subscription); bus.start(); DocumentMessage testMessage = new DocumentMessage(1, "test"); testMessage.getHeader().setReplyChannelName("replyChannel");